From 0bd8753197274d2f6bddc54bce40d3d18da70df9 Mon Sep 17 00:00:00 2001 From: liugddx Date: Fri, 21 Aug 2026 17:37:44 +0800 Subject: [PATCH] fix(desktop): drain untracked runtime host before update Generated-by: Codex --- .../runtime-host-desktop-candidate.test.ts | 33 ++++- .../runtime-host-desktop-manager.test.ts | 101 ++++++++++++- apps/desktop/src/main/runtime-host-boot.ts | 3 + .../main/runtime-host-desktop-candidate.ts | 6 +- .../src/main/runtime-host-desktop-manager.ts | 23 ++- .../candidate-launch-barrier.test.ts | 137 ++++++++++++++++++ .../src/client/candidate-launch-barrier.ts | 118 +++++++++++++++ packages/runtime-host/src/client/index.ts | 4 + 8 files changed, 417 insertions(+), 8 deletions(-) create mode 100644 packages/runtime-host/src/__tests__/candidate-launch-barrier.test.ts create mode 100644 packages/runtime-host/src/client/candidate-launch-barrier.ts diff --git a/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts b/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts index 9f8ce91cf7..a87a64d2a6 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts @@ -5,7 +5,11 @@ import type { IpcMain } from 'electron'; import type { BotIncomingMessage, BotRegistry } from '@maka/runtime/bots'; import type { ComputerUseToolSet } from '@maka/runtime/computer-use-tools'; import type { MakaTool } from '@maka/runtime/tool-runtime'; -import type { ClientCapabilityProvider, RuntimeHostConnection } from '@maka/runtime-host/client'; +import type { + ClientCapabilityProvider, + ConnectOrSpawnRuntimeHostInput, + RuntimeHostConnection, +} from '@maka/runtime-host/client'; import { SESSION_CONTINUITY_SCHEMA_VERSION, type ClientCapabilityCallFrame, @@ -20,8 +24,10 @@ import { z } from 'zod'; import { createAttachmentApprovalRegistry } from '../attachment-approval.js'; import { createDesktopRuntimeHostCandidate as createCandidate, + startDesktopRuntimeHostCandidate, type DesktopRuntimeHostCandidateControls, type DesktopRuntimeHostCandidateDeps, + type DesktopRuntimeHostCandidateStartInput, } from '../runtime-host-desktop-candidate.js'; import { RuntimeHostSessionObservationRegistry } from '../runtime-host-session-observation-registry.js'; import { desktopSessionResourceKey } from '../../shared/runtime-host-identity.js'; @@ -29,6 +35,31 @@ import { desktopSessionResourceKey } from '../../shared/runtime-host-identity.js const TEST_HOST_ID = 'a'.repeat(64); const TEST_TARGET_EPOCH = 'test-target-epoch'; +test('uses the manager-owned launch barrier for local candidate startup', async () => { + let connectedRoot: string | undefined; + const result = await startDesktopRuntimeHostCandidate({ + rootPath: 'C:\\workspace', + candidateEntrypoint: 'candidate.js', + ipcMain: { + epoch: TEST_TARGET_EPOCH, + isActive: () => true, + }, + candidateLaunchBarrier: { + connect: async (input: ConnectOrSpawnRuntimeHostInput) => { + connectedRoot = input.rootPath; + return { kind: 'failed', reason: 'startup_timeout' }; + }, + pause: () => undefined, + retireExcept: async () => undefined, + resume: () => undefined, + release: () => undefined, + }, + } as unknown as DesktopRuntimeHostCandidateStartInput); + + assert.deepEqual(result, { kind: 'failed', reason: 'startup_timeout' }); + assert.equal(connectedRoot, 'C:\\workspace'); +}); + function createDesktopRuntimeHostCandidate( connection: RuntimeHostConnection, candidateDeps: DesktopRuntimeHostCandidateDeps, diff --git a/apps/desktop/src/main/__tests__/runtime-host-desktop-manager.test.ts b/apps/desktop/src/main/__tests__/runtime-host-desktop-manager.test.ts index 2128fd324a..da9762da31 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-desktop-manager.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-desktop-manager.test.ts @@ -120,16 +120,108 @@ test('quiesces reconnect and waits for the Host process before update install', await owner.close(); }); -test('keeps the current Host when update preparation reports active tasks', async () => { +test('retires unadopted candidates before draining the tracked Host', async () => { + const events: string[] = []; + const current = candidateHarness({ + disconnectOnPrepare: true, + onPrepare: () => events.push('prepare-host'), + }); + const owner = await startRuntimeHostDesktopManager({ + candidateLaunchBarrier: { + connect: async () => assert.fail('mocked candidate startup bypasses the barrier'), + pause: () => events.push('pause-launches'), + retireExcept: async (pid: number) => { + events.push(`retire-except:${pid}`); + }, + resume: () => events.push('resume-launches'), + release: () => events.push('release-launches'), + }, + } as unknown as DesktopRuntimeHostCandidateStartInput, { + startCandidate: async () => ready(current.candidate), + waitForHostExit: async (pid) => { + events.push(`wait:${pid}`); + }, + }); + + const preparation = await owner.prepareForUpdate(false); + assert.equal(preparation.kind, 'prepared'); + assert.deepEqual(events, [ + 'pause-launches', + 'retire-except:42', + 'prepare-host', + 'wait:42', + ]); + if (preparation.kind === 'prepared') preparation.rollback(); + assert.equal(events.at(-1), 'resume-launches'); + await owner.close(); + assert.equal(events.at(-1), 'release-launches'); +}); + +test('resumes candidate launches when active tasks block the update', async () => { + const events: string[] = []; const current = candidateHarness({ activeTasks: true }); + const owner = await startRuntimeHostDesktopManager({ + candidateLaunchBarrier: { + connect: async () => assert.fail('mocked candidate startup bypasses the barrier'), + pause: () => events.push('pause'), + retireExcept: async () => { + events.push('retire'); + }, + resume: () => events.push('resume'), + release: () => events.push('release'), + }, + } as unknown as DesktopRuntimeHostCandidateStartInput, { + startCandidate: async () => ready(current.candidate), + }); + + assert.deepEqual(await owner.prepareForUpdate(false), { kind: 'active_tasks' }); + assert.deepEqual(events, ['pause', 'retire', 'resume']); + await owner.close(); + assert.equal(events.at(-1), 'release'); +}); + +test('resumes candidate launches when candidate retirement fails', async () => { + const events: string[] = []; + const current = candidateHarness(); + const owner = await startRuntimeHostDesktopManager({ + candidateLaunchBarrier: { + connect: async () => assert.fail('mocked candidate startup bypasses the barrier'), + pause: () => events.push('pause'), + retireExcept: async () => { + events.push('retire'); + throw new Error('retirement failed'); + }, + resume: () => events.push('resume'), + release: () => events.push('release'), + }, + } as unknown as DesktopRuntimeHostCandidateStartInput, { + startCandidate: async () => ready(current.candidate), + }); + + await assert.rejects(owner.prepareForUpdate(false), /retirement failed/); + assert.deepEqual(events, ['pause', 'retire', 'resume']); + await owner.handleBotIncomingMessage({ text: 'still connected' } as BotIncomingMessage); + assert.equal(current.botMessages, 1); + await owner.close(); +}); + +test('keeps active-task confirmation bound to the current Host', async () => { + const current = candidateHarness({ activeTasks: true, disconnectOnPrepare: true }); + const waitedFor: number[] = []; const owner = await startRuntimeHostDesktopManager({} as DesktopRuntimeHostCandidateStartInput, { startCandidate: async () => ready(current.candidate), + waitForHostExit: async (pid) => { + waitedFor.push(pid); + }, }); assert.deepEqual(await owner.prepareForUpdate(false), { kind: 'active_tasks' }); await owner.handleBotIncomingMessage({ text: 'still connected' } as BotIncomingMessage); assert.equal(current.botMessages, 1); - assert.deepEqual(current.prepareUpgradeAuthorities, [false]); + const authorized = await owner.prepareForUpdate(true); + assert.equal(authorized.kind, 'prepared'); + assert.deepEqual(current.prepareUpgradeAuthorities, [false, true]); + assert.deepEqual(waitedFor, [42]); await owner.close(); }); @@ -672,6 +764,7 @@ function candidateHarness( hostId?: string; finalizeFailures?: Error[]; disconnectOnFinalizeFailure?: boolean; + onPrepare?: () => void; } = {}, ) { let resolveClosed: (() => void) | undefined; @@ -693,7 +786,11 @@ function candidateHarness( get lifecycleState() { return lifecycleState; }, + async queryHostDiagnostics() { + return { pid: 42 }; + }, async prepareHostUpgrade(allowInterruptActiveTasks: boolean) { + options.onPrepare?.(); prepareUpgradeCalls += 1; prepareUpgradeAuthorities.push(allowInterruptActiveTasks); if (options.activeTasks && !allowInterruptActiveTasks) { diff --git a/apps/desktop/src/main/runtime-host-boot.ts b/apps/desktop/src/main/runtime-host-boot.ts index 5883af3d96..76932aa736 100644 --- a/apps/desktop/src/main/runtime-host-boot.ts +++ b/apps/desktop/src/main/runtime-host-boot.ts @@ -21,6 +21,7 @@ import { buildMcpTools } from '@maka/runtime/mcp-tools'; import { createClientRuntimeHostCredentialStore, createClientRuntimeHostProfileCatalog, + createRuntimeHostCandidateLaunchBarrier, LOCAL_RUNTIME_HOST_PROFILE, loadOrCreateRuntimeHostClientInstanceId, } from "@maka/runtime-host/client"; @@ -168,6 +169,7 @@ const userDataDir = app.getPath("userData"); const runtimeHostClientInstanceId = await loadOrCreateRuntimeHostClientInstanceId( join(userDataDir, "runtime-host-client.json"), ); +const runtimeHostCandidateLaunchBarrier = createRuntimeHostCandidateLaunchBarrier(); const runtimeHostCredentialStore = createClientRuntimeHostCredentialStore(userDataDir); const runtimeHostProfileCatalog = createClientRuntimeHostProfileCatalog( userDataDir, @@ -509,6 +511,7 @@ runtimeHostManager = await startRuntimeHostDesktopManager( rootPath: workspaceRoot, clientInstanceId: runtimeHostClientInstanceId, generation: runtimeHostGeneration, + candidateLaunchBarrier: runtimeHostCandidateLaunchBarrier, // The Desktop E2E composition lives behind its own entry module, which // release packaging drops: picking it here is what keeps FakeBackend and // the E2E bootstrap out of the shipped Runtime Host. diff --git a/apps/desktop/src/main/runtime-host-desktop-candidate.ts b/apps/desktop/src/main/runtime-host-desktop-candidate.ts index b6b3069fe7..22fe5fd70d 100644 --- a/apps/desktop/src/main/runtime-host-desktop-candidate.ts +++ b/apps/desktop/src/main/runtime-host-desktop-candidate.ts @@ -13,6 +13,7 @@ import { type ConnectOrSpawnRuntimeHostInput, type ConnectOrSpawnRuntimeHostResult, type RuntimeHostConnection, + type RuntimeHostCandidateLaunchBarrier, type RemoteRuntimeHostProfile, } from "@maka/runtime-host/client"; import { @@ -138,6 +139,7 @@ export interface DesktopRuntimeHostCandidateStartInput readonly generation?: string; readonly takeoverHostEpoch?: string; readonly signal?: AbortSignal; + readonly candidateLaunchBarrier?: RuntimeHostCandidateLaunchBarrier; readonly remote?: { readonly profile: RemoteRuntimeHostProfile; readonly credential: string; @@ -259,7 +261,9 @@ export async function startDesktopRuntimeHostCandidate( ipcMain, ); } - const connection = await connectOrSpawnRuntimeHost(connectInput(input)); + const connection = input.candidateLaunchBarrier + ? await input.candidateLaunchBarrier.connect(connectInput(input)) + : await connectOrSpawnRuntimeHost(connectInput(input)); if (connection.kind !== "connected") return connection; try { return { diff --git a/apps/desktop/src/main/runtime-host-desktop-manager.ts b/apps/desktop/src/main/runtime-host-desktop-manager.ts index f2de4b09ec..ed8ee13f7a 100644 --- a/apps/desktop/src/main/runtime-host-desktop-manager.ts +++ b/apps/desktop/src/main/runtime-host-desktop-manager.ts @@ -453,24 +453,38 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager { this.#requireTarget(LOCAL_RUNTIME_HOST_PROFILE.id), ); const quiescence = lifecycle.quiesce(); + let launchBarrierPaused = false; + const resume = () => { + if (launchBarrierPaused) { + launchBarrierPaused = false; + this.#baseInput.candidateLaunchBarrier?.resume(); + } + quiescence.resume(); + }; try { if ( quiescence.current.hostLifecycleMode === 'service' || quiescence.current.hostLifecycleMode === 'remote' ) { - return { kind: 'prepared', rollback: quiescence.resume }; + return { kind: 'prepared', rollback: resume }; } + this.#baseInput.candidateLaunchBarrier?.pause(); + launchBarrierPaused = this.#baseInput.candidateLaunchBarrier !== undefined; + const diagnostics = await quiescence.current.client.queryHostDiagnostics(); + // The adopted Host still owns the root here, so every other owned launch + // can be settled without allowing it to become a late election winner. + await this.#baseInput.candidateLaunchBarrier?.retireExcept(diagnostics.pid); const result = await quiescence.current.client.prepareHostUpgrade( allowInterruptActiveTasks, ); if (result.kind === 'active_tasks') { - quiescence.resume(); + resume(); return result; } await this.waitForHostExit(result.pid); - return { kind: 'prepared', rollback: quiescence.resume }; + return { kind: 'prepared', rollback: resume }; } catch (error) { - quiescence.resume(); + resume(); throw error; } } @@ -489,6 +503,7 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager { const results = await Promise.allSettled( [...this.#targets.values()].map((target) => this.#removeTarget(target)), ); + this.#baseInput.candidateLaunchBarrier?.release(); this.#ipcMain.close(); const failures = results.filter( (result): result is PromiseRejectedResult => result.status === 'rejected', diff --git a/packages/runtime-host/src/__tests__/candidate-launch-barrier.test.ts b/packages/runtime-host/src/__tests__/candidate-launch-barrier.test.ts new file mode 100644 index 0000000000..850ad55348 --- /dev/null +++ b/packages/runtime-host/src/__tests__/candidate-launch-barrier.test.ts @@ -0,0 +1,137 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import type { ConnectOrSpawnRuntimeHostInput } from '../client/connect-or-spawn.js'; +import { + createRuntimeHostCandidateLaunchBarrierWithDependencies, + type RuntimeHostCandidateLaunchBarrierDependencies, +} from '../client/candidate-launch-barrier.js'; +import type { OwnedCandidateAttempt } from '../client/launcher.js'; + +const connectInput = {} as ConnectOrSpawnRuntimeHostInput; +const launchInput = { + rootPath: 'C:\\workspace', + expectedRootId: 'root-id', + entrypoint: 'candidate.js', +}; + +test('retires every owned candidate except the adopted Host while launches are paused', async () => { + const first = candidateAttempt(42); + const second = candidateAttempt(77); + const attempts = [first, second]; + const barrier = createRuntimeHostCandidateLaunchBarrierWithDependencies({ + launchCandidate: () => ({ spawned: Promise.resolve(attempts.shift()!.attempt) }), + connect: async (_input, launchCandidate) => { + await Promise.all([ + launchCandidate(launchInput).spawned, + launchCandidate(launchInput).spawned, + ]); + return { kind: 'failed', reason: 'startup_timeout' }; + }, + retireTimeoutMs: 25, + }); + + await barrier.connect(connectInput); + barrier.pause(); + await barrier.retireExcept(42); + + assert.deepEqual(first.settleTimeouts, []); + assert.deepEqual(second.settleTimeouts, [25]); + barrier.release(); + assert.equal(first.releaseCalls, 1); + assert.equal(second.releaseCalls, 0); +}); + +test('waits for a committed candidate spawn before retirement', async () => { + const late = candidateAttempt(77); + let resolveSpawn!: (attempt: OwnedCandidateAttempt) => void; + const spawned = new Promise((resolve) => { + resolveSpawn = resolve; + }); + const barrier = createRuntimeHostCandidateLaunchBarrierWithDependencies({ + launchCandidate: () => ({ spawned }), + connect: async (_input, launchCandidate) => { + void launchCandidate(launchInput); + return { kind: 'failed', reason: 'startup_timeout' }; + }, + retireTimeoutMs: 25, + }); + + await barrier.connect(connectInput); + barrier.pause(); + const retirement = barrier.retireExcept(42); + await new Promise((resolve) => setImmediate(resolve)); + assert.deepEqual(late.settleTimeouts, []); + resolveSpawn(late.attempt); + await retirement; + assert.deepEqual(late.settleTimeouts, [25]); +}); + +test('pause blocks new candidate connections until rollback resumes them', async () => { + let connectCalls = 0; + const dependencies: RuntimeHostCandidateLaunchBarrierDependencies = { + launchCandidate: () => assert.fail('this connection does not need a candidate'), + connect: async () => { + connectCalls += 1; + return { kind: 'failed', reason: 'startup_timeout' }; + }, + retireTimeoutMs: 25, + }; + const barrier = createRuntimeHostCandidateLaunchBarrierWithDependencies(dependencies); + + barrier.pause(); + await assert.rejects(barrier.connect(connectInput), /candidate launches are paused/); + assert.equal(connectCalls, 0); + barrier.resume(); + assert.deepEqual(await barrier.connect(connectInput), { + kind: 'failed', + reason: 'startup_timeout', + }); + assert.equal(connectCalls, 1); +}); + +test('releases a candidate that finishes spawning after the barrier closes', async () => { + const late = candidateAttempt(77); + let resolveSpawn!: (attempt: OwnedCandidateAttempt) => void; + const spawned = new Promise((resolve) => { + resolveSpawn = resolve; + }); + const barrier = createRuntimeHostCandidateLaunchBarrierWithDependencies({ + launchCandidate: () => ({ spawned }), + connect: async (_input, launchCandidate) => { + void launchCandidate(launchInput); + return { kind: 'failed', reason: 'startup_timeout' }; + }, + retireTimeoutMs: 25, + }); + + await barrier.connect(connectInput); + barrier.release(); + resolveSpawn(late.attempt); + await new Promise((resolve) => setImmediate(resolve)); + + assert.equal(late.releaseCalls, 1); + assert.deepEqual(late.settleTimeouts, []); +}); + +function candidateAttempt(pid: number) { + let releaseCalls = 0; + const settleTimeouts: number[] = []; + const attempt: OwnedCandidateAttempt = { + pid, + startupFailure: new Promise(() => undefined), + releaseToEnvironment() { + releaseCalls += 1; + }, + async settle(timeoutMs) { + settleTimeouts.push(timeoutMs); + return false; + }, + }; + return { + attempt, + settleTimeouts, + get releaseCalls() { + return releaseCalls; + }, + }; +} diff --git a/packages/runtime-host/src/client/candidate-launch-barrier.ts b/packages/runtime-host/src/client/candidate-launch-barrier.ts new file mode 100644 index 0000000000..e53fd08f7a --- /dev/null +++ b/packages/runtime-host/src/client/candidate-launch-barrier.ts @@ -0,0 +1,118 @@ +import type { + ConnectOrSpawnRuntimeHostInput, + ConnectOrSpawnRuntimeHostResult, +} from './connect-or-spawn.js'; +import { connectOrSpawnRuntimeHostWithDependencies } from './connect-or-spawn.js'; +import { + launchOwnedRuntimeHostCandidate, + type CandidateLauncher, + type OwnedCandidateAttempt, +} from './launcher.js'; + +const DEFAULT_RETIRE_TIMEOUT_MS = 1_000; + +export interface RuntimeHostCandidateLaunchBarrier { + connect(input: ConnectOrSpawnRuntimeHostInput): Promise; + pause(): void; + retireExcept(protectedPid: number): Promise; + resume(): void; + release(): void; +} + +export interface RuntimeHostCandidateLaunchBarrierDependencies { + readonly launchCandidate: typeof launchOwnedRuntimeHostCandidate; + readonly connect: ( + input: ConnectOrSpawnRuntimeHostInput, + launchCandidate: CandidateLauncher, + ) => Promise; + readonly retireTimeoutMs: number; +} + +const defaultDependencies: RuntimeHostCandidateLaunchBarrierDependencies = { + launchCandidate: launchOwnedRuntimeHostCandidate, + connect: (input, launchCandidate) => + connectOrSpawnRuntimeHostWithDependencies(input, { + launchCandidate, + random: Math.random, + }), + retireTimeoutMs: DEFAULT_RETIRE_TIMEOUT_MS, +}; + +export function createRuntimeHostCandidateLaunchBarrier(): RuntimeHostCandidateLaunchBarrier { + return createRuntimeHostCandidateLaunchBarrierWithDependencies(defaultDependencies); +} + +export function createRuntimeHostCandidateLaunchBarrierWithDependencies( + dependencies: RuntimeHostCandidateLaunchBarrierDependencies, +): RuntimeHostCandidateLaunchBarrier { + return new RuntimeHostCandidateLaunchBarrierImpl(dependencies); +} + +class RuntimeHostCandidateLaunchBarrierImpl implements RuntimeHostCandidateLaunchBarrier { + readonly #attempts = new Set(); + readonly #pending = new Set>(); + #state: 'active' | 'paused' | 'released' = 'active'; + + constructor(private readonly dependencies: RuntimeHostCandidateLaunchBarrierDependencies) {} + + connect(input: ConnectOrSpawnRuntimeHostInput): Promise { + if (this.#state !== 'active') { + return Promise.reject(new Error('Runtime Host candidate launches are paused')); + } + return this.dependencies.connect(input, (candidate) => this.#launch(candidate)); + } + + pause(): void { + if (this.#state !== 'active') { + throw new Error('Runtime Host candidate launch barrier is not active'); + } + this.#state = 'paused'; + } + + async retireExcept(protectedPid: number): Promise { + if (this.#state !== 'paused') { + throw new Error('Runtime Host candidate launches must be paused before retirement'); + } + await Promise.allSettled([...this.#pending]); + const retiring = [...this.#attempts].filter((attempt) => attempt.pid !== protectedPid); + await Promise.all( + retiring.map(async (attempt) => { + await attempt.settle(this.dependencies.retireTimeoutMs); + this.#attempts.delete(attempt); + }), + ); + } + + resume(): void { + if (this.#state === 'paused') this.#state = 'active'; + } + + release(): void { + if (this.#state === 'released') return; + this.#state = 'released'; + for (const attempt of this.#attempts) attempt.releaseToEnvironment(); + this.#attempts.clear(); + } + + #launch(input: Parameters[0]): ReturnType { + if (this.#state !== 'active') { + throw new Error('Runtime Host candidate launches are paused'); + } + const launch = this.dependencies.launchCandidate(input); + const pending = launch.spawned; + this.#pending.add(pending); + void pending.then( + (attempt) => { + this.#pending.delete(pending); + if (this.#state === 'released') { + attempt.releaseToEnvironment(); + return; + } + this.#attempts.add(attempt); + void attempt.startupFailure?.finally(() => this.#attempts.delete(attempt)); + }, + () => this.#pending.delete(pending), + ); + return launch; + } +} diff --git a/packages/runtime-host/src/client/index.ts b/packages/runtime-host/src/client/index.ts index bc19e6c9b5..21f3d15e9a 100644 --- a/packages/runtime-host/src/client/index.ts +++ b/packages/runtime-host/src/client/index.ts @@ -74,6 +74,10 @@ export { type ConnectOrSpawnRuntimeHostInput, type ConnectOrSpawnRuntimeHostResult, } from './connect-or-spawn.js'; +export { + createRuntimeHostCandidateLaunchBarrier, + type RuntimeHostCandidateLaunchBarrier, +} from './candidate-launch-barrier.js'; export { runHostedExecution } from './hosted-execution.js'; export { type ClientCapabilityProvider } from './client-capability.js'; export {