From b7438df9f8440601e8688bfd8f62d75eaa3a402a Mon Sep 17 00:00:00 2001 From: wutongyuonce <147830929+wutongyuonce@users.noreply.github.com> Date: Wed, 23 Sep 2026 16:55:17 +0800 Subject: [PATCH 1/2] perf(desktop): update onboarding from targeted Session reads Generated-by: OpenAI Codex --- apps/desktop/e2e/fixtures.ts | 23 ++ apps/desktop/package.json | 1 + .../perf/onboarding-session-update.spec.ts | 62 ++++++ apps/desktop/renderer-architecture.json | 10 +- .../measure-onboarding-session-update.mjs | 204 +++++++++++++++++ .../onboarding-incremental-preload.test.ts | 207 ++++++++++++++++++ .../onboarding-service-incremental.test.ts | 166 ++++++++++++++ .../__tests__/use-onboarding-snapshot.test.ts | 113 +++++++++- apps/desktop/src/main/onboarding-ipc-main.ts | 6 + apps/desktop/src/main/onboarding-service.ts | 185 ++++++++++------ apps/desktop/src/main/runtime-host-boot.ts | 4 + apps/desktop/src/preload/bridge-contract.d.ts | 13 ++ apps/desktop/src/preload/preload.ts | 72 +++++- .../desktop/onboarding-snapshot-bridge.ts | 64 ++++++ .../src/renderer/use-onboarding-snapshot.ts | 180 +++++++++------ .../desktop-onboarding-incremental.md | 147 +++++++++++++ .../core/src/__tests__/onboarding.test.ts | 16 +- packages/core/src/onboarding.ts | 15 +- 18 files changed, 1329 insertions(+), 159 deletions(-) create mode 100644 apps/desktop/perf/onboarding-session-update.spec.ts create mode 100644 apps/desktop/scripts/measure-onboarding-session-update.mjs create mode 100644 apps/desktop/src/main/__tests__/onboarding-incremental-preload.test.ts create mode 100644 apps/desktop/src/main/__tests__/onboarding-service-incremental.test.ts create mode 100644 apps/desktop/src/renderer/platform/desktop/onboarding-snapshot-bridge.ts create mode 100644 docs/architecture/desktop-onboarding-incremental.md diff --git a/apps/desktop/e2e/fixtures.ts b/apps/desktop/e2e/fixtures.ts index f384bafab80..1e911e64bb7 100644 --- a/apps/desktop/e2e/fixtures.ts +++ b/apps/desktop/e2e/fixtures.ts @@ -213,6 +213,24 @@ async function seedRailRenderSessions(userDataDir: string): Promise { } } +async function seedOnboardingPerfSessions(userDataDir: string, count: number): Promise { + const store = createSessionStore(path.join(userDataDir, 'workspaces', 'default')); + try { + for (let index = 0; index < count; index += 1) { + await store.create({ + cwd: path.join(userDataDir, 'project'), + llmConnectionSlug: 'e2e', + model: 'claude-sonnet-4-5-20250929', + permissionMode: 'ask', + name: `Onboarding perf row ${index}`, + labels: [], + }); + } + } finally { + await store.close?.(); + } +} + async function seedParentRemovalSessions(userDataDir: string): Promise { const workspaceRoot = path.join(userDataDir, 'workspaces', 'default'); const store = createSessionStore(workspaceRoot); @@ -403,6 +421,7 @@ export async function withE2eWindow( gitReviewExtraFiles, parentRemovalSessions, railRenderSessions, + onboardingPerfSessions, newTaskProject, tracePath, testInfo, @@ -421,6 +440,7 @@ export async function withE2eWindow( gitReviewExtraFiles?: number; parentRemovalSessions?: boolean; railRenderSessions?: boolean; + onboardingPerfSessions?: number; newTaskProject?: boolean; tracePath?: string; /** Attaches captured main/renderer console output when the test fails. */ @@ -441,6 +461,9 @@ export async function withE2eWindow( if (seed) await seedE2eConnection(userDataDir); if (parentRemovalSessions) await seedParentRemovalSessions(userDataDir); if (railRenderSessions) await seedRailRenderSessions(userDataDir); + if (onboardingPerfSessions !== undefined) { + await seedOnboardingPerfSessions(userDataDir, onboardingPerfSessions); + } if (invocableSkills) await seedE2eInvocableSkills(userDataDir); if (gitReviewExtraFiles !== undefined) { await seedE2eGitReviewProject(userDataDir, gitReviewExtraFiles); diff --git a/apps/desktop/package.json b/apps/desktop/package.json index e00b65f73d0..1ef831f3e0e 100644 --- a/apps/desktop/package.json +++ b/apps/desktop/package.json @@ -46,6 +46,7 @@ "test:dist": "node --test --test-force-exit \"dist/main/**/*.test.js\" scripts/dev-app-runtime.test.mjs scripts/vite-workspace-packages.test.mjs scripts/vite-client-plugin-slots.test.mjs scripts/build-executor-plugins.test.mjs", "e2e": "npm run build:with-deps && playwright test --config e2e/playwright.config.ts", "measure": "npm run build:with-deps && playwright test --config perf/playwright.config.ts", + "measure:onboarding": "node scripts/measure-onboarding-session-update.mjs", "build:with-deps": "npm run build:workspace-deps && npm run build", "smoke:real-window": "npm run build:with-deps && node ../../scripts/desktop-real-window-smoke.mjs", "smoke:programmatic-window": "npm run build:with-deps && node ../../scripts/desktop-real-window-smoke.mjs --programmatic-only", diff --git a/apps/desktop/perf/onboarding-session-update.spec.ts b/apps/desktop/perf/onboarding-session-update.spec.ts new file mode 100644 index 00000000000..b0d5862444b --- /dev/null +++ b/apps/desktop/perf/onboarding-session-update.spec.ts @@ -0,0 +1,62 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { test, withE2eWindow } from '../e2e/fixtures'; + +// Run only when measuring a PR. The complete read is the pre-change path; +// the targeted read is the new path. This records actual Electron bridge and +// Host time, but has no injected network latency. The synthetic companion +// script covers 30 ms transport latency and Host request/byte counts. +for (const count of [100, 1_000, 5_000]) { + test(`onboarding bridge read cost with ${count} Sessions`, async () => { + await withE2eWindow({ + seed: true, + onboardingPerfSessions: count, + readinessSelector: '.maka-composer', + readinessTimeoutMs: 300_000, + }, async (page) => { + const samples = await page.evaluate(async () => { + const initial = await window.maka.onboarding.getSnapshot(); + const sessionId = initial.sessions[Math.floor(initial.sessions.length / 2)]?.id; + if (!sessionId) throw new Error('Onboarding performance fixture has no Session'); + const sample = async (read: () => Promise) => { + const durations: number[] = []; + let bytes = 0; + for (let index = 0; index < 5; index += 1) { + const start = performance.now(); + const result = await read(); + durations.push(performance.now() - start); + bytes += new TextEncoder().encode(JSON.stringify(result)).byteLength; + } + durations.sort((a, b) => a - b); + return { medianMs: durations[2], p95Ms: durations[4], bytesPerRead: bytes / 5 }; + }; + return { + sessionCount: initial.sessions.length, + full: await sample(() => window.maka.onboarding.getSnapshot()), + targeted: await sample(() => window.maka.onboarding.getSessionUpdate(sessionId)), + }; + }); + if (samples.sessionCount < count) { + throw new Error(`Only ${samples.sessionCount} of ${count} seeded Sessions appeared`); + } + console.log(JSON.stringify({ kind: 'electron-onboarding', ...samples })); + }); + }); +} diff --git a/apps/desktop/renderer-architecture.json b/apps/desktop/renderer-architecture.json index a23679da6d0..2cad212ab46 100644 --- a/apps/desktop/renderer-architecture.json +++ b/apps/desktop/renderer-architecture.json @@ -3796,11 +3796,7 @@ } }, "src/renderer/use-onboarding-snapshot.ts": { - "bridgePaths": { - "window.maka.connections.subscribeEvents": 1, - "window.maka.onboarding.getSnapshot": 1, - "window.maka.sessions.subscribeChanges": 1 - }, + "bridgePaths": {}, "environmentCapabilities": {}, "hookCalls": { "useEffect": 1, @@ -3812,9 +3808,7 @@ "unresolvedDependencies": 0, "actionFactories": [], "dependencyPaths": { - "./locales/onboarding-copy.js": 1, - "@maka/core/onboarding-milestone": 1, - "@maka/core/redaction": 1, + "./platform/desktop/onboarding-snapshot-bridge.js": 2, "@maka/ui": 1, "react": 1 } diff --git a/apps/desktop/scripts/measure-onboarding-session-update.mjs b/apps/desktop/scripts/measure-onboarding-session-update.mjs new file mode 100644 index 00000000000..0979a6ff87e --- /dev/null +++ b/apps/desktop/scripts/measure-onboarding-session-update.mjs @@ -0,0 +1,204 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +// Run after `npm run build`. The "before" path invokes the same complete +// snapshot used by the old Renderer invalidation; the "after" path invokes +// one targeted change. This is a synthetic Host transport, not an Electron UI +// trace. It deliberately does not include the initial bootstrap cost. + +import { performance } from 'node:perf_hooks'; +import { createOnboardingService } from '../dist/main/onboarding-service.js'; +import { + applyOnboardingSessionUpdate, + createOnboardingSnapshotPoller, +} from '../dist/renderer/use-onboarding-snapshot.js'; +import { readRuntimeHostSessions } from '../../../packages/runtime-host/dist/client/catalog-reader.js'; + +const counts = [100, 1_000, 5_000]; +const latencies = [0, 30]; +const delay = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); + +function percentile(samples, fraction) { + const sorted = [...samples].sort((a, b) => a - b); + return Number(sorted[Math.ceil(fraction * sorted.length) - 1].toFixed(2)); +} + +async function measure(count, roundTripMs) { + const rows = Array.from({ length: count }, (_, index) => ({ + id: `session-${index}`, backend: 'plugin-executor', llmConnectionSlug: '', + model: '', connectionLocked: false, + })); + const rowsById = new Map(rows.map((row) => [row.id, row])); + let requests = []; + const connection = { + async request(_operation, input) { + if (roundTripMs) await delay(roundTripMs); + const result = input.kind === 'get' + ? { kind: 'session', session: rowsById.get(input.sessionId) ?? null } + : (() => { + const offset = input.kind === 'list_start' ? 0 : Number(input.cursor); + const next = Math.min(offset + 32, rows.length); + return { + kind: 'page', revision: 1, sessions: rows.slice(offset, next), + nextCursor: next < rows.length ? String(next) : null, + }; + })(); + requests.push({ + kind: input.kind, + bytes: Buffer.byteLength(JSON.stringify(result)), + rows: result.kind === 'page' ? result.sessions.length : Number(result.session !== null), + }); + return result; + }, + }; + const service = createOnboardingService({ + listConnections: async () => [], + getDefaultSlug: async () => null, + listSessions: () => readRuntimeHostSessions(connection), + getSession: async (id) => (await connection.request('session.catalog.query', { + kind: 'get', sessionId: id, + })).session, + getMilestones: async () => [{ id: 'initial_onboarding', completedAt: 1 }], + upsertMilestone: async () => [{ id: 'initial_onboarding', completedAt: 1 }], + hasCredential: async () => true, + }); + await service.getSnapshot(); // Both paths start after identical coverage. + if (process.argv.includes('--burst')) { + const results = {}; + for (const name of ['before_full_snapshot', 'after_targeted_update']) { + let ipcBytes = 0; + const poller = createOnboardingSnapshotPoller({ + getSnapshot: async () => { + const result = await service.getSnapshot(); + ipcBytes += Buffer.byteLength(JSON.stringify(result)); + return result; + }, + ...(name === 'after_targeted_update' ? { + getSessionUpdate: async (id) => { + const result = await service.getSessionUpdate(id); + ipcBytes += Buffer.byteLength(JSON.stringify(result)); + return { ...result, sessionId: id }; + }, + } : {}), + }, { onSnapshot() {}, onSessionUpdate() {}, onError(error) { throw new Error(error); } }, () => 'en'); + await poller.pull(); + const durations = []; + let listRequests = 0; + let getRequests = 0; + let hostBytes = 0; + let transferredBytes = 0; + const sampleCount = roundTripMs ? 5 : 10; + const sessionId = `session-${Math.floor(count / 2)}`; + for (let sample = 0; sample < sampleCount; sample += 1) { + requests = []; + ipcBytes = 0; + const start = performance.now(); + const first = name === 'after_targeted_update' + ? poller.pullSession(sessionId) + : poller.pull(); + for (let event = 0; event < 10; event += 1) { + if (name === 'after_targeted_update') void poller.pullSession(sessionId); + else void poller.pull(); + } + await first; + durations.push(performance.now() - start); + listRequests += requests.filter(({ kind }) => kind.startsWith('list_')).length; + getRequests += requests.filter(({ kind }) => kind === 'get').length; + hostBytes += requests.reduce((sum, request) => sum + request.bytes, 0); + transferredBytes += ipcBytes; + } + poller.dispose(); + results[name] = { + medianMs: percentile(durations, 0.5), p95Ms: percentile(durations, 0.95), + listRequestsPerBurst: listRequests / sampleCount, + getRequestsPerBurst: getRequests / sampleCount, + hostBytesPerBurst: Math.round(hostBytes / sampleCount), + ipcBytesPerBurst: Math.round(transferredBytes / sampleCount), + }; + } + return { count, roundTripMs, burstNotifications: 11, ...results }; + } + const sampleCount = roundTripMs ? 7 : 30; + const results = {}; + for (const [name, run] of [ + ['before_full_snapshot', () => service.getSnapshot()], + ['after_targeted_update', () => service.getSessionUpdate(`session-${Math.floor(count / 2)}`)], + ]) { + const durations = []; + let listRequests = 0; + let getRequests = 0; + let hostBytes = 0; + let ipcBytes = 0; + let materializedRows = 0; + for (let index = 0; index < sampleCount; index++) { + requests = []; + const start = performance.now(); + const response = await run(); + durations.push(performance.now() - start); + listRequests += requests.filter(({ kind }) => kind.startsWith('list_')).length; + getRequests += requests.filter(({ kind }) => kind === 'get').length; + hostBytes += requests.reduce((sum, request) => sum + request.bytes, 0); + materializedRows += requests.reduce((sum, request) => sum + request.rows, 0); + ipcBytes += Buffer.byteLength(JSON.stringify(response)); + } + results[name] = { + medianMs: percentile(durations, 0.5), p95Ms: percentile(durations, 0.95), + listRequestsPerChange: listRequests / sampleCount, + getRequestsPerChange: getRequests / sampleCount, + hostBytesPerChange: Math.round(hostBytes / sampleCount), + ipcBytesPerChange: Math.round(ipcBytes / sampleCount), + materializedRowsPerChange: materializedRows / sampleCount, + }; + } + return { count, roundTripMs, sampleCount, ...results }; +} + +if (process.argv.includes('--renderer-copy')) { + for (const count of counts) { + let snapshot = { + state: { kind: 'needs_connection' }, milestones: [], sessions: [], connections: [], + defaultSlug: null, chatModelChoices: [], + sessionSendOutcomes: Object.fromEntries( + Array.from({ length: count }, (_, index) => [`session-${index}`, { kind: 'ready' }]), + ), + }; + const durations = []; + for (let index = 0; index < 230; index += 1) { + const update = { + kind: 'delta', sessionId: `session-${Math.floor(count / 2)}`, + outcome: index % 2 === 0 + ? { kind: 'blocked', reason: 'fake_backend', connectionLocked: false } + : { kind: 'ready' }, + }; + const start = performance.now(); + snapshot = applyOnboardingSessionUpdate(snapshot, update); + if (index >= 30) durations.push(performance.now() - start); + } + console.log(JSON.stringify({ + kind: 'renderer-record-copy', count, + medianMs: percentile(durations, 0.5), p95Ms: percentile(durations, 0.95), + })); + } +} else { + for (const count of counts) { + for (const roundTripMs of latencies) { + console.log(JSON.stringify(await measure(count, roundTripMs))); + } + } +} diff --git a/apps/desktop/src/main/__tests__/onboarding-incremental-preload.test.ts b/apps/desktop/src/main/__tests__/onboarding-incremental-preload.test.ts new file mode 100644 index 00000000000..1125e293f9f --- /dev/null +++ b/apps/desktop/src/main/__tests__/onboarding-incremental-preload.test.ts @@ -0,0 +1,207 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { createRequire } from 'node:module'; +import test from 'node:test'; +import { fileURLToPath } from 'node:url'; +import { runInNewContext } from 'node:vm'; +import { build } from 'esbuild'; +import type { MakaBridge } from '../../preload/bridge-contract.js'; +import type { OnboardingSnapshot } from '../onboarding-service.js'; +import { desktopSessionKey } from '../../shared/runtime-host-identity.js'; +import { createOnboardingSnapshotPoller } from '../../renderer/use-onboarding-snapshot.js'; + +const owners = [ + { hostId: 'local-host', targetEpoch: 'local-epoch', profileId: 'local', profileName: 'Local', profileKind: 'local', profileAccess: 'owner', readiness: 'ready' }, + { hostId: 'remote-host', targetEpoch: 'remote-epoch', profileId: 'remote', profileName: 'Remote', profileKind: 'remote', profileAccess: 'owner', readiness: 'ready' }, +] as const; +const guest = { hostId: 'guest-host', targetEpoch: 'guest-epoch', profileId: 'guest', profileName: 'Guest', profileKind: 'remote', profileAccess: 'session_guest', readiness: 'ready' }; + +function hostSnapshot(hostId: string): OnboardingSnapshot { + const id = hostId === 'local-host' ? 'local-task' : 'remote-task'; + return { + state: { kind: 'needs_connection' }, milestones: [], + sessions: [{ + id, revision: 1, activityAt: 1, name: id, + isFlagged: false, isArchived: false, labels: [], hasUnread: false, + status: 'active', backend: 'plugin-executor', llmConnectionSlug: '', + model: '', connectionLocked: false, permissionMode: 'ask', + } as OnboardingSnapshot['sessions'][number]], + connections: [], defaultSlug: null, chatModelChoices: [], + sessionSendOutcomes: { [id]: { kind: 'ready' } }, + }; +} + +async function harness() { + const listeners = new Map void>>(); + let failRemote = false; + let heldRemote: Promise | null = null; + const calls: Array<{ channel: string; hostId?: string }> = []; + const ipcRenderer = { + on(channel: string, listener: (...args: unknown[]) => void) { + const set = listeners.get(channel) ?? new Set(); + set.add(listener); + listeners.set(channel, set); + }, + off(channel: string, listener: (...args: unknown[]) => void) { + listeners.get(channel)?.delete(listener); + }, + send() {}, + async invoke(channel: string, scope?: { hostId?: string }, sessionId?: string) { + calls.push({ channel, hostId: scope?.hostId }); + switch (channel) { + case 'runtime-host:activeIdentity': return owners[0]; + case 'runtime-host:identities': return [...owners, guest]; + case 'runtime-host:awaitReady': return { ready: true }; + case 'onboarding:getSnapshot': + if (scope?.hostId === 'remote-host' && failRemote) throw new Error('remote offline'); + if (scope?.hostId === 'remote-host' && heldRemote) { + const held = heldRemote; + heldRemote = null; + return held; + } + return hostSnapshot(scope!.hostId!); + case 'onboarding:getSessionUpdate': + assert.equal(sessionId, scope?.hostId === 'remote-host' ? 'remote-task' : 'local-task'); + return { + kind: 'delta', outcome: { kind: 'blocked', reason: 'fake_backend', connectionLocked: false }, + state: { kind: 'needs_connection' }, milestones: [], + }; + case 'session-local:catalog': return owners.map((owner) => ({ + scope: owner, sessions: hostSnapshot(owner.hostId).sessions, authoritative: true, + })); + case 'session-collaboration:mount:list': return []; + default: throw new Error('Unexpected channel: ' + channel); + } + }, + }; + let bridge: MakaBridge | undefined; + const bundle = await build({ + entryPoints: [fileURLToPath(new URL('../../../src/preload/preload.ts', import.meta.url))], + bundle: true, write: false, platform: 'node', format: 'cjs', external: ['electron'], + }); + const require = createRequire(import.meta.url); + runInNewContext(bundle.outputFiles[0]!.text, { + require: (id: string) => id === 'electron' ? { + ipcRenderer, + contextBridge: { exposeInMainWorld: (name: string, value: MakaBridge) => { + if (name === 'maka') bridge = value; + } }, + } : require(id), + process: { env: {} }, Buffer, console, setTimeout, clearTimeout, TextEncoder, TextDecoder, + crypto: globalThis.crypto, + }); + assert.ok(bridge); + return { + bridge, + calls, + failRemote: () => { failRemote = true; }, + holdRemote: (read: Promise) => { heldRemote = read; }, + emitRemoteSessionChange: () => { + for (const listener of listeners.get('sessions:changed') ?? []) { + listener({}, owners[1], { reason: 'updated', sessionId: 'remote-task', ts: 1 }); + } + }, + replaceRemote: () => { + for (const listener of listeners.get('runtime-host-profiles:changed') ?? []) { + listener({}, { + profileId: 'remote', profileName: 'Remote', profileKind: 'remote', + profileAccess: 'owner', hostId: 'remote-host', epoch: 'replacement-epoch', + readiness: 'ready', isDefault: false, + }); + } + }, + }; +} + +test('a remote Owner delta is scoped and a later failed full read keeps its last projection', async () => { + const fixture = await harness(); + const localKey = desktopSessionKey({ hostId: 'local-host', sessionId: 'local-task' }); + const remoteKey = desktopSessionKey({ hostId: 'remote-host', sessionId: 'remote-task' }); + const first = await fixture.bridge.onboarding.getSnapshot(); + assert.deepEqual(first.sessionSendOutcomes[localKey], { kind: 'ready' }); + assert.deepEqual(first.sessionSendOutcomes[remoteKey], { kind: 'ready' }); + fixture.calls.length = 0; + const update = await fixture.bridge.onboarding.getSessionUpdate(remoteKey); + assert.equal(update?.kind, 'delta'); + if (update?.kind !== 'delta') return; + assert.equal(update.defaultHost, undefined); + assert.equal(update.sessionId, remoteKey); + assert.deepEqual(fixture.calls.filter(({ channel }) => channel === 'onboarding:getSessionUpdate'), [ + { channel: 'onboarding:getSessionUpdate', hostId: 'remote-host' }, + ]); + assert.equal(fixture.calls.some(({ channel }) => channel === 'onboarding:getSnapshot'), false); + fixture.failRemote(); + const afterFailure = await fixture.bridge.onboarding.getSnapshot(); + assert.deepEqual(afterFailure.sessionSendOutcomes[remoteKey], update.outcome); + assert.deepEqual(afterFailure.sessionSendOutcomes[localKey], { kind: 'ready' }); + fixture.replaceRemote(); + assert.equal((await fixture.bridge.onboarding.getSessionUpdate(remoteKey))?.kind, 'resync'); + const guestKey = desktopSessionKey({ hostId: 'guest-host', sessionId: 'guest-task' }); + assert.equal(await fixture.bridge.onboarding.getSessionUpdate(guestKey), null); +}); + +test('a Renderer Session event crosses preload to only its owning Host update', async () => { + const fixture = await harness(); + let receiveUpdate!: () => void; + const observed = new Promise((resolve) => { receiveUpdate = resolve; }); + const poller = createOnboardingSnapshotPoller({ + getSnapshot: () => fixture.bridge.onboarding.getSnapshot(), + getSessionUpdate: (id) => fixture.bridge.onboarding.getSessionUpdate(id), + }, { + onSnapshot: () => {}, + onSessionUpdate: (update) => { + assert.equal(update.sessionId, desktopSessionKey({ hostId: 'remote-host', sessionId: 'remote-task' })); + receiveUpdate(); + }, + onError: (message) => assert.fail(message), + }, () => 'en'); + await poller.pull(); + const unsubscribe = fixture.bridge.sessions.subscribeChanges((event) => { + if (event.sessionId) void poller.pullSession(event.sessionId); + else void poller.pull(); + }); + try { + fixture.calls.length = 0; + fixture.emitRemoteSessionChange(); + await observed; + assert.deepEqual(fixture.calls.filter(({ channel }) => channel.startsWith('onboarding:')), [ + { channel: 'onboarding:getSessionUpdate', hostId: 'remote-host' }, + ]); + } finally { + unsubscribe(); + poller.dispose(); + } +}); + +test('an older complete read cannot erase a newer targeted outcome', async () => { + const fixture = await harness(); + const remoteKey = desktopSessionKey({ hostId: 'remote-host', sessionId: 'remote-task' }); + await fixture.bridge.onboarding.getSnapshot(); + let release!: (snapshot: OnboardingSnapshot) => void; + fixture.holdRemote(new Promise((resolve) => { release = resolve; })); + const oldRead = fixture.bridge.onboarding.getSnapshot(); + await new Promise((resolve) => setImmediate(resolve)); + const update = await fixture.bridge.onboarding.getSessionUpdate(remoteKey); + assert.equal(update?.kind, 'delta'); + release(hostSnapshot('remote-host')); + const latest = await oldRead; + assert.equal(latest.sessionSendOutcomes[remoteKey]?.kind, 'blocked'); +}); diff --git a/apps/desktop/src/main/__tests__/onboarding-service-incremental.test.ts b/apps/desktop/src/main/__tests__/onboarding-service-incremental.test.ts new file mode 100644 index 00000000000..3be55ab7008 --- /dev/null +++ b/apps/desktop/src/main/__tests__/onboarding-service-incremental.test.ts @@ -0,0 +1,166 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import test from 'node:test'; +import type { ProjectedLlmConnection } from '@maka/core/llm-connections'; +import { resolveConnectionModelCatalog } from '@maka/core/model-catalog'; +import type { SessionSummary } from '@maka/core/session'; +import type { OnboardingMilestone } from '@maka/core/onboarding'; +import { createOnboardingService } from '../onboarding-service.js'; + +const storedConnection = { + connectionId: 'connection-1', slug: 'primary', name: 'Primary', + providerType: 'openai-compatible' as const, defaultModel: 'gpt-5', + enabled: true, createdAt: 1, updatedAt: 1, +}; +const connection: ProjectedLlmConnection = { + ...storedConnection, + catalogEntries: resolveConnectionModelCatalog(storedConnection), +}; + +function session(id: string, backend: SessionSummary['backend'] = 'plugin-executor'): SessionSummary { + return { + id, name: id, isFlagged: false, isArchived: false, labels: [], + hasUnread: false, status: 'active', backend, + llmConnectionId: 'connection-1', llmConnectionSlug: 'primary', + model: 'gpt-5', connectionLocked: false, permissionMode: 'ask', + }; +} + +function harness(initial: SessionSummary[] = []) { + const sessions = new Map(initial.map((row) => [row.id, row])); + let milestones: OnboardingMilestone[] = []; + let listReads = 0; + let getReads = 0; + let failGet = false; + let credentialPresent = true; + let nextGet: Promise | null = null; + const service = createOnboardingService({ + listConnections: async () => [connection], + getDefaultSlug: async () => 'primary', + listSessions: async () => { listReads += 1; return [...sessions.values()]; }, + getSession: async (id) => { + getReads += 1; + if (failGet) throw new Error('Host read failed'); + if (nextGet) { + const held = nextGet; + nextGet = null; + return held; + } + return sessions.get(id) ?? null; + }, + getMilestones: async () => milestones, + upsertMilestone: async (id, status) => { + milestones = [{ id, ...(status === 'completed' ? { completedAt: 1 } : { skippedAt: 1 }) }]; + return milestones; + }, + hasCredential: async () => credentialPresent, + }); + return { + service, sessions, + listReads: () => listReads, + getReads: () => getReads, + failNextGet: () => { failGet = true; }, + setCredentialPresent: (present: boolean) => { credentialPresent = present; }, + holdNextGet: (held: Promise) => { nextGet = held; }, + }; +} + +test('first and last Session use targeted reads and preserve the settled milestone', async () => { + const fixture = harness(); + assert.equal((await fixture.service.getSnapshot()).state.kind, 'ready_empty'); + fixture.sessions.set('first', session('first')); + const created = await fixture.service.getSessionUpdate('first'); + assert.equal(created.kind, 'delta'); + if (created.kind !== 'delta') return; + assert.equal(created.state.kind, 'ready_with_history'); + assert.deepEqual(created.outcome, { kind: 'ready' }); + assert.equal(created.milestones[0]?.id, 'initial_onboarding'); + fixture.sessions.set('first', { ...session('first'), isArchived: true, status: 'aborted' }); + const archived = await fixture.service.getSessionUpdate('first'); + assert.equal(archived.kind === 'delta' && archived.state.kind, 'ready_with_history'); + fixture.sessions.delete('first'); + const deleted = await fixture.service.getSessionUpdate('first'); + assert.equal(deleted.kind, 'delta'); + if (deleted.kind !== 'delta') return; + assert.equal(deleted.state.kind, 'ready_empty'); + assert.equal(deleted.outcome, null); + assert.deepEqual(deleted.milestones, created.milestones); + assert.equal(fixture.listReads(), 1); + assert.equal(fixture.getReads(), 3); +}); + +test('global credential changes rebuild active and rail readiness from Core rules', async () => { + const fixture = harness([session('native', 'ai-sdk')]); + const ready = await fixture.service.getSnapshot(); + assert.equal(ready.state.kind, 'ready_with_history'); + assert.deepEqual(ready.sessionSendOutcomes.native, { kind: 'ready' }); + fixture.setCredentialPresent(false); + const blocked = await fixture.service.getSnapshot(); + assert.equal(blocked.state.kind, 'needs_connection_credentials'); + assert.deepEqual(blocked.sessionSendOutcomes.native, { + kind: 'blocked', reason: 'missing_api_key', connectionLocked: false, + }); + assert.equal(fixture.listReads(), 2, 'a global change is allowed a complete read'); +}); + +test('a named change reads one row and never lists the catalog again', async () => { + const fixture = harness(Array.from({ length: 1_000 }, (_, i) => session(`s-${i}`))); + await fixture.service.getSnapshot(); + fixture.sessions.set('s-500', session('s-500', 'fake')); + const update = await fixture.service.getSessionUpdate('s-500'); + assert.equal(update.kind, 'delta'); + if (update.kind !== 'delta') return; + assert.deepEqual(update.outcome, { + kind: 'blocked', reason: 'fake_backend', connectionLocked: false, + }); + assert.equal(fixture.listReads(), 1); + assert.equal(fixture.getReads(), 1); + assert.equal(JSON.stringify(update).includes('s-999'), false); +}); + +test('failed targeted reads leave the accepted baseline intact', async () => { + const fixture = harness([session('kept')]); + await fixture.service.getSnapshot(); + fixture.sessions.delete('kept'); + fixture.failNextGet(); + await assert.rejects(fixture.service.getSessionUpdate('kept'), /Host read failed/); + assert.equal(fixture.listReads(), 1); +}); + +test('an update before initial coverage requests a resync', async () => { + const fixture = harness(); + assert.deepEqual(await fixture.service.getSessionUpdate('unknown'), { kind: 'resync' }); + assert.equal(fixture.getReads(), 0); +}); + +test('a late targeted answer cannot replace a newer full observation', async () => { + const fixture = harness([session('one')]); + await fixture.service.getSnapshot(); + let release!: (row: SessionSummary | null) => void; + fixture.holdNextGet(new Promise((resolve) => { release = resolve; })); + const pending = fixture.service.getSessionUpdate('one'); + await Promise.resolve(); + fixture.setCredentialPresent(false); + await fixture.service.getSnapshot(); + release(session('one')); + assert.deepEqual(await pending, { kind: 'resync' }); + assert.equal(fixture.listReads(), 2); +}); 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 16cbf1a0d0e..7f00a85f36b 100644 --- a/apps/desktop/src/main/__tests__/use-onboarding-snapshot.test.ts +++ b/apps/desktop/src/main/__tests__/use-onboarding-snapshot.test.ts @@ -21,6 +21,7 @@ import { strict as assert } from 'node:assert'; import { describe, it } from 'node:test'; import type { OnboardingState } from '@maka/core/onboarding'; import { + applyOnboardingSessionUpdate, createOnboardingSnapshotPoller, getOnboardingActivationCandidate, onboardingSnapshotProjectionEqual, @@ -127,7 +128,117 @@ describe('onboardingSnapshotProjectionEqual', () => { }); }); +it('one outcome delta leaves unrelated outcomes and default state in place', () => { + const initial = { + ...READY_SNAPSHOT, + sessionSendOutcomes: { first: { kind: 'ready' } as const, second: { kind: 'ready' } as const }, + }; + const changed = applyOnboardingSessionUpdate(initial, { + kind: 'delta', sessionId: 'second', + outcome: { kind: 'blocked', reason: 'fake_backend', connectionLocked: false }, + }); + assert.deepEqual(changed.sessionSendOutcomes.first, { kind: 'ready' }); + assert.equal(changed.state, initial.state); + assert.equal(initial.sessionSendOutcomes.second.kind, 'ready'); + assert.equal(applyOnboardingSessionUpdate(changed, { + kind: 'delta', sessionId: 'second', outcome: changed.sessionSendOutcomes.second, + }), changed, 'an unchanged high-frequency outcome does not copy the entire record'); + const removed = applyOnboardingSessionUpdate(changed, { + kind: 'delta', sessionId: 'second', outcome: null, + }); + assert.equal(removed.sessionSendOutcomes.second, undefined); + assert.deepEqual(removed.sessionSendOutcomes.first, { kind: 'ready' }); +}); + describe('createOnboardingSnapshotPoller', () => { + it('uses one targeted read for a named event after bootstrap', async () => { + let fullReads = 0; + const updates: string[] = []; + const emitted: unknown[] = []; + const poller = createOnboardingSnapshotPoller({ + getSnapshot: async () => { fullReads++; return READY_SNAPSHOT; }, + getSessionUpdate: async (id) => { + updates.push(id); + return { kind: 'delta', sessionId: id, outcome: { kind: 'ready' } }; + }, + }, { + onSnapshot: (snapshot) => emitted.push(snapshot), + onSessionUpdate: (update) => emitted.push(update), + onError: (error) => assert.fail(error), + }, () => 'zh-CN'); + await poller.pull(); + await poller.pullSession('one'); + assert.equal(fullReads, 1); + assert.deepEqual(updates, ['one']); + assert.equal(emitted.length, 2); + }); + + it('keeps the accepted snapshot when a targeted Host read fails', async () => { + const snapshots: OnboardingSnapshot[] = []; + const errors: string[] = []; + const poller = createOnboardingSnapshotPoller({ + getSnapshot: async () => READY_SNAPSHOT, + getSessionUpdate: async () => { throw new Error('Host disconnected'); }, + }, { + onSnapshot: (snapshot) => snapshots.push(snapshot), + onError: (message) => errors.push(message), + }, () => 'zh-CN'); + await poller.pull(); + await poller.pullSession('one'); + assert.deepEqual(snapshots, [READY_SNAPSHOT]); + assert.equal(errors.length, 1); + }); + + it('coalesces a repeat while a targeted read is in flight', async () => { + let resolveFirst!: (value: { kind: 'delta'; sessionId: string; outcome: { kind: 'ready' } }) => void; + let calls = 0; + const emitted: string[] = []; + const poller = createOnboardingSnapshotPoller({ + getSnapshot: async () => READY_SNAPSHOT, + getSessionUpdate: (id) => { + calls++; + return calls === 1 + ? new Promise((resolve) => { resolveFirst = resolve; }) + : Promise.resolve({ kind: 'delta' as const, sessionId: id, outcome: { kind: 'ready' as const } }); + }, + }, { + onSnapshot: () => {}, + onSessionUpdate: (update) => emitted.push(update.sessionId), + onError: (error) => assert.fail(error), + }, () => 'zh-CN'); + await poller.pull(); + const first = poller.pullSession('one'); + void poller.pullSession('one'); + resolveFirst({ kind: 'delta', sessionId: 'one', outcome: { kind: 'ready' } }); + await first; + assert.equal(calls, 2); + assert.deepEqual(emitted, ['one'], 'the older response stays unpublished'); + }); + + it('bounds distinct pending IDs and falls back to one complete resync', async () => { + let release!: (value: { kind: 'delta'; sessionId: string; outcome: { kind: 'ready' } }) => void; + let fullReads = 0; + let targetedReads = 0; + const poller = createOnboardingSnapshotPoller({ + getSnapshot: async () => { fullReads++; return READY_SNAPSHOT; }, + getSessionUpdate: (id) => { + targetedReads++; + return new Promise((resolve) => { release = resolve; }); + }, + }, { + onSnapshot: () => {}, + onSessionUpdate: () => assert.fail('superseded update must not publish'), + onError: (error) => assert.fail(error), + }, () => 'zh-CN'); + await poller.pull(); + const first = poller.pullSession('active'); + for (let index = 0; index < 65; index++) void poller.pullSession(`pending-${index}`); + release({ kind: 'delta', sessionId: 'active', outcome: { kind: 'ready' } }); + await first; + assert.equal(targetedReads, 1); + assert.equal(fullReads, 2); + }); + it('scrubs getSnapshot rejections before routing them to onError', async () => { const events: Array<{ type: 'snap' | 'err'; payload: unknown }> = []; const poller = createOnboardingSnapshotPoller( @@ -175,7 +286,7 @@ describe('createOnboardingSnapshotPoller', () => { resolvers[1]!(READY_SNAPSHOT); await pull1; await pull2; - assert.deepEqual(events, [NEEDS_CONNECTION_SNAPSHOT, READY_SNAPSHOT]); + assert.deepEqual(events, [READY_SNAPSHOT], 'the superseded full response stays unpublished'); }); it('collapses repeated invalidations during one pull into a single follow-up', async () => { diff --git a/apps/desktop/src/main/onboarding-ipc-main.ts b/apps/desktop/src/main/onboarding-ipc-main.ts index d617fbfee5e..96a10b9e43a 100644 --- a/apps/desktop/src/main/onboarding-ipc-main.ts +++ b/apps/desktop/src/main/onboarding-ipc-main.ts @@ -38,6 +38,12 @@ export function registerOnboardingIpc(deps: OnboardingIpcDeps): void { handleReconnectableRead(target, 'onboarding:getSnapshot', async () => deps.onboardingService.getSnapshot(), ); + handleReconnectableRead(target, 'onboarding:getSessionUpdate', async (_event, sessionId: unknown) => { + if (typeof sessionId !== 'string' || sessionId.length === 0) { + throw new Error('INVALID_SESSION_ID'); + } + return deps.onboardingService.getSessionUpdate(sessionId); + }); target.handle('onboarding:setMilestone', async (_event, id: unknown, status: unknown) => { // Service throws INVALID_MILESTONE_ID / INVALID_MILESTONE_STATUS // for bad inputs; let the error propagate so the renderer sees diff --git a/apps/desktop/src/main/onboarding-service.ts b/apps/desktop/src/main/onboarding-service.ts index 24e396ff0f7..acc29d2cc72 100644 --- a/apps/desktop/src/main/onboarding-service.ts +++ b/apps/desktop/src/main/onboarding-service.ts @@ -39,14 +39,14 @@ * * Credential adapters project ordinary read failures to `false` and * propagate connection failures before they reach this service. + * After a complete read, named Session changes use a targeted Host lookup + * against the retained, per-Host readiness inputs and history membership. * * Milestone input validation lives here too: setMilestone arguments * are checked against the closed enum + status union before reaching * the SettingsStore. */ -import { collapseSessionRevisions } from '@maka/core/session-revisions'; - import { deriveOnboardingState, hasSettledInitialOnboarding, @@ -66,10 +66,7 @@ import type { LlmConnection } from '@maka/core/llm-connections'; export interface OnboardingSnapshot { state: OnboardingState; milestones: OnboardingMilestone[]; - /** - * Session list, included so the renderer can populate the sidebar - * without a separate `sessions:list` IPC. - */ + /** Complete authenticated Owner seed for preload's sidebar catalog. */ sessions: SessionSummary[]; /** Default Host connection projection used to seed the shell. */ connections: ProjectedLlmConnection[]; @@ -82,6 +79,7 @@ export interface OnboardingServiceDeps { listConnections(): Promise; getDefaultSlug(): Promise; listSessions(): Promise; + getSession(sessionId: string): Promise; getMilestones(): Promise; upsertMilestone( id: OnboardingMilestoneId, @@ -98,12 +96,30 @@ export interface OnboardingServiceDeps { export interface OnboardingService { getSnapshot(): Promise; + getSessionUpdate(sessionId: string): Promise; setMilestone( id: unknown, status: unknown, ): Promise; } +export type OnboardingSessionUpdate = + | { kind: 'resync' } + | { + kind: 'delta'; + outcome: SessionSendProjection | null; + state: OnboardingState; + milestones: OnboardingMilestone[]; + }; + +interface OnboardingBaseline { + readonly sessionIds: Set; + readonly connections: ProjectedLlmConnection[]; + readonly defaultSlug: string | null; + readonly secrets: Readonly>; + milestones: OnboardingMilestone[]; +} + /** * Build the desktop OnboardingService. The constructor takes injected * deps (rather than reading the global stores) so the service is @@ -111,44 +127,99 @@ export interface OnboardingService { * real stores in tests. */ export function createOnboardingService(deps: OnboardingServiceDeps): OnboardingService { - return { - async getSnapshot(): Promise { - const [connections, defaultSlug, sessions, milestones] = await Promise.all([ - deps.listConnections(), - deps.getDefaultSlug(), - deps.listSessions(), - deps.getMilestones(), - ]); - - // @kenji PR110b perf gate: per-connection credential lookup must - // run in parallel, NOT serialized. Even with 4-5 connections, - // async credential-store reads can add up to noticeable startup - // latency on cold open. - const secretEntries = await Promise.all( - connections.map(async (connection) => { - const hasSecret = await deps.hasCredential(connection); - return [connection.slug, hasSecret] as const; - }), - ); - const secrets: Record = Object.fromEntries(secretEntries); + let baseline: OnboardingBaseline | null = null; + let revision = 0; + let updateTail: Promise = Promise.resolve(); + + async function loadSnapshot(credentialFailureFallback = false): Promise { + const startedAt = revision; + const [connections, defaultSlug, sessions, milestones] = await Promise.all([ + deps.listConnections(), + deps.getDefaultSlug(), + deps.listSessions(), + deps.getMilestones(), + ]); - const logicalSessions = collapseSessionRevisions(sessions); - const state = deriveOnboardingState({ + // Credential reads stay parallel and read-only. A global connection + // invalidation is allowed to rebuild every Session's projection. + const secretEntries = await Promise.all( + connections.map(async (connection) => { + try { + return [connection.slug, await deps.hasCredential(connection)] as const; + } catch (error) { + if (!credentialFailureFallback) throw error; + return [connection.slug, false] as const; + } + }), + ); + const secrets: Record = Object.fromEntries(secretEntries); + // Revision collapse always retains one member of every nonempty family. + // Onboarding only needs presence, so a second all-Session fold is unnecessary. + const hasHistory = sessions.length > 0; + const state = deriveOnboardingState({ + connections, + defaultSlug: defaultSlug ?? undefined, + hasHistory, + secrets, + }); + const settledMilestones = hasHistory && !hasSettledInitialOnboarding(milestones) + ? await deps.upsertMilestone('initial_onboarding', 'completed') + : milestones; + if (startedAt === revision) { + baseline = { + sessionIds: new Set(sessions.map(({ id }) => id)), connections, - defaultSlug: defaultSlug ?? undefined, - sessions: logicalSessions, + defaultSlug, secrets, - }); - - // Backfill: existing users who already have sessions but no - // initial_onboarding milestone (upgraded from before this PR) - // get auto-marked as completed so the hero never appears. - if (logicalSessions.length > 0 && !hasSettledInitialOnboarding(milestones)) { - const updated = await deps.upsertMilestone('initial_onboarding', 'completed'); - return buildSnapshot(state, updated, sessions, connections, defaultSlug, secrets); - } + milestones: settledMilestones, + }; + revision += 1; + } + return buildSnapshot(state, settledMilestones, sessions, connections, defaultSlug, secrets); + } + + async function readSessionUpdate(sessionId: string): Promise { + const observed = baseline; + if (!observed) return { kind: 'resync' }; + const session = await deps.getSession(sessionId); + // A concurrent global read or milestone write changed the basis used to + // project this answer. Let the caller resync instead of publishing it. + if (baseline !== observed) return { kind: 'resync' }; + const hasHistory = session !== null || + observed.sessionIds.size > (observed.sessionIds.has(sessionId) ? 1 : 0); + const milestones = hasHistory && !hasSettledInitialOnboarding(observed.milestones) + ? await deps.upsertMilestone('initial_onboarding', 'completed') + : observed.milestones; + if (baseline !== observed) return { kind: 'resync' }; + if (session === null) observed.sessionIds.delete(sessionId); + else observed.sessionIds.add(sessionId); + if (milestones !== observed.milestones) { + observed.milestones = milestones; + } + revision += 1; + return { + kind: 'delta', + outcome: session === null ? null : projectSessionSendOutcome({ + session, + connections: observed.connections, + hasSecret: (slug) => observed.secrets[slug] ?? false, + }), + state: deriveOnboardingState({ + connections: observed.connections, + defaultSlug: observed.defaultSlug ?? undefined, + hasHistory, + secrets: observed.secrets, + }), + milestones: observed.milestones, + }; + } - return buildSnapshot(state, milestones, sessions, connections, defaultSlug, secrets); + return { + getSnapshot: () => loadSnapshot(), + getSessionUpdate(sessionId: string): Promise { + const pending = updateTail.then(() => readSessionUpdate(sessionId)); + updateTail = pending.then(() => undefined, () => undefined); + return pending; }, async setMilestone(id: unknown, status: unknown): Promise { @@ -161,34 +232,12 @@ export function createOnboardingService(deps: OnboardingServiceDeps): Onboarding } // Timestamp is stamped inside the store (Date.now()); renderer // never controls it. - const milestones = await deps.upsertMilestone(id, status); - // After the write, re-derive snapshot. State could change (e.g. - // the user finished `first_chat_sent` while in `ready_empty` - // → next derive should reflect new history). Re-using the - // already-fetched milestones avoids a settings round-trip. - const [connections, defaultSlug, sessions] = await Promise.all([ - deps.listConnections(), - deps.getDefaultSlug(), - deps.listSessions(), - ]); - const secretEntries = await Promise.all( - connections.map(async (connection) => { - try { - return [connection.slug, await deps.hasCredential(connection)] as const; - } catch { - return [connection.slug, false] as const; - } - }), - ); - const secrets: Record = Object.fromEntries(secretEntries); - const logicalSessions = collapseSessionRevisions(sessions); - const state = deriveOnboardingState({ - connections, - defaultSlug: defaultSlug ?? undefined, - sessions: logicalSessions, - secrets, - }); - return buildSnapshot(state, milestones, sessions, connections, defaultSlug, secrets); + await deps.upsertMilestone(id, status); + revision += 1; + baseline = null; + // The global milestone write may alter which guide is shown; rebuild + // from the same authoritative sources as an explicit full refresh. + return loadSnapshot(true); }, }; } diff --git a/apps/desktop/src/main/runtime-host-boot.ts b/apps/desktop/src/main/runtime-host-boot.ts index 9545b823640..deec5eb0c0f 100644 --- a/apps/desktop/src/main/runtime-host-boot.ts +++ b/apps/desktop/src/main/runtime-host-boot.ts @@ -1771,6 +1771,10 @@ function registerHostClientIpc( }, listSessions: async () => (await client.listSessions()).map(toDesktopHostSessionSummary), + getSession: async (sessionId) => { + const session = await client.getSession(sessionId); + return session === null ? null : toDesktopHostSessionSummary(session); + }, getMilestones: async () => (await settingsStore.get()).onboarding.milestones, upsertMilestone: (id, status) => diff --git a/apps/desktop/src/preload/bridge-contract.d.ts b/apps/desktop/src/preload/bridge-contract.d.ts index 36f4c7297df..24397ebb283 100644 --- a/apps/desktop/src/preload/bridge-contract.d.ts +++ b/apps/desktop/src/preload/bridge-contract.d.ts @@ -326,6 +326,18 @@ export interface OnboardingSnapshot { sessionSendOutcomes: Record; } +export type DesktopOnboardingSessionUpdate = + | { kind: 'resync' } + | { + kind: 'delta'; + sessionId: string; + outcome: import('@maka/core/session-send-projection').SessionSendProjection | null; + defaultHost?: { + state: OnboardingState; + milestones: OnboardingMilestone[]; + }; + }; + export interface DesktopTaskSubmissionReadinessRequest { connectionSlug?: string; model?: string; @@ -1640,6 +1652,7 @@ export interface MakaBridge { }; onboarding: { getSnapshot(): Promise; + getSessionUpdate(sessionId: string): Promise; setMilestone( id: OnboardingMilestoneId, status: 'completed' | 'skipped', diff --git a/apps/desktop/src/preload/preload.ts b/apps/desktop/src/preload/preload.ts index b0454298d2a..a97b2cc8b01 100644 --- a/apps/desktop/src/preload/preload.ts +++ b/apps/desktop/src/preload/preload.ts @@ -48,6 +48,7 @@ import { import type { MakaBridge, OnboardingSnapshot, + DesktopOnboardingSessionUpdate, DesktopTaskSubmissionReadinessRequest, PermissionActionResult, PermissionOverlayStartResult, @@ -302,6 +303,9 @@ let lastDesktopSessionCatalog: RuntimeHostSessionCatalogCoverage = { sessions: [], completeHostIds: [], }; +/** Accepted Owner projections survive a failed read from one unrelated Host. */ +const onboardingOutcomesByScope = new Map>(); +const onboardingOutcomeVersions = new Map(); const newTaskChangeListeners = new Set<() => void>(); let previousMainProcessInterruptionRead: Promise | undefined; @@ -350,6 +354,8 @@ ipcRenderer.on( ) { runtimeHostScopes.delete(previousScopeKey); runtimeHostMetadata.delete(previousScopeKey); + onboardingOutcomesByScope.delete(previousScopeKey); + onboardingOutcomeVersions.delete(previousScopeKey); if (change.removed) { runtimeHostProfiles.delete(change.profileId); } @@ -460,6 +466,8 @@ async function runtimeHostScopeList(): Promise { if (authoritativeScopeKeys.has(scopeKey)) continue; runtimeHostScopes.delete(scopeKey); runtimeHostMetadata.delete(scopeKey); + onboardingOutcomesByScope.delete(scopeKey); + onboardingOutcomeVersions.delete(scopeKey); } return readyScopes; } @@ -918,6 +926,9 @@ async function loadDesktopOnboardingSnapshot(): Promise { scope.hostId !== defaultScope.hostId || scope.targetEpoch !== defaultScope.targetEpoch, ), ]; + const startedVersions = new Map(scopes.map((scope) => [ + runtimeHostScopeKey(scope), onboardingOutcomeVersions.get(runtimeHostScopeKey(scope)) ?? 0, + ])); const results = await Promise.allSettled( scopes.map(async (scope) => ({ scope, @@ -928,6 +939,9 @@ async function loadDesktopOnboardingSnapshot(): Promise { if (!primary || primary.status === 'rejected') { throw primary?.reason ?? new Error('Default Runtime Host onboarding is unavailable'); } + if (runtimeHostScopeKey(await activeRuntimeHostRef()) !== runtimeHostScopeKey(defaultScope)) { + throw new Error('Default Runtime Host changed during onboarding read'); + } const snapshots = results.flatMap((result) => result.status === 'fulfilled' ? [result.value] @@ -939,6 +953,16 @@ async function loadDesktopOnboardingSnapshot(): Promise { const ownerSnapshots = snapshots.filter( ({ scope }) => runtimeHostMetadataFor(scope)?.profileAccess === 'owner', ); + for (const { scope, snapshot } of ownerSnapshots) { + const scopeKey = runtimeHostScopeKey(scope); + if ((onboardingOutcomeVersions.get(scopeKey) ?? 0) !== startedVersions.get(scopeKey)) { + continue; + } + onboardingOutcomesByScope.set( + scopeKey, + new Map(Object.entries(snapshot.sessionSendOutcomes)), + ); + } const completeHostIds = [...new Set(ownerSnapshots.map(({ scope }) => scope.hostId))]; catalogSeed.commit({ sessions: reconcileRuntimeHostSessionCatalog(lastDesktopSessionCatalog.sessions, { @@ -957,12 +981,53 @@ async function loadDesktopOnboardingSnapshot(): Promise { sessions, sessionSendOutcomes: Object.assign( {}, - ...snapshots.map(({ scope, snapshot }) => - projectOnboardingSendOutcomes(scope, snapshot.sessionSendOutcomes)), + ...[...runtimeHostScopes.values()].flatMap((scope) => { + const outcomes = onboardingOutcomesByScope.get(runtimeHostScopeKey(scope)); + return outcomes && runtimeHostMetadataFor(scope)?.profileAccess === 'owner' + ? [projectOnboardingSendOutcomes(scope, Object.fromEntries(outcomes))] + : []; + }), ), }; } +async function loadDesktopOnboardingSessionUpdate( + sessionId: string, +): Promise { + let ref: Awaited>; + try { + ref = await runtimeHostSessionRef(sessionId); + } catch { + return { kind: 'resync' }; + } + const { scope, sessionId: hostSessionId } = ref; + const metadata = runtimeHostMetadataFor(scope); + if (!metadata) return { kind: 'resync' }; + if (metadata.profileAccess !== 'owner') return null; + const update = await invokeWhenReady( + 'onboarding:getSessionUpdate', scope, hostSessionId, + ) as import('../main/onboarding-service.js').OnboardingSessionUpdate; + if (update.kind === 'resync') return update; + if (runtimeHostMetadataFor(scope)?.profileId !== metadata.profileId) { + return { kind: 'resync' }; + } + const outcomes = onboardingOutcomesByScope.get(runtimeHostScopeKey(scope)); + if (!outcomes) return { kind: 'resync' }; + if (update.outcome === null) outcomes.delete(hostSessionId); + else outcomes.set(hostSessionId, update.outcome); + const scopeKey = runtimeHostScopeKey(scope); + onboardingOutcomeVersions.set(scopeKey, (onboardingOutcomeVersions.get(scopeKey) ?? 0) + 1); + const currentDefault = await activeRuntimeHostRef(); + return { + kind: 'delta', + sessionId, + outcome: update.outcome, + ...(runtimeHostScopeKey(currentDefault) === runtimeHostScopeKey(scope) + ? { defaultHost: { state: update.state, milestones: update.milestones } } + : {}), + }; +} + function projectShellRunUpdate( scope: DesktopTargetScope, update: ShellRunUpdate, @@ -3210,6 +3275,9 @@ const makaBridge = { getSnapshot(): Promise { return loadDesktopOnboardingSnapshot(); }, + getSessionUpdate(sessionId: string): Promise { + return loadDesktopOnboardingSessionUpdate(sessionId); + }, async setMilestone( id: OnboardingMilestoneId, status: 'completed' | 'skipped', diff --git a/apps/desktop/src/renderer/platform/desktop/onboarding-snapshot-bridge.ts b/apps/desktop/src/renderer/platform/desktop/onboarding-snapshot-bridge.ts new file mode 100644 index 00000000000..25974e8e1d3 --- /dev/null +++ b/apps/desktop/src/renderer/platform/desktop/onboarding-snapshot-bridge.ts @@ -0,0 +1,64 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { generalizedErrorMessageForLocale } from '@maka/core/redaction'; +import { hasSettledInitialOnboarding } from '@maka/core/onboarding-milestone'; +import type { UiLocale } from '@maka/core/ui-locale'; +import type { OnboardingSnapshot } from '../../../preload/bridge-contract.js'; +import { getOnboardingCopy } from '../../locales/onboarding-copy.js'; + +export function getOnboardingActivationCandidate( + snapshot: Pick | null, + hasWorkspaceHistory: boolean, +): { llmConnectionSlug: string; model: string } | undefined { + if ( + snapshot?.state.kind !== 'ready_empty' || + hasWorkspaceHistory || + hasSettledInitialOnboarding(snapshot.milestones) + ) { + return undefined; + } + return { + llmConnectionSlug: snapshot.state.connectionSlug, + model: snapshot.state.model, + }; +} + +export function onboardingSnapshotErrorMessage(error: unknown, locale: UiLocale): string { + const fallback = getOnboardingCopy(locale).snapshotErrorFallback; + return generalizedErrorMessageForLocale(error, fallback, locale); +} + +export const desktopOnboardingSnapshotDeps = { + getSnapshot: () => window.maka.onboarding.getSnapshot(), + getSessionUpdate: (sessionId: string) => window.maka.onboarding.getSessionUpdate(sessionId), + subscribeInvalidations(onInvalidate: (sessionId?: string) => void) { + const unsubscribeSessions = window.maka.sessions.subscribeChanges((event) => + onInvalidate(event.sessionId)); + const unsubscribeConnections = window.maka.connections.subscribeEvents(() => onInvalidate()); + const unsubscribeProfiles = window.maka.runtimeHostProfiles.subscribeChanges((event) => { + if (event.profileAccess === 'owner' || event.isDefault) onInvalidate(); + }); + return () => { + unsubscribeSessions(); + unsubscribeConnections(); + unsubscribeProfiles(); + }; + }, +}; diff --git a/apps/desktop/src/renderer/use-onboarding-snapshot.ts b/apps/desktop/src/renderer/use-onboarding-snapshot.ts index c17c1f49d12..01690f23c5a 100644 --- a/apps/desktop/src/renderer/use-onboarding-snapshot.ts +++ b/apps/desktop/src/renderer/use-onboarding-snapshot.ts @@ -21,11 +21,11 @@ * `useOnboardingSnapshot` — renderer hook over the PR110b IPC. * * @kenji + @xuan PR110c review gates: - * 1. Renderer NEVER re-derives provider readiness; only consumes - * `onboarding:getSnapshot()`. Connections, secrets, default + * 1. Renderer NEVER re-derives provider readiness; it consumes + * `onboarding:getSnapshot()` and targeted onboarding updates. Connections, secrets, default * slugs etc. are not touched. - * 2. Invalidation uses ONLY existing event channels — - * `sessions:changed` and `connections:event`. No new event bus + * 2. Invalidation uses existing event channels — + * `sessions:changed`, `connections:event`, and Host profile changes. No new event bus * for PR110c. * 3. `refresh()` is provided for action-driven re-pulls (e.g. * "the user just clicked '打开设置 · 模型' so re-pull when the @@ -33,12 +33,15 @@ */ import { useCallback, useEffect, useRef, useState } from 'react'; -import { generalizedErrorMessageForLocale } from '@maka/core/redaction'; import { type UiLocale } from '@maka/core/ui-locale'; -import { hasSettledInitialOnboarding } from '@maka/core/onboarding-milestone'; import { useUiLocale, valuesEqual } from '@maka/ui'; import type { OnboardingSnapshot } from '../preload/bridge-contract.js'; -import { getOnboardingCopy } from './locales/onboarding-copy.js'; +import type { DesktopOnboardingSessionUpdate } from '../preload/bridge-contract.js'; +import { + desktopOnboardingSnapshotDeps, + onboardingSnapshotErrorMessage, +} from './platform/desktop/onboarding-snapshot-bridge.js'; +export { getOnboardingActivationCandidate } from './platform/desktop/onboarding-snapshot-bridge.js'; /** * Hook return type — `snapshot` is `null` while the initial getSnapshot @@ -57,13 +60,14 @@ export interface UseOnboardingSnapshotResult { export interface UseOnboardingSnapshotDeps { /** Fetch the current snapshot. */ getSnapshot: () => Promise; + getSessionUpdate?: (sessionId: string) => Promise; /** * Subscribe to invalidation signals. The handler is fired * (debounced internally by the caller if needed) whenever an * upstream event suggests the snapshot may be stale. Return value * is an unsubscribe function. */ - subscribeInvalidations: (onInvalidate: () => void) => () => void; + subscribeInvalidations: (onInvalidate: (sessionId?: string) => void) => () => void; } /** @@ -71,23 +75,6 @@ export interface UseOnboardingSnapshotDeps { * guide is settled or workspace history exists, normal Composer preference * rules own new-task selection again. */ -export function getOnboardingActivationCandidate( - snapshot: Pick | null, - hasWorkspaceHistory: boolean, -): { llmConnectionSlug: string; model: string } | undefined { - if ( - snapshot?.state.kind !== 'ready_empty' || - hasWorkspaceHistory || - hasSettledInitialOnboarding(snapshot.milestones) - ) { - return undefined; - } - return { - llmConnectionSlug: snapshot.state.connectionSlug, - model: snapshot.state.model, - }; -} - /** * `sessions` is excluded: it is boot-time seed data (the session catalog is * the live authority) whose rows churn on every background message event, @@ -142,6 +129,10 @@ export function useOnboardingSnapshotImpl( ); setError(null); }, + onSessionUpdate: (update) => { + setSnapshot((prev) => prev === null ? prev : applyOnboardingSessionUpdate(prev, update)); + setError(null); + }, onError: (message) => { setError(message); }, @@ -152,8 +143,9 @@ export function useOnboardingSnapshotImpl( const poller = pollerRef.current!; poller.activate(); void poller.pull(); - const unsubscribe = deps.subscribeInvalidations(() => { - void poller.pull(); + const unsubscribe = deps.subscribeInvalidations((sessionId) => { + if (sessionId) void poller.pullSession(sessionId); + else void poller.pull(); }); return () => { unsubscribe(); @@ -173,8 +165,8 @@ export function useOnboardingSnapshotImpl( } /** - * React-less poller. Serializes getSnapshot IPCs — an invalidation while a - * pull is in flight schedules a single follow-up — and gates callbacks on + * React-less poller. Serializes complete and targeted IPCs — an invalidation while a + * read is in flight schedules a bounded follow-up — and gates callbacks on * the active flag plus a dispose-bumped ticket so pending responses cannot * write after the first-run surface unmounts. Extracted from * `useOnboardingSnapshotImpl` so the pull discipline is testable without a @@ -182,6 +174,7 @@ export function useOnboardingSnapshotImpl( */ export interface OnboardingSnapshotPollerCallbacks { onSnapshot(snapshot: OnboardingSnapshot): void; + onSessionUpdate?(update: Extract): void; onError(message: string): void; } @@ -190,19 +183,24 @@ export interface OnboardingSnapshotPoller { activate(): void; /** Fetch the latest snapshot unless disposed. */ pull(): Promise; + /** Refresh one Session's projection after an identified change. */ + pullSession(sessionId: string): Promise; /** Stop accepting callbacks. Pending IPC responses become no-ops. */ dispose(): void; } export function createOnboardingSnapshotPoller( - deps: Pick, + deps: Pick, callbacks: OnboardingSnapshotPollerCallbacks, getLocale: () => UiLocale, ): OnboardingSnapshotPoller { let inflightTicket = 0; let active = true; let inflight: Promise | null = null; - let pullAgain = false; + let fullPending = false; + let hasSnapshot = false; + const pendingSessions = new Set(); + const maxPendingSessions = 64; function emitSnapshot(snapshot: OnboardingSnapshot): void { if (!active) return; @@ -218,53 +216,113 @@ export function createOnboardingSnapshotPoller( const ticket = ++inflightTicket; try { const next = await deps.getSnapshot(); - if (!active || ticket !== inflightTicket) return; // unmounted or re-disposed + if (!active || ticket !== inflightTicket || fullPending) return; + hasSnapshot = true; emitSnapshot(next); } catch (err) { - if (!active || ticket !== inflightTicket) return; + if (!active || ticket !== inflightTicket || fullPending) return; emitError(onboardingSnapshotErrorMessage(err, getLocale())); } } + async function runSessionUpdate(sessionId: string): Promise { + const ticket = ++inflightTicket; + try { + const update = await deps.getSessionUpdate!(sessionId); + if (!active || ticket !== inflightTicket) return; + if (update?.kind === 'resync') { + fullPending = true; + } else if (update?.kind === 'delta' && !fullPending && !pendingSessions.has(sessionId)) { + callbacks.onSessionUpdate?.(update); + } + } catch (error) { + if (!active || ticket !== inflightTicket || fullPending) return; + emitError(onboardingSnapshotErrorMessage(error, getLocale())); + } + } + + function drain(): Promise { + if (!active) return Promise.resolve(); + if (inflight !== null) return inflight; + const loop = (async () => { + do { + if (fullPending) { + fullPending = false; + pendingSessions.clear(); + await runPull(); + } else { + const sessionId = pendingSessions.values().next().value; + if (sessionId === undefined) break; + pendingSessions.delete(sessionId); + await runSessionUpdate(sessionId); + } + } while (active && (fullPending || pendingSessions.size > 0)); + inflight = null; + })(); + inflight = loop; + return loop; + } + return { activate(): void { active = true; }, pull(): Promise { if (!active) return Promise.resolve(); - // Invalidations arriving while a pull is in flight collapse into one - // follow-up, so the IPC rate tracks pull latency, not event rate. - if (inflight !== null) { - pullAgain = true; - return inflight; + fullPending = true; + pendingSessions.clear(); + return drain(); + }, + pullSession(sessionId: string): Promise { + if (!active) return Promise.resolve(); + if ( + !deps.getSessionUpdate || + (!hasSnapshot && inflight === null) || + (pendingSessions.size >= maxPendingSessions && !pendingSessions.has(sessionId)) + ) { + fullPending = true; + pendingSessions.clear(); + return drain(); } - const loop = (async () => { - do { - pullAgain = false; - await runPull(); - } while (pullAgain && active); - inflight = null; - })(); - inflight = loop; - return loop; + if (!fullPending) pendingSessions.add(sessionId); + return drain(); }, dispose(): void { active = false; inflightTicket += 1; + fullPending = false; + pendingSessions.clear(); + hasSnapshot = false; }, }; } -export function onboardingSnapshotErrorMessage(error: unknown, locale: UiLocale): string { - const fallback = getOnboardingCopy(locale).snapshotErrorFallback; - return generalizedErrorMessageForLocale(error, fallback, locale); +export function applyOnboardingSessionUpdate( + snapshot: OnboardingSnapshot, + update: Extract, +): OnboardingSnapshot { + const previous = snapshot.sessionSendOutcomes[update.sessionId]; + const outcomeChanged = update.outcome === null + ? previous !== undefined + : !valuesEqual(previous, update.outcome); + const state = update.defaultHost?.state ?? snapshot.state; + const milestones = update.defaultHost?.milestones ?? snapshot.milestones; + if (!outcomeChanged && valuesEqual(state, snapshot.state) && + valuesEqual(milestones, snapshot.milestones)) return snapshot; + const sessionSendOutcomes = outcomeChanged + ? { ...snapshot.sessionSendOutcomes } + : snapshot.sessionSendOutcomes; + if (outcomeChanged) { + if (update.outcome === null) delete sessionSendOutcomes[update.sessionId]; + else sessionSendOutcomes[update.sessionId] = update.outcome; + } + return { ...snapshot, state, milestones, sessionSendOutcomes }; } /** - * Default renderer binding: subscribes to `sessions:changed` and - * `connections:event` so any session lifecycle (create / delete / archive / - * rebound / message-appended) or connection change (verified / disabled / - * removed) invalidates the snapshot. + * Default renderer binding lives in the Desktop platform adapter. Named + * Session events use targeted reads; connection and Owner profile changes + * request complete snapshots. * * Settings changes are NOT subscribed: there is no existing * settings-wide event channel and PR110c is not inventing one. If a @@ -278,17 +336,5 @@ export function onboardingSnapshotErrorMessage(error: unknown, locale: UiLocale) export function useOnboardingSnapshot(): UseOnboardingSnapshotResult { // Bind to the live IPC bridge. `deps` is memoized as a module-level // object so the effect deps stay stable across re-renders. - return useOnboardingSnapshotImpl(LIVE_DEPS); + return useOnboardingSnapshotImpl(desktopOnboardingSnapshotDeps); } - -const LIVE_DEPS: UseOnboardingSnapshotDeps = { - getSnapshot: () => window.maka.onboarding.getSnapshot(), - subscribeInvalidations(onInvalidate) { - const unsubscribeSessions = window.maka.sessions.subscribeChanges(() => onInvalidate()); - const unsubscribeConnections = window.maka.connections.subscribeEvents(() => onInvalidate()); - return () => { - unsubscribeSessions(); - unsubscribeConnections(); - }; - }, -}; diff --git a/docs/architecture/desktop-onboarding-incremental.md b/docs/architecture/desktop-onboarding-incremental.md new file mode 100644 index 00000000000..78ab2733fd6 --- /dev/null +++ b/docs/architecture/desktop-onboarding-incremental.md @@ -0,0 +1,147 @@ + + +# Desktop onboarding Session changes + +Issue: [#5619](https://github.com/apache/maka/issues/5619) + +Source baseline: `bc0786ee61219025ce89d9d7d4271bf7ce81edaa`. The PR is +rebased on `c7d205a42`. Measurements below used Node 22.22.2 and Electron +43.4.1 on macOS. The synthetic run preceded the final rebase; the Electron +run was repeated after it. + +## Contract + +An initial or global onboarding read remains a complete, authenticated Owner +snapshot. A subsequent identified Session event updates only that Session's +send projection. First and last history transitions use the same Core onboarding +rule as full reads. A failed targeted read leaves the last accepted projection +in place. Unknown authority, a changed Host epoch, or missing initial coverage +requests a complete resync. + +| Rule or state | Authority module | Interface / seam | Callers | Evidence | +| --- | --- | --- | --- | --- | +| Session existence and fields | Owning Runtime Host catalog | `session.catalog.query(get)` | Main onboarding service | Main targeted-read test; Electron bridge measurement | +| Readiness and history result | `@maka/core` onboarding and send projection | `deriveOnboardingState(hasHistory)` and `projectSessionSendOutcome` | Main onboarding service | Core and Main owner tests | +| Per-Host readiness inputs and milestone backfill | Desktop Main onboarding service | `getSnapshot`, `getSessionUpdate`, `setMilestone` | Preload | Main first/last, credential, and late-result tests | +| Owner/Guest scope, epoch, and multi-Host outcome merge | Desktop preload | `window.maka.onboarding` bridge | Renderer adapter | Preload multi-Owner, Guest, failure, and epoch seam tests | +| Sidebar rows and catalog coverage | Existing Desktop Session catalog refresher | `beginSeed`, `refresh`, `admit`, `evict` | Preload and Renderer catalog | Existing catalog seed and coverage tests | +| Invalidation order and 64-ID bound | Renderer onboarding poller | `pull`, `pullSession`, `dispose` | Onboarding React hook | Poller coalescing, overflow, and disposal tests | +| Desktop event subscriptions | Renderer Desktop platform adapter | `desktopOnboardingSnapshotDeps` | Onboarding React hook | Renderer-to-preload event seam test | + +## Reads and updates + +The initial preload read asks each ready Owner Host for its full onboarding +snapshot. It retains the existing authenticated catalog seed and independent +catalog refresh. Main records the observed Session IDs and connection readiness +inputs for each Host. A named change asks only the owning Host for that Session. +Main changes membership of one ID, calls the existing Core projection, and +returns one outcome plus the Host's onboarding state and milestones. Preload +accepts the answer only while that Owner profile and epoch remain current, +then marks whether the Host is the current default. It keeps the last accepted +outcome map for a temporarily failing Owner and fences a late full read against +a newer targeted update. Renderer applies the one +outcome and, for the default Host, its state and milestones. Guest changes do +not change Owner onboarding. + +Connection, profile, unknown-Session, and manual invalidations request a full +snapshot. Main invalidates a targeted answer when its full-read inputs change +during the read. Renderer serializes reads, coalesces repeated invalidations, +and bounds the pending set at 64 distinct Session IDs before falling back to +one complete resync. It never publishes a response after disposal. A failed read preserves its +previous snapshot and exposes the existing generalized error. + +## Performance and limits + +The steady named onboarding path performs one Host catalog `get` and transfers +one send outcome. The ordinary Session catalog has its own subscriber and may +also read that row for the same event; the one-`get` figures below measure the +onboarding path, not total Desktop requests. The onboarding path performs no +Host catalog list-page request and never serializes the all-Session outcome map +across IPC. Bootstrap and genuinely global changes may still read all pages. +The Renderer currently stores outcomes in a plain record; +replacing a changed property copies that record. Unchanged high-frequency +outcomes return the same record. The sidebar's existing stale-badge selector +still scales with visible Session count when the record changes. Making that +local render work constant would require a separate state ownership change. +`node apps/desktop/scripts/measure-onboarding-session-update.mjs --renderer-copy` +measured 200 changed-outcome copies after warm-up. With 5,000 outcomes, the +record copy took 0.50 ms median and 0.62 ms p95 on this machine. The 100 and +1,000 row medians were below the clock's useful resolution. This excludes +React rendering and the existing stale-badge selector. + +## Verification + +Owner-service tests cover first/last history, milestone backfill, targeted +projection parity, failed reads, and invalidated coverage. Preload seam tests +cover Owner/Guest routing and epoch replacement. The Renderer poller tests +cover coalescing, late responses, and targeted versus global invalidation. +The performance fixture records Host list/get counts, IPC bytes, and latency +for 100, 1,000, and 5,000 Sessions under identical before/after inputs. + +### Synthetic single-change baseline + +Run `npm run build && node apps/desktop/scripts/measure-onboarding-session-update.mjs` +from the repository root. The "before" path invokes the retained complete +snapshot operation used by the previous Renderer invalidation. The "after" +path invokes the new targeted operation after the same bootstrap. On Node +22.22.2 with a fake 32-row Host catalog, seven samples per case at artificial +30 ms transport RTT produced: + +| Sessions | Full snapshot median / p95 | Targeted median / p95 | Full list pages | Targeted gets | Full / targeted IPC bytes | +| ---: | ---: | ---: | ---: | ---: | ---: | +| 100 | 127.72 / 129.35 ms | 32.21 / 32.25 ms | 4 | 1 | 13,871 / 138 | +| 1,000 | 1,019.58 / 1,031.75 ms | 32.12 / 32.56 ms | 32 | 1 | 138,971 / 138 | +| 5,000 | 4,997.34 / 5,011.30 ms | 32.11 / 32.30 ms | 157 | 1 | 702,971 / 138 | + +The fake Host bytes were 10,926 / 135, 109,806 / 136, and 553,431 / 137 +for full / targeted responses at those sizes. `IPC bytes` are JSON-encoded +payload estimates; Electron structured-clone bytes were not measured. +Zero-RTT local medians in the same run were 0.03 / 0, +0.60 / 0, and 1.43 / 0 ms. These sub-millisecond targeted values are +below useful stopwatch resolution and are reported as request/byte evidence, +not a precise CPU speedup. + +### Electron bridge measurements + +`apps/desktop/perf/onboarding-session-update.spec.ts` seeds a throwaway real +workspace, launches Electron, and measures five complete and five targeted +bridge reads in the same window. Run it from `apps/desktop` with +`npx playwright test perf/onboarding-session-update.spec.ts --config perf/playwright.config.ts --trace on`. +On the same machine, the traced run produced: + +| Sessions | Complete median / p95 | Targeted median / p95 | Complete / targeted JSON bytes | +| ---: | ---: | ---: | ---: | +| 100 | 7.6 / 9.2 ms | 0.5 / 1.0 ms | 97,992 / 419 | +| 1,000 | 46.5 / 51.2 ms | 0.5 / 2.3 ms | 968,292 / 419 | +| 5,000 | 197.1 / 214.2 ms | 0.5 / 1.2 ms | 4,840,292 / 419 | + +These are bridge read timings, not user-interaction or end-to-end paint +timings. The large-session first paint remains outside this change. They have +no injected network latency; the synthetic run above measures that case. + +With `--burst`, the synthetic script gives the existing poller eleven rapid +notifications for the same Session. The poller coalesces them into two reads. +At 30 ms RTT, five samples per size yielded: + +| Sessions | Complete burst median / p95 | Targeted burst median / p95 | List pages / targeted gets | Complete / targeted JSON bytes | +| ---: | ---: | ---: | ---: | ---: | +| 100 | 255.19 / 256.41 ms | 63.94 / 64.33 ms | 8 / 2 | 27,742 / 276 | +| 1,000 | 2,043.08 / 2,055.72 ms | 63.62 / 64.30 ms | 64 / 2 | 277,942 / 276 | +| 5,000 | 10,003.45 / 10,009.98 ms | 64.27 / 64.34 ms | 314 / 2 | 1,405,942 / 276 | diff --git a/packages/core/src/__tests__/onboarding.test.ts b/packages/core/src/__tests__/onboarding.test.ts index a3685d297fe..1bdd4217dd6 100644 --- a/packages/core/src/__tests__/onboarding.test.ts +++ b/packages/core/src/__tests__/onboarding.test.ts @@ -55,12 +55,26 @@ function derive(input: Partial = {}): OnboardingStat return deriveOnboardingState({ connections: input.connections ?? [], defaultSlug: input.defaultSlug, - sessions: input.sessions ?? [], + hasHistory: input.hasHistory ?? false, secrets: input.secrets ?? {}, }); } describe('deriveOnboardingState', () => { + it('uses history presence without requiring a full Session list', () => { + assert.deepEqual( + derive({ + connections: [realConnection()], + secrets: { 'anthropic-live': true }, + hasHistory: true, + }), + { + kind: 'ready_with_history', + connectionSlug: 'anthropic-live', + model: 'claude-sonnet-4-5-20250929', + }, + ); + }); it('routes the current default to the exact actionable fix', () => { const cases: Array<[LlmConnection, Record, OnboardingState]> = [ [ diff --git a/packages/core/src/onboarding.ts b/packages/core/src/onboarding.ts index 3e3d5290d41..8b6d8f3c40a 100644 --- a/packages/core/src/onboarding.ts +++ b/packages/core/src/onboarding.ts @@ -22,7 +22,6 @@ import { isConnectionReady, isRealConnection } from './connection-readiness.js'; import { normalizeOpenAiCodexConnection } from './model-catalog.js'; import { connectionEnabledModelIds, type LlmConnection } from './llm-connections.js'; -import type { SessionSummary } from './session.js'; export { hasSettledInitialOnboarding } from './onboarding-milestone.js'; /** Derived UI state; every non-ready variant identifies one actionable repair path. */ @@ -39,12 +38,8 @@ export interface DeriveOnboardingStateInput { connections: ReadonlyArray; /** Legacy preference used only to order otherwise-valid candidates. */ defaultSlug?: string | null; - /** - * All sessions known to storage. `ready_with_history` counts ANY - * non-deleted session, including archived and aborted ones — those - * are still user history. - */ - sessions: ReadonlyArray; + /** Any non-deleted session counts as history, including archived and aborted ones. */ + hasHistory: boolean; /** * Map of `slug → hasSecret` for every real connection in * `connections`. Caller resolves this asynchronously (credential @@ -76,7 +71,7 @@ export function deriveOnboardingState(input: DeriveOnboardingStateInput): Onboar requestedModel: model, }); if (verdict.ready) { - return hasHistory(input.sessions) + return input.hasHistory ? { kind: 'ready_with_history', connectionSlug: connection.slug, model: verdict.model } : { kind: 'ready_empty', connectionSlug: connection.slug, model: verdict.model }; } @@ -122,10 +117,6 @@ export function deriveOnboardingState(input: DeriveOnboardingStateInput): Onboar : { kind: 'blocked', reason: 'all_connections_unhealthy' }; } -function hasHistory(sessions: ReadonlyArray): boolean { - return sessions.length > 0; -} - /** Closed milestone ids persisted in settings. */ export const ONBOARDING_MILESTONE_IDS = [ 'initial_onboarding', From 15c2e5da7a41c489ce8ccbdc8144ff2199c15708 Mon Sep 17 00:00:00 2001 From: wutongyuonce <147830929+wutongyuonce@users.noreply.github.com> Date: Wed, 23 Sep 2026 21:39:33 +0800 Subject: [PATCH 2/2] fix(desktop): serialize onboarding reads and recover failed deltas --- .../onboarding-service-incremental.test.ts | 64 +++++++++++++++++-- .../__tests__/use-onboarding-snapshot.test.ts | 8 ++- apps/desktop/src/main/onboarding-service.ts | 49 +++++++------- .../src/renderer/use-onboarding-snapshot.ts | 1 + .../desktop-onboarding-incremental.md | 19 ++++-- 5 files changed, 98 insertions(+), 43 deletions(-) diff --git a/apps/desktop/src/main/__tests__/onboarding-service-incremental.test.ts b/apps/desktop/src/main/__tests__/onboarding-service-incremental.test.ts index 3be55ab7008..e0647d8699f 100644 --- a/apps/desktop/src/main/__tests__/onboarding-service-incremental.test.ts +++ b/apps/desktop/src/main/__tests__/onboarding-service-incremental.test.ts @@ -52,10 +52,19 @@ function harness(initial: SessionSummary[] = []) { let failGet = false; let credentialPresent = true; let nextGet: Promise | null = null; + let nextList: Promise | null = null; const service = createOnboardingService({ listConnections: async () => [connection], getDefaultSlug: async () => 'primary', - listSessions: async () => { listReads += 1; return [...sessions.values()]; }, + listSessions: async () => { + listReads += 1; + if (nextList) { + const held = nextList; + nextList = null; + return held; + } + return [...sessions.values()]; + }, getSession: async (id) => { getReads += 1; if (failGet) throw new Error('Host read failed'); @@ -80,6 +89,7 @@ function harness(initial: SessionSummary[] = []) { failNextGet: () => { failGet = true; }, setCredentialPresent: (present: boolean) => { credentialPresent = present; }, holdNextGet: (held: Promise) => { nextGet = held; }, + holdNextList: (held: Promise) => { nextList = held; }, }; } @@ -151,16 +161,58 @@ test('an update before initial coverage requests a resync', async () => { assert.equal(fixture.getReads(), 0); }); -test('a late targeted answer cannot replace a newer full observation', async () => { - const fixture = harness([session('one')]); +test('a complete read waits for an in-flight targeted read before replacing its basis', async () => { + const fixture = harness([session('one', 'ai-sdk')]); await fixture.service.getSnapshot(); let release!: (row: SessionSummary | null) => void; fixture.holdNextGet(new Promise((resolve) => { release = resolve; })); const pending = fixture.service.getSessionUpdate('one'); await Promise.resolve(); fixture.setCredentialPresent(false); - await fixture.service.getSnapshot(); - release(session('one')); - assert.deepEqual(await pending, { kind: 'resync' }); + const full = fixture.service.getSnapshot(); + await Promise.resolve(); + assert.equal(fixture.listReads(), 1, 'the complete read must wait for the targeted read'); + release(session('one', 'ai-sdk')); + assert.equal((await pending).kind, 'delta'); + assert.deepEqual((await full).sessionSendOutcomes.one, { + kind: 'blocked', reason: 'missing_api_key', connectionLocked: false, + }); assert.equal(fixture.listReads(), 2); }); + +test('a targeted read arriving during a complete read uses the newly accepted basis', async () => { + const fixture = harness([session('one', 'ai-sdk')]); + await fixture.service.getSnapshot(); + let release!: (rows: SessionSummary[]) => void; + fixture.holdNextList(new Promise((resolve) => { release = resolve; })); + fixture.setCredentialPresent(false); + const full = fixture.service.getSnapshot(); + await Promise.resolve(); + const pending = fixture.service.getSessionUpdate('one'); + await Promise.resolve(); + assert.equal(fixture.getReads(), 0, 'the targeted read must wait for the complete read'); + release([session('one', 'ai-sdk')]); + await full; + const update = await pending; + assert.equal(update.kind, 'delta'); + if (update.kind !== 'delta') return; + assert.deepEqual(update.outcome, { + kind: 'blocked', reason: 'missing_api_key', connectionLocked: false, + }); +}); + +test('a milestone write waits for an in-flight targeted read', async () => { + const fixture = harness(); + await fixture.service.getSnapshot(); + let release!: (row: SessionSummary | null) => void; + fixture.holdNextGet(new Promise((resolve) => { release = resolve; })); + const pending = fixture.service.getSessionUpdate('one'); + await Promise.resolve(); + const written = fixture.service.setMilestone('initial_onboarding', 'skipped'); + release(null); + const update = await pending; + assert.equal(update.kind, 'delta'); + if (update.kind !== 'delta') return; + assert.deepEqual(update.milestones, [], 'the earlier read must not borrow the later milestone'); + assert.equal((await written).milestones[0]?.id, 'initial_onboarding'); +}); 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 7f00a85f36b..269e1857be9 100644 --- a/apps/desktop/src/main/__tests__/use-onboarding-snapshot.test.ts +++ b/apps/desktop/src/main/__tests__/use-onboarding-snapshot.test.ts @@ -173,11 +173,12 @@ describe('createOnboardingSnapshotPoller', () => { assert.equal(emitted.length, 2); }); - it('keeps the accepted snapshot when a targeted Host read fails', async () => { + it('keeps the accepted snapshot on a targeted failure until a complete resync succeeds', async () => { const snapshots: OnboardingSnapshot[] = []; const errors: string[] = []; + let fullReads = 0; const poller = createOnboardingSnapshotPoller({ - getSnapshot: async () => READY_SNAPSHOT, + getSnapshot: async () => ++fullReads === 1 ? READY_SNAPSHOT : NEEDS_CONNECTION_SNAPSHOT, getSessionUpdate: async () => { throw new Error('Host disconnected'); }, }, { onSnapshot: (snapshot) => snapshots.push(snapshot), @@ -185,7 +186,8 @@ describe('createOnboardingSnapshotPoller', () => { }, () => 'zh-CN'); await poller.pull(); await poller.pullSession('one'); - assert.deepEqual(snapshots, [READY_SNAPSHOT]); + assert.deepEqual(snapshots, [READY_SNAPSHOT, NEEDS_CONNECTION_SNAPSHOT]); + assert.equal(fullReads, 2, 'the failed delta needs an authoritative resync'); assert.equal(errors.length, 1); }); diff --git a/apps/desktop/src/main/onboarding-service.ts b/apps/desktop/src/main/onboarding-service.ts index acc29d2cc72..8e7eaed3102 100644 --- a/apps/desktop/src/main/onboarding-service.ts +++ b/apps/desktop/src/main/onboarding-service.ts @@ -128,11 +128,15 @@ interface OnboardingBaseline { */ export function createOnboardingService(deps: OnboardingServiceDeps): OnboardingService { let baseline: OnboardingBaseline | null = null; - let revision = 0; let updateTail: Promise = Promise.resolve(); + function enqueue(read: () => Promise): Promise { + const pending = updateTail.then(read); + updateTail = pending.then(() => undefined, () => undefined); + return pending; + } + async function loadSnapshot(credentialFailureFallback = false): Promise { - const startedAt = revision; const [connections, defaultSlug, sessions, milestones] = await Promise.all([ deps.listConnections(), deps.getDefaultSlug(), @@ -165,16 +169,13 @@ export function createOnboardingService(deps: OnboardingServiceDeps): Onboarding const settledMilestones = hasHistory && !hasSettledInitialOnboarding(milestones) ? await deps.upsertMilestone('initial_onboarding', 'completed') : milestones; - if (startedAt === revision) { - baseline = { - sessionIds: new Set(sessions.map(({ id }) => id)), - connections, - defaultSlug, - secrets, - milestones: settledMilestones, - }; - revision += 1; - } + baseline = { + sessionIds: new Set(sessions.map(({ id }) => id)), + connections, + defaultSlug, + secrets, + milestones: settledMilestones, + }; return buildSnapshot(state, settledMilestones, sessions, connections, defaultSlug, secrets); } @@ -182,21 +183,16 @@ export function createOnboardingService(deps: OnboardingServiceDeps): Onboarding const observed = baseline; if (!observed) return { kind: 'resync' }; const session = await deps.getSession(sessionId); - // A concurrent global read or milestone write changed the basis used to - // project this answer. Let the caller resync instead of publishing it. - if (baseline !== observed) return { kind: 'resync' }; const hasHistory = session !== null || observed.sessionIds.size > (observed.sessionIds.has(sessionId) ? 1 : 0); const milestones = hasHistory && !hasSettledInitialOnboarding(observed.milestones) ? await deps.upsertMilestone('initial_onboarding', 'completed') : observed.milestones; - if (baseline !== observed) return { kind: 'resync' }; if (session === null) observed.sessionIds.delete(sessionId); else observed.sessionIds.add(sessionId); if (milestones !== observed.milestones) { observed.milestones = milestones; } - revision += 1; return { kind: 'delta', outcome: session === null ? null : projectSessionSendOutcome({ @@ -215,11 +211,9 @@ export function createOnboardingService(deps: OnboardingServiceDeps): Onboarding } return { - getSnapshot: () => loadSnapshot(), + getSnapshot: () => enqueue(() => loadSnapshot()), getSessionUpdate(sessionId: string): Promise { - const pending = updateTail.then(() => readSessionUpdate(sessionId)); - updateTail = pending.then(() => undefined, () => undefined); - return pending; + return enqueue(() => readSessionUpdate(sessionId)); }, async setMilestone(id: unknown, status: unknown): Promise { @@ -232,12 +226,13 @@ export function createOnboardingService(deps: OnboardingServiceDeps): Onboarding } // Timestamp is stamped inside the store (Date.now()); renderer // never controls it. - await deps.upsertMilestone(id, status); - revision += 1; - baseline = null; - // The global milestone write may alter which guide is shown; rebuild - // from the same authoritative sources as an explicit full refresh. - return loadSnapshot(true); + return enqueue(async () => { + await deps.upsertMilestone(id, status); + baseline = null; + // The global milestone write may alter which guide is shown; rebuild + // from the same authoritative sources as an explicit full refresh. + return loadSnapshot(true); + }); }, }; } diff --git a/apps/desktop/src/renderer/use-onboarding-snapshot.ts b/apps/desktop/src/renderer/use-onboarding-snapshot.ts index 01690f23c5a..b451cfa46a6 100644 --- a/apps/desktop/src/renderer/use-onboarding-snapshot.ts +++ b/apps/desktop/src/renderer/use-onboarding-snapshot.ts @@ -237,6 +237,7 @@ export function createOnboardingSnapshotPoller( } } catch (error) { if (!active || ticket !== inflightTicket || fullPending) return; + fullPending = true; emitError(onboardingSnapshotErrorMessage(error, getLocale())); } } diff --git a/docs/architecture/desktop-onboarding-incremental.md b/docs/architecture/desktop-onboarding-incremental.md index 78ab2733fd6..7091da89704 100644 --- a/docs/architecture/desktop-onboarding-incremental.md +++ b/docs/architecture/desktop-onboarding-incremental.md @@ -32,8 +32,9 @@ An initial or global onboarding read remains a complete, authenticated Owner snapshot. A subsequent identified Session event updates only that Session's send projection. First and last history transitions use the same Core onboarding rule as full reads. A failed targeted read leaves the last accepted projection -in place. Unknown authority, a changed Host epoch, or missing initial coverage -requests a complete resync. +in place until a complete recovery read succeeds. Unknown authority, a changed +Host epoch, missing initial coverage, or a failed targeted read requests a +complete resync. | Rule or state | Authority module | Interface / seam | Callers | Evidence | | --- | --- | --- | --- | --- | @@ -61,11 +62,14 @@ outcome and, for the default Host, its state and milestones. Guest changes do not change Owner onboarding. Connection, profile, unknown-Session, and manual invalidations request a full -snapshot. Main invalidates a targeted answer when its full-read inputs change -during the read. Renderer serializes reads, coalesces repeated invalidations, +snapshot. Main serializes complete reads, targeted reads, and milestone writes +per Host so each targeted projection uses the latest accepted full-read inputs. +Renderer serializes reads, coalesces repeated invalidations, and bounds the pending set at 64 distinct Session IDs before falling back to -one complete resync. It never publishes a response after disposal. A failed read preserves its -previous snapshot and exposes the existing generalized error. +one complete resync. It never publishes a response after disposal. A failed +targeted read preserves the previous snapshot, exposes the existing generalized +error, and schedules a complete resync. A failed complete read still preserves +the previous snapshot and exposes the error. ## Performance and limits @@ -89,7 +93,8 @@ React rendering and the existing stale-badge selector. ## Verification Owner-service tests cover first/last history, milestone backfill, targeted -projection parity, failed reads, and invalidated coverage. Preload seam tests +projection parity, failed reads, and ordering of full reads, targeted reads, +and milestone writes. Preload seam tests cover Owner/Guest routing and epoch replacement. The Renderer poller tests cover coalescing, late responses, and targeted versus global invalidation. The performance fixture records Host list/get counts, IPC bytes, and latency