You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by wy...@apache.org on 2026/09/16 02:25:13 UTC
(maka) branch feat/runtime-host-rust updated: feat(desktop): manage native hosts over SSH and WSL
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a commit to branch feat/runtime-host-rust
in repository https://gitbox.apache.org/repos/asf/maka.git
The following commit(s) were added to refs/heads/feat/runtime-host-rust by this push:
new 17c1d8ca9c feat(desktop): manage native hosts over SSH and WSL
17c1d8ca9c is described below
commit 17c1d8ca9ceb27030d1a88abce2045a154e5f6a6
Author: Wang <m4...@gmail.com>
AuthorDate: Wed Sep 16 10:24:58 2026 +0800
feat(desktop): manage native hosts over SSH and WSL
Coordinate remote retirement through trusted operators with exact connection identities while retaining per-profile reconnect fences.
Generated-by: OpenAI Codex
---
.../__tests__/runtime-host-desktop-manager.test.ts | 61 +++++++++--
.../src/main/native-runtime-host-command.ts | 104 +++++++++++++++++++
.../src/main/native-runtime-host-management.ts | 68 ++++++------
apps/desktop/src/main/runtime-host-boot.ts | 36 ++++++-
apps/desktop/src/main/runtime-host-client.ts | 4 +
.../src/main/runtime-host-desktop-candidate.ts | 1 +
.../src/main/runtime-host-desktop-manager.ts | 72 +++++++++----
apps/desktop/src/preload/bridge-contract.d.ts | 1 +
apps/desktop/src/preload/preload.ts | 4 +-
.../native-runtime-host-management-dialog.tsx | 17 +--
.../renderer/settings/projects-settings-page.tsx | 2 +-
.../settings/runtime-host-profiles-section.tsx | 17 ++-
.../src/renderer/settings/settings-surface.tsx | 11 +-
crates/cli/src/args.rs | 10 +-
crates/cli/src/deployment/update.rs | 2 +-
crates/cli/src/host_client.rs | 2 +
crates/protocol/src/host.rs | 9 +-
crates/runtime-host/src/server.rs | 14 ++-
crates/runtime-host/src/server/diagnostics.rs | 10 +-
crates/runtime-host/src/server/dispatch.rs | 5 +-
crates/runtime-host/src/server/handshake.rs | 18 ++--
crates/runtime-host/src/server/retirement.rs | 33 +++++-
.../runtime-host/tests/integration/retirement.rs | 57 ++++++++++-
docs/rust-runtime.md | 3 +
docs/rust-runtime.zh-CN.md | 3 +
tests/fixtures/client-native-managed-entry.mjs | 114 ++++++++++++++++-----
tests/fixtures/client.mjs | 29 +++---
27 files changed, 562 insertions(+), 145 deletions(-)
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 b3b099ee82..1af20325f9 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
@@ -33,6 +33,7 @@ import {
type HostHandoffAction,
type OpenHostHandoffSurface,
HostHandoffRequiredError,
+ LOCAL_RUNTIME_HOST_PROFILE,
} from '@maka/runtime-host/client';
import {
INTERACTIVE_RUNTIME_HOST_COMPOSITION_ID,
@@ -208,17 +209,17 @@ test('native management rejects stale retirement and retains stop intent until c
waitForHostExit: async () => {},
});
try {
- await assert.rejects(owner.runNativeLocalHostChange(async (scope) => {
+ await assert.rejects(owner.runNativeHostChange({ profile: LOCAL_RUNTIME_HOST_PROFILE }, async (scope) => {
await scope.retire({ ...deployment, configRevision: 1 });
}), /no longer matches/);
assert.equal(first.prepareRetirementCalls, 0);
assert.equal(resumes, 1);
- await owner.runNativeLocalHostChange(async (scope) => {
+ await owner.runNativeHostChange({ profile: LOCAL_RUNTIME_HOST_PROFILE }, async (scope) => {
assert.equal(await scope.retire(deployment), true);
scope.hold();
});
assert.equal(first.prepareRetirementCalls, 1);
- await assert.rejects(owner.runNativeLocalHostChange(async (scope) => {
+ await assert.rejects(owner.runNativeHostChange({ profile: LOCAL_RUNTIME_HOST_PROFILE }, async (scope) => {
scope.hold();
throw new Error('activation receipt lost');
}), /receipt lost/);
@@ -226,7 +227,7 @@ test('native management rejects stale retirement and retains stop intent until c
assert.equal(starts, 1);
assert.equal(pauses, 2); // Failed start reused the retained suspension.
assert.equal(resumes, 1);
- await owner.runNativeLocalHostChange(async (scope) => {
+ await owner.runNativeHostChange({ profile: LOCAL_RUNTIME_HOST_PROFILE }, async (scope) => {
scope.hold();
scope.resumeOnSuccess();
});
@@ -238,6 +239,53 @@ test('native management rejects stale retirement and retains stop intent until c
}
});
+test('native remote management pins its epoch and uses only its operator without affecting local ownership', async () => {
+ const target = { ...remoteTarget('native-remote'), profileIncarnationId: 'persisted-incarnation' };
+ const local = candidateHarness();
+ const remote = candidateHarness({ ownership: 'external', hostId: target.profile.rootId });
+ const deployment = {
+ deploymentId: 'f7672ac5-42f0-4456-8a46-0e36f6d6137b', configRevision: 2,
+ rootId: target.profile.rootId, rootPath: '/remote/root', executable: '/remote/maka', sha256: 'a'.repeat(64),
+ mode: 'on_demand' as const, websocket: '127.0.0.1:0',
+ };
+ const owner = await startRuntimeHostDesktopManager({
+ candidateLaunchBarrier: {
+ pause() { assert.fail('remote mutation paused local launches'); },
+ resume() { assert.fail('remote mutation resumed local launches'); }, async retireExcept() {},
+ },
+ } as unknown as DesktopRuntimeHostCandidateStartInput, {
+ startCandidate: async (input) => ready(input.profileTarget?.profile.kind === 'remote' ? remote.candidate : local.candidate),
+ waitForHostExit: async () => { assert.fail('remote PID is not a local process'); },
+ });
+ try {
+ await owner.enable(target);
+ await owner.waitUntilReady(target.profile.id);
+ await assert.rejects(owner.runNativeHostChange({ ...target, profileIncarnationId: 'replaced' }, async () => {
+ assert.fail('replaced profile entered mutation');
+ }), /target changed/);
+ await assert.rejects(owner.runNativeHostChange(target, async (scope) => {
+ await scope.retire(deployment, 'stale-epoch', async () => { assert.fail('stale operator invoked'); });
+ }), /no longer matches/);
+ await owner.runNativeHostChange(target, async (scope) => {
+ assert.equal(await scope.retire(deployment, 'test-host-epoch', async (id) => {
+ assert.equal(id, 'desktop-connection');
+ assert.equal(remote.closeCalls, 0);
+ return false;
+ }), false);
+ });
+ await owner.runNativeHostChange(target, async (scope) => {
+ assert.equal(await scope.retire(deployment, 'test-host-epoch', async () => true), true);
+ scope.hold();
+ });
+ assert.equal(remote.prepareRetirementCalls, 0);
+ assert.equal(remote.closeCalls, 1);
+ assert.equal(local.closeCalls, 0);
+ await assert.rejects(owner.runNativeHostChange(remoteTarget('guest', 'default', 'session_guest'), async () => {}), /Guest/);
+ } finally {
+ await owner.close();
+ }
+});
+
test('native retirement refusal releases a new pause, but ambiguous retirement retains it', async () => {
let uncertain = false;
const first = candidateHarness({
@@ -253,10 +301,10 @@ test('native retirement refusal releases a new pause, but ambiguous retirement r
startCandidate: async () => ready(first.candidate),
});
try {
- await owner.runNativeLocalHostChange(async (scope) => assert.equal(await scope.retire(null), false));
+ await owner.runNativeHostChange({ profile: LOCAL_RUNTIME_HOST_PROFILE }, async (scope) => assert.equal(await scope.retire(null), false));
assert.equal(resumes, 1);
uncertain = true;
- await assert.rejects(owner.runNativeLocalHostChange(async (scope) => {
+ await assert.rejects(owner.runNativeHostChange({ profile: LOCAL_RUNTIME_HOST_PROFILE }, async (scope) => {
await scope.retire(null);
}), /reply lost/);
assert.equal(resumes, 1);
@@ -1987,6 +2035,7 @@ function candidateHarness(
hostPid: 42,
...(options.ownedProcess ? { ownedProcess: options.ownedProcess } : {}),
client: {
+ connectionId: 'desktop-connection',
hostId: options.hostId ?? 'test-host',
hostEpoch: options.hostEpoch ?? 'test-host-epoch',
get lifecycleState() {
diff --git a/apps/desktop/src/main/native-runtime-host-command.ts b/apps/desktop/src/main/native-runtime-host-command.ts
new file mode 100644
index 0000000000..803b32cf2f
--- /dev/null
+++ b/apps/desktop/src/main/native-runtime-host-command.ts
@@ -0,0 +1,104 @@
+/*
+ * 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 { spawn } from 'node:child_process';
+import {
+ normalizeRuntimeHostSshDestination,
+ normalizeRuntimeHostWslDistribution,
+ resolveSystemRuntimeHostWslExecutable,
+ runtimeHostSshOperatorRemoteCommand,
+} from '@maka/runtime-host/client';
+import { runtimeHostOperatorInvocation, type RuntimeHostNativeOperatorCommand } from '@maka/runtime-host/operator';
+
+export type NativeRuntimeHostOperator =
+ | { readonly kind: 'local'; readonly executable: string }
+ | { readonly kind: 'ssh'; readonly destination: string; readonly sshPort?: number;
+ readonly operator: RuntimeHostNativeOperatorCommand }
+ | { readonly kind: 'wsl'; readonly distribution: string;
+ readonly operator: RuntimeHostNativeOperatorCommand<'posix'> };
+
+/** Drain and reap even on excess output: never kill an in-flight durable mutation. */
+export function runNativeRuntimeHostCommand(
+ target: NativeRuntimeHostOperator,
+ args: readonly string[],
+ readOnly = false,
+): Promise<string> {
+ const command = invocation(target, args);
+ return new Promise((resolve, reject) => {
+ const child = spawn(command.executable, command.args, {
+ windowsHide: true, stdio: ['ignore', 'pipe', 'pipe'],
+ });
+ const output: Buffer[] = [];
+ let bytes = 0;
+ let errorText = '';
+ // A readonly query can be abandoned. Mutations must settle or explicitly
+ // report an unknown outcome when the underlying connection breaks.
+ const timer = readOnly ? setTimeout(() => child.kill(), 15_000) : undefined;
+ child.stdout.on('data', (chunk: Buffer) => {
+ bytes += chunk.length;
+ if (bytes <= 256 * 1024) output.push(chunk);
+ });
+ child.stderr.setEncoding('utf8');
+ child.stderr.on('data', (chunk: string) => { errorText = (errorText + chunk).slice(-8192); });
+ child.once('error', (error) => {
+ clearTimeout(timer);
+ reject(error);
+ });
+ child.once('close', (code) => {
+ clearTimeout(timer);
+ if (code !== 0 || bytes > 256 * 1024) {
+ reject(new Error(`Native Host operation was not confirmed; refresh status before retrying. ${errorText.trim()}`));
+ } else resolve(Buffer.concat(output).toString('utf8'));
+ });
+ });
+}
+
+function invocation(target: NativeRuntimeHostOperator, args: readonly string[]) {
+ switch (target.kind) {
+ case 'local':
+ return { executable: target.executable, args: ['host', ...args] };
+ case 'wsl': {
+ const command = runtimeHostOperatorInvocation(target.operator, args);
+ return {
+ executable: resolveSystemRuntimeHostWslExecutable(),
+ args: ['--distribution', normalizeRuntimeHostWslDistribution(target.distribution),
+ '--exec', command.executable, ...command.args],
+ };
+ }
+ case 'ssh': {
+ if (target.sshPort !== undefined &&
+ (!Number.isSafeInteger(target.sshPort) || target.sshPort < 1 || target.sshPort > 65_535)) {
+ throw new Error('Invalid SSH port');
+ }
+ return {
+ executable: 'ssh',
+ args: [
+ '-T', '-o', 'BatchMode=yes', '-o', 'ControlMaster=no', '-o', 'ControlPath=none',
+ '-o', 'ClearAllForwardings=yes', '-o', 'RemoteCommand=none', '-o', 'ConnectTimeout=15',
+ '-o', 'ServerAliveInterval=15', '-o', 'ServerAliveCountMax=5',
+ ...(target.sshPort === undefined ? [] : ['-p', String(target.sshPort)]),
+ normalizeRuntimeHostSshDestination(target.destination),
+ runtimeHostSshOperatorRemoteCommand(target.operator, args),
+ ],
+ };
+ }
+ }
+}
+
diff --git a/apps/desktop/src/main/native-runtime-host-management.ts b/apps/desktop/src/main/native-runtime-host-management.ts
index e65a1e312a..049366a299 100644
--- a/apps/desktop/src/main/native-runtime-host-management.ts
+++ b/apps/desktop/src/main/native-runtime-host-management.ts
@@ -18,10 +18,11 @@
*/
-import { spawn } from 'node:child_process';
import { decodeRuntimeHostActivationFrame } from '@maka/runtime-host/operator';
+import { z } from 'zod';
import {
nativeRuntimeHostDeploymentSchema,
+ decodeNativeRuntimeHostDeploymentStatus,
type NativeRuntimeHostDeployment,
} from '../shared/native-runtime-host-deployment.js';
import {
@@ -34,31 +35,34 @@ import {
type NativeRuntimeHostMutation,
type NativeRuntimeHostSettings,
} from '../shared/native-runtime-host-management.js';
-import { readNativeRuntimeHostDeployment } from './native-runtime-host-deployment.js';
+import { runNativeRuntimeHostCommand, type NativeRuntimeHostOperator } from './native-runtime-host-command.js';
export interface NativeRuntimeHostChangeScope {
/** Once a mutation may have run, reconnect stays paused until confirmed ready. */
hold(): void;
- retire(deployment: NativeRuntimeHostDeployment | null): Promise<boolean>;
+ retire(deployment: NativeRuntimeHostDeployment | null, hostEpoch?: string,
+ prepareRemote?: (connectionId: string) => Promise<boolean>): Promise<boolean>;
resumeOnSuccess(): void;
}
export function createNativeRuntimeHostManagement(input: {
- readonly executable: string;
+ readonly operator: NativeRuntimeHostOperator;
readonly rootId: string;
- readonly rootPath: string;
+ readonly rootPath?: string;
readonly change: <T>(run: (scope: NativeRuntimeHostChangeScope) => Promise<T>) => Promise<T>;
}) {
- const read = () => readNativeRuntimeHostDeployment(input.executable, input.rootId);
- const execute = (args: string[]) => runNativeRuntimeHostCommand(input.executable, ['host', ...args]);
+ const execute = (args: string[], readOnly = false) => runNativeRuntimeHostCommand(input.operator, args, readOnly);
const rooted = (action: string) => [action, '--root-id', input.rootId];
+ const read = async () => decodeNativeRuntimeHostDeploymentStatus(
+ JSON.parse(await execute(rooted('status'), true)), input.rootId,
+ );
const requireCurrent = async (expected: NativeRuntimeHostExpected) => {
const status = await read();
if (status.kind !== 'installed' || status.deployment.deploymentId !== expected.deploymentId ||
status.deployment.configRevision !== expected.configRevision) {
throw new Error('Native Host deployment changed; refresh before applying this operation');
}
- return status.deployment;
+ return status;
};
const activate = async (deployment: NativeRuntimeHostDeployment): Promise<NativeRuntimeHostMutation> => {
const receipt = decodeRuntimeHostActivationFrame((await execute([...rooted('activate'), '--framed'])).trim());
@@ -71,6 +75,18 @@ export function createNativeRuntimeHostManagement(input: {
hostEpoch: receipt.hostEpoch, pid: receipt.pid, port: receipt.endpoint.port,
} };
};
+ const retire = (scope: NativeRuntimeHostChangeScope, deployment: NativeRuntimeHostDeployment, hostEpoch?: string) =>
+ scope.retire(deployment, hostEpoch, async (connectionId) => {
+ if (!hostEpoch) throw new Error('Native Host epoch is unavailable; refresh before retiring it');
+ const result = z.discriminatedUnion('kind', [
+ z.object({ kind: z.literal('active_tasks') }).strict(),
+ z.object({ kind: z.literal('prepared'), pid: z.number().int().positive().max(0xffff_ffff) }).strict(),
+ ]).parse(JSON.parse(await execute([
+ 'retire', '--root', deployment.rootPath, '--expected-host-epoch', hostEpoch,
+ '--handoff-connection-id', connectionId,
+ ])));
+ return result.kind === 'prepared';
+ });
const mutate = async (
action: 'stop' | 'restart' | 'uninstall' | 'update' | 'reconcile',
current: NativeRuntimeHostDeployment,
@@ -97,10 +113,12 @@ export function createNativeRuntimeHostManagement(input: {
(status.kind === 'installed' && status.deployment.admission !== 'revoked')) {
throw new Error('Native Host is already installed or incomplete; refresh deployment status');
}
+ const rootPath = status.kind === 'installed' ? status.deployment.rootPath : input.rootPath;
+ if (!rootPath) throw new Error('Native Host installation requires a known Root path');
if (!await scope.retire(null)) return { kind: 'active_tasks' };
scope.hold();
const installed = nativeRuntimeHostDeploymentSchema.parse(JSON.parse(await execute([
- 'install', '--root', input.rootPath, ...settingsArguments(request.settings),
+ 'install', '--root', rootPath, ...settingsArguments(request.settings),
])));
if (installed.rootId !== input.rootId || installed.admission !== undefined) {
throw new Error('Native Host install returned another Root or revoked deployment');
@@ -121,7 +139,9 @@ export function createNativeRuntimeHostManagement(input: {
return result;
}
- const current = await requireCurrent(request.expected);
+ const observed = await requireCurrent(request.expected);
+ const current = observed.deployment;
+ const hostEpoch = observed.host.kind === 'connected' ? observed.host.identity.hostEpoch : undefined;
if (request.action === 'update' || request.action === 'reconcile') {
// Stage under native authority while the old Host is still serving.
scope.hold();
@@ -134,13 +154,13 @@ export function createNativeRuntimeHostManagement(input: {
if (staged.kind !== 'active_tasks' || !staged.target) {
throw new Error('Native Host did not confirm a pending update');
}
- if (!await scope.retire(current)) return staged;
+ if (!await retire(scope, current, hostEpoch)) return staged;
const result = await mutate('reconcile', current);
if (result.kind === 'ready') scope.resumeOnSuccess();
return result;
}
- if (!await scope.retire(current)) return { kind: 'active_tasks', deployment: current };
+ if (!await retire(scope, current, hostEpoch)) return { kind: 'active_tasks', deployment: current };
scope.hold();
const result = await mutate(request.action, current);
if (result.kind === 'ready') scope.resumeOnSuccess();
@@ -151,7 +171,7 @@ export function createNativeRuntimeHostManagement(input: {
const request = nativeRuntimeHostManagementRequestSchema.parse(value);
if (request.action === 'status') return { status: await read() };
if (request.action === 'logs') {
- const logs = nativeRuntimeHostLogsSchema.parse(JSON.parse(await execute(rooted('logs'))));
+ const logs = nativeRuntimeHostLogsSchema.parse(JSON.parse(await execute(rooted('logs'), true)));
return { status: await read(), logs };
}
return input.change(async (scope) => {
@@ -211,25 +231,3 @@ function validateMutation(
}
throw new Error('Native Host operation returned an unexpected revision or outcome');
}
-
-/** Drain and reap even on excess output: never kill an in-flight durable mutation. */
-function runNativeRuntimeHostCommand(executable: string, args: string[]): Promise<string> {
- return new Promise((resolve, reject) => {
- const child = spawn(executable, args, { windowsHide: true, stdio: ['ignore', 'pipe', 'pipe'] });
- const output: Buffer[] = [];
- let bytes = 0;
- let errorText = '';
- child.stdout.on('data', (chunk: Buffer) => {
- bytes += chunk.length;
- if (bytes <= 256 * 1024) output.push(chunk);
- });
- child.stderr.setEncoding('utf8');
- child.stderr.on('data', (chunk: string) => { errorText = (errorText + chunk).slice(-8192); });
- child.once('error', reject);
- child.once('close', (code) => {
- if (code !== 0 || bytes > 256 * 1024) {
- reject(new Error(`Native Host operation was not confirmed; refresh status before retrying. ${errorText.trim()}`));
- } else resolve(Buffer.concat(output).toString('utf8'));
- });
- });
-}
diff --git a/apps/desktop/src/main/runtime-host-boot.ts b/apps/desktop/src/main/runtime-host-boot.ts
index 9f87146522..68dd53e301 100644
--- a/apps/desktop/src/main/runtime-host-boot.ts
+++ b/apps/desktop/src/main/runtime-host-boot.ts
@@ -242,6 +242,7 @@ import { createDesktopLocalRuntimeHostRemoteAccess } from './runtime-host-local-
import { createDesktopRuntimeHostOnboarding } from "./runtime-host-onboarding.js";
import { createDesktopRuntimeHostManagement } from "./runtime-host-management.js";
import { createNativeRuntimeHostManagement } from './native-runtime-host-management.js';
+import type { NativeRuntimeHostOperator } from './native-runtime-host-command.js';
import { createDesktopRuntimeHostLocalManagement } from './runtime-host-local-management.js';
import { createDesktopRuntimeHostPeerMeshManagement } from './runtime-host-peer-mesh-management.js';
import { registerExternalAgentSetupIpc } from "./external-agent-setup-ipc-main.js";
@@ -762,16 +763,43 @@ const runtimeHostOnboarding = createDesktopRuntimeHostOnboarding({
mainWindowController.send("runtime-host-onboarding:changed", snapshot),
});
const nativeRuntimeHostManagement = isE2e ? undefined : createNativeRuntimeHostManagement({
- executable: nativeHostExecutable,
+ operator: { kind: 'local', executable: nativeHostExecutable },
rootId: startupLocalStorageRoot.rootId,
rootPath: startupLocalStorageRoot.canonicalPath,
change: (run) => {
if (!runtimeHostManager) throw new Error('Runtime Host manager is unavailable');
- return runtimeHostManager.runNativeLocalHostChange(run);
+ return runtimeHostManager.runNativeHostChange({ profile: LOCAL_RUNTIME_HOST_PROFILE }, run);
},
});
-ipcMain.handle('runtime-host-management:native', (_event, request: unknown) =>
- nativeRuntimeHostManagement?.run(request) ?? null);
+ipcMain.handle('runtime-host-management:native', async (_event, request: unknown, profileId: unknown = 'local') => {
+ if (typeof profileId !== 'string' || !profileId || profileId.length > 128) {
+ throw new Error('Invalid native Host profile ID');
+ }
+ if (profileId === 'local') return nativeRuntimeHostManagement?.run(request) ?? null;
+ const target = await runtimeHostProfileCatalog.resolve(profileId);
+ const profile = target.profile;
+ let operator: NativeRuntimeHostOperator;
+ if (profile.kind === 'environment' && profile.operator.kind === 'native') {
+ operator = { kind: 'wsl', distribution: profile.provider.distribution, operator: profile.operator };
+ } else if (profile.kind === 'remote' && profile.access !== 'session_guest' &&
+ profile.transport.kind === 'ssh' && profile.transport.activation?.operator.kind === 'native') {
+ operator = {
+ kind: 'ssh', destination: profile.transport.destination,
+ ...(profile.transport.sshPort === undefined ? {} : { sshPort: profile.transport.sshPort }),
+ operator: profile.transport.activation.operator,
+ };
+ } else {
+ throw new Error('This profile has no native Host operator');
+ }
+ return createNativeRuntimeHostManagement({
+ operator,
+ rootId: profile.rootId,
+ change: (run) => {
+ if (!runtimeHostManager) throw new Error('Runtime Host manager is unavailable');
+ return runtimeHostManager.runNativeHostChange(target, run);
+ },
+ }).run(request);
+});
const localRuntimeHostManagement = createDesktopRuntimeHostLocalManagement({
remoteAccess: localRuntimeHostRemoteAccess,
diff --git a/apps/desktop/src/main/runtime-host-client.ts b/apps/desktop/src/main/runtime-host-client.ts
index ab21f66cd7..bc7cd54a98 100644
--- a/apps/desktop/src/main/runtime-host-client.ts
+++ b/apps/desktop/src/main/runtime-host-client.ts
@@ -290,6 +290,10 @@ export class DesktopRuntimeHostClient {
return this.connection.hostEpoch;
}
+ get connectionId(): string {
+ return this.connection.connectionId;
+ }
+
get hostId(): string {
return this.rootId;
}
diff --git a/apps/desktop/src/main/runtime-host-desktop-candidate.ts b/apps/desktop/src/main/runtime-host-desktop-candidate.ts
index de3cca6e2b..366d0f62e2 100644
--- a/apps/desktop/src/main/runtime-host-desktop-candidate.ts
+++ b/apps/desktop/src/main/runtime-host-desktop-candidate.ts
@@ -235,6 +235,7 @@ export interface DesktopRuntimeHostCandidateStartInput
readonly profileTarget?: {
readonly profile: PersistedRuntimeHostProfile;
readonly credential?: string;
+ readonly profileIncarnationId?: string;
readonly sshInteraction?: RuntimeHostSshInteraction;
};
}
diff --git a/apps/desktop/src/main/runtime-host-desktop-manager.ts b/apps/desktop/src/main/runtime-host-desktop-manager.ts
index 18d9646ec2..0e5ee924ed 100644
--- a/apps/desktop/src/main/runtime-host-desktop-manager.ts
+++ b/apps/desktop/src/main/runtime-host-desktop-manager.ts
@@ -109,7 +109,7 @@ export interface RuntimeHostDesktopManager {
signal?: AbortSignal,
): Promise<void>;
runManagedLocalHostChange<T>(change: () => Promise<T>): Promise<T>;
- runNativeLocalHostChange<T>(change: (scope: NativeRuntimeHostChangeScope) => Promise<T>): Promise<T>;
+ runNativeHostChange<T>(target: ResolvedRuntimeHostProfile, change: (scope: NativeRuntimeHostChangeScope) => Promise<T>): Promise<T>;
setDefaultProfile(profileId: string): void;
retireOwnedLocalHost(mode: RuntimeHostRetirementMode): Promise<DesktopLocalHostRetirement>;
prepareOwnedLocalHostQuit(mode: RuntimeHostRetirementMode): Promise<'ready' | 'active_tasks'>;
@@ -235,6 +235,7 @@ interface DesktopRuntimeHostTargetGeneration {
state: RuntimeHostDesktopTargetState;
hostId?: string;
lifecycle?: RuntimeHostReconnectLifecycle<DesktopRuntimeHostCandidate>;
+ nativeSuspension?: { resume(): void };
unsubscribeLifecycle?: () => void;
unsubscribeRoutes?: () => void;
skipPeerRouteRefreshOnce?: boolean;
@@ -312,7 +313,6 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager {
#defaultProfileId: string = LOCAL_RUNTIME_HOST_PROFILE.id;
#localHostRetirement: Extract<PreparedLocalHostRetirement, { kind: 'retired' }> | undefined;
#localHostRetirementTask: DesktopLocalHostRetirementTask | undefined;
- #nativeLocalSuspension: { resume(): void } | undefined;
#closed = false;
#closeTask: Promise<void> | undefined;
@@ -765,20 +765,36 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager {
});
}
- runNativeLocalHostChange<T>(change: (scope: NativeRuntimeHostChangeScope) => Promise<T>): Promise<T> {
- return this.#mutateTarget(LOCAL_RUNTIME_HOST_PROFILE.id, async () => {
- const lifecycle = this.#requireLifecycle(this.#requireTarget(LOCAL_RUNTIME_HOST_PROFILE.id));
- const alreadyPaused = this.#nativeLocalSuspension !== undefined;
- if (!this.#nativeLocalSuspension) {
+ runNativeHostChange<T>(expectedTarget: ResolvedRuntimeHostProfile, change: (scope: NativeRuntimeHostChangeScope) => Promise<T>): Promise<T> {
+ return this.#mutateTarget(expectedTarget.profile.id, async () => {
+ if (expectedTarget.profile.kind === 'remote' && expectedTarget.profile.access === 'session_guest') {
+ throw new Error('Session Guest profiles cannot manage native Hosts');
+ }
+ const target = this.#targets.get(expectedTarget.profile.id);
+ if (!target) {
+ if (expectedTarget.profile.kind === 'local') throw new Error('Local Host manager is unavailable');
+ // Disabled profiles have no reconnect or client lifetime to suspend.
+ // Their native operator still acquires the exact deployment and Root.
+ return change({ hold() {}, async retire() { return true; }, resumeOnSuccess() {} });
+ }
+ if (!sameResolvedRuntimeHostProfileTarget(target.target, expectedTarget) ||
+ target.target.profileIncarnationId !== expectedTarget.profileIncarnationId) {
+ throw new Error('Native Host management target changed or is not an owner');
+ }
+ const local = expectedTarget.profile.kind === 'local';
+ const barrier = local ? this.#baseInput.candidateLaunchBarrier : undefined;
+ const lifecycle = this.#requireLifecycle(target);
+ const alreadyPaused = target.nativeSuspension !== undefined;
+ if (!target.nativeSuspension) {
const suspension = await lifecycle.suspend();
try {
- this.#baseInput.candidateLaunchBarrier?.pause();
+ barrier?.pause();
} catch (error) {
suspension.resume();
throw error;
}
- this.#nativeLocalSuspension = { resume: () => {
- this.#baseInput.candidateLaunchBarrier?.resume();
+ target.nativeSuspension = { resume: () => {
+ barrier?.resume();
suspension.resume();
} };
}
@@ -787,27 +803,36 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager {
let succeeded = false;
try {
// Finish pre-existing launches before a management command can acquire Root.
- await this.#baseInput.candidateLaunchBarrier?.retireExcept(lifecycle.current?.hostPid ?? -1);
+ await barrier?.retireExcept(lifecycle.current?.hostPid ?? -1);
const result = await change({
hold: () => { keepPaused = true; },
resumeOnSuccess: () => { resumeOnSuccess = true; },
- retire: async (deployment) => {
+ retire: async (deployment, hostEpoch, prepareRemote) => {
const current = lifecycle.current;
if (!current) return true; // Native RootOwner still proves exclusivity.
- if (deployment
- ? current.hostOwnership !== 'managed' || current.client.hostId !== deployment.rootId ||
- current.managedDeployment?.deploymentId !== deployment.deploymentId ||
- current.managedDeployment.configRevision !== deployment.configRevision
- : current.hostOwnership !== 'owned_ephemeral') {
+ const matches = deployment
+ ? current.client.hostId === deployment.rootId && (local
+ ? current.hostOwnership === 'managed' &&
+ current.managedDeployment?.deploymentId === deployment.deploymentId &&
+ current.managedDeployment.configRevision === deployment.configRevision
+ : hostEpoch !== undefined && current.client.hostEpoch === hostEpoch)
+ : local && current.hostOwnership === 'owned_ephemeral';
+ if (!matches) {
throw new Error('Connected native Host no longer matches the management target');
}
keepPaused = true; // RPC failure may follow a successful retirement.
- const retirement = await current.client.prepareHostRetirement('refuse_active_work');
- if (retirement.kind === 'active_tasks') {
+ const retirement = local
+ ? await current.client.prepareHostRetirement('refuse_active_work')
+ : undefined;
+ if (!local && !prepareRemote) throw new Error('Native remote retirement requires its trusted operator');
+ const prepared = retirement
+ ? retirement.kind === 'prepared'
+ : await prepareRemote!(current.client.connectionId);
+ if (!prepared) {
keepPaused = alreadyPaused;
return false;
}
- await this.waitForHostExit(retirement.pid);
+ if (retirement?.kind === 'prepared') await this.waitForHostExit(retirement.pid);
await current.close();
return true;
},
@@ -816,8 +841,8 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager {
return result;
} finally {
if ((!keepPaused || (succeeded && resumeOnSuccess)) && !this.#closed) {
- const suspension = this.#nativeLocalSuspension;
- this.#nativeLocalSuspension = undefined;
+ const suspension = target.nativeSuspension;
+ target.nativeSuspension = undefined;
suspension?.resume();
}
}
@@ -1444,6 +1469,9 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager {
...(input.profileTarget.credential === undefined
? {}
: { credential: input.profileTarget.credential }),
+ ...(input.profileTarget.profileIncarnationId === undefined
+ ? {}
+ : { profileIncarnationId: input.profileTarget.profileIncarnationId }),
}
: { profile: LOCAL_RUNTIME_HOST_PROFILE };
const epoch = randomUUID();
diff --git a/apps/desktop/src/preload/bridge-contract.d.ts b/apps/desktop/src/preload/bridge-contract.d.ts
index af098a6d30..be5225710e 100644
--- a/apps/desktop/src/preload/bridge-contract.d.ts
+++ b/apps/desktop/src/preload/bridge-contract.d.ts
@@ -882,6 +882,7 @@ export interface MakaBridge {
runtimeHostManagement: {
runNative(
request: import('../shared/native-runtime-host-management.js').NativeRuntimeHostManagementRequest,
+ profileId?: string,
): Promise<import('../shared/native-runtime-host-management.js').NativeRuntimeHostManagementResult | null>;
run(
profileId: string,
diff --git a/apps/desktop/src/preload/preload.ts b/apps/desktop/src/preload/preload.ts
index b03dc91cca..80bbe9c371 100644
--- a/apps/desktop/src/preload/preload.ts
+++ b/apps/desktop/src/preload/preload.ts
@@ -1680,8 +1680,8 @@ const makaBridge = {
},
},
runtimeHostManagement: {
- runNative(request) {
- return ipcRenderer.invoke('runtime-host-management:native', request);
+ runNative(request, profileId = 'local') {
+ return ipcRenderer.invoke('runtime-host-management:native', request, profileId);
},
run(
profileId: string,
diff --git a/apps/desktop/src/renderer/settings/native-runtime-host-management-dialog.tsx b/apps/desktop/src/renderer/settings/native-runtime-host-management-dialog.tsx
index 9fe4e46936..7a36951a70 100644
--- a/apps/desktop/src/renderer/settings/native-runtime-host-management-dialog.tsx
+++ b/apps/desktop/src/renderer/settings/native-runtime-host-management-dialog.tsx
@@ -31,11 +31,14 @@ import type {
import { getSettingsProjectsCopy } from '../locales/settings-projects-copy.js';
import { RuntimeHostProjectDirectoryEditor } from './runtime-host-project-directory-editor.js';
-export function NativeRuntimeHostManagementDialog(props: { readonly onClose: () => void }) {
+export function NativeRuntimeHostManagementDialog(props: {
+ readonly target: { readonly id: string; readonly name: string };
+ readonly onClose: () => void;
+}) {
const locale = useUiLocale();
const zh = locale === 'zh-CN';
const copy = {
- title: zh ? '本机 Host' : 'Local Host',
+ title: props.target.name,
refresh: zh ? '刷新' : 'Refresh',
install: zh ? '安装并启动' : 'Install and start',
start: zh ? '启动' : 'Start',
@@ -44,7 +47,9 @@ export function NativeRuntimeHostManagementDialog(props: { readonly onClose: ()
uninstall: zh ? '卸载（保留数据）' : 'Uninstall (retain data)',
confirm: zh ? '确认卸载' : 'Confirm uninstall',
cancel: zh ? '取消' : 'Cancel',
- update: zh ? '应用此 Desktop 的 Host 版本' : 'Apply this Desktop’s Host version',
+ update: props.target.id === 'local'
+ ? (zh ? '应用此 Desktop 的 Host 版本' : 'Apply this Desktop’s Host version')
+ : (zh ? '应用所选 Host 版本' : 'Apply selected Host version'),
edit: zh ? '编辑配置' : 'Edit configuration',
save: zh ? '应用配置与当前版本' : 'Apply settings and current version',
reconcile: zh ? '继续待处理更新' : 'Continue pending update',
@@ -80,17 +85,17 @@ export function NativeRuntimeHostManagementDialog(props: { readonly onClose: ()
const alive = useRef(true);
useEffect(() => {
alive.current = true;
- void window.maka.runtimeHostManagement.runNative({ action: 'status' }).then((value) => {
+ void window.maka.runtimeHostManagement.runNative({ action: 'status' }, props.target.id).then((value) => {
if (alive.current && value) setResult(value);
}, (cause: unknown) => { if (alive.current) setError(String(cause)); });
return () => { alive.current = false; };
- }, []);
+ }, [props.target.id]);
async function run(request: NativeRuntimeHostManagementRequest) {
setBusy(true);
setError(undefined);
try {
- const value = await window.maka.runtimeHostManagement.runNative(request);
+ const value = await window.maka.runtimeHostManagement.runNative(request, props.target.id);
if (!alive.current) return;
if (!value) throw new Error('Native Host management is unavailable');
setResult(value);
diff --git a/apps/desktop/src/renderer/settings/projects-settings-page.tsx b/apps/desktop/src/renderer/settings/projects-settings-page.tsx
index 0a5fa0a080..716e40ce89 100644
--- a/apps/desktop/src/renderer/settings/projects-settings-page.tsx
+++ b/apps/desktop/src/renderer/settings/projects-settings-page.tsx
@@ -77,7 +77,7 @@ export function ProjectsSettingsPage(props: {
): Promise<UpdateAppSettingsResult>;
onRetryRuntimeHost(): Promise<void>;
onRemoteHostAdded(profileId: string): void;
- onManageNativeHost(): void;
+ onManageNativeHost(target: { readonly id: string; readonly name: string }): void;
}) {
const host = useOptionalRuntimeHostSettingsTarget();
const locale = useUiLocale();
diff --git a/apps/desktop/src/renderer/settings/runtime-host-profiles-section.tsx b/apps/desktop/src/renderer/settings/runtime-host-profiles-section.tsx
index eeeb006e64..9585bcdcd7 100644
--- a/apps/desktop/src/renderer/settings/runtime-host-profiles-section.tsx
+++ b/apps/desktop/src/renderer/settings/runtime-host-profiles-section.tsx
@@ -92,7 +92,7 @@ function createRemoteHostDraft() {
export function RuntimeHostProfilesSection(props: {
readonly onRemoteHostAdded: (profileId: string) => void;
- readonly onManageNativeHost: () => void;
+ readonly onManageNativeHost: (target: { readonly id: string; readonly name: string }) => void;
}) {
const locale = useUiLocale();
const copy = getSettingsProjectsCopy(locale).runtimeHost;
@@ -418,7 +418,9 @@ export function RuntimeHostProfilesSection(props: {
{nativeManagementAvailable ? <SettingsRow
label={locale === 'zh-CN' ? '本机 Host' : 'Local Host'}
end={<Button variant="secondary" size="sm" label={copy.manage}
- isDisabled={switching} onClick={props.onManageNativeHost} />}
+ isDisabled={switching} onClick={() => props.onManageNativeHost({
+ id: 'local', name: locale === 'zh-CN' ? '本机 Host' : 'Local Host',
+ })} />}
/> : nativeManagementAvailable === false ? <SettingsRow
label={copy.thisComputerRemoteAccess}
description={
@@ -653,6 +655,10 @@ export function RuntimeHostProfilesSection(props: {
{connectedEntries.map((entry) => {
const profile = entry.profile;
if (profile.kind === 'local') return null;
+ const nativeManagement = profile.kind === 'environment'
+ ? profile.operator.kind === 'native'
+ : profile.access !== 'session_guest' && profile.transport.kind === 'ssh' &&
+ profile.transport.activation?.operator.kind === 'native';
const managementTarget: RuntimeHostManagementTarget | undefined =
entry.managedService
? profile.kind === 'environment'
@@ -745,7 +751,12 @@ export function RuntimeHostProfilesSection(props: {
onChanged={() => void reload()}
onWorkingChange={setSwitching}
items={[
- ...(managementTarget && !entry.pairingPending
+ ...(nativeManagement && !entry.pairingPending ? [{
+ label: copy.manage,
+ isDisabled: switching,
+ onClick: () => props.onManageNativeHost({ id: profile.id, name: profile.name }),
+ }] : []),
+ ...(managementTarget && !nativeManagement && !entry.pairingPending
? [{
label: copy.manage,
isDisabled: switching,
diff --git a/apps/desktop/src/renderer/settings/settings-surface.tsx b/apps/desktop/src/renderer/settings/settings-surface.tsx
index bb0af08a9a..5fc8e32889 100644
--- a/apps/desktop/src/renderer/settings/settings-surface.tsx
+++ b/apps/desktop/src/renderer/settings/settings-surface.tsx
@@ -295,7 +295,7 @@ function SettingsSurfaceContent(
const [clientSettings, setClientSettings] = useState(
initialClientSettings ?? defaultSettings,
);
- const [showNativeManagement, setShowNativeManagement] = useState(false);
+ const [nativeManagementTarget, setNativeManagementTarget] = useState<{ readonly id: string; readonly name: string }>();
const [runtimeHostSettings, setRuntimeHostSettings] = useState<
SettingsResourceState<RuntimeHostAppSettings>
>(() => createSettingsResourceState(
@@ -1045,7 +1045,7 @@ function SettingsSurfaceContent(
archivedTasks={props.archivedTasks}
onTaskImported={props.onTaskImported}
onRemoteHostAdded={props.onRemoteHostAdded}
- onManageNativeHost={() => setShowNativeManagement(true)}
+ onManageNativeHost={setNativeManagementTarget}
openProviderCatalog={providerCatalogRequested}
initialConnectionSlug={props.initialConnectionSlug}
initialCreateProviderType={createProviderRequest}
@@ -1067,14 +1067,15 @@ function SettingsSurfaceContent(
</section>
)}
/>
- {showNativeManagement ? <NativeRuntimeHostManagementDialog
- onClose={() => setShowNativeManagement(false)} /> : null}
+ {nativeManagementTarget ? <NativeRuntimeHostManagementDialog
+ key={nativeManagementTarget.id} target={nativeManagementTarget}
+ onClose={() => setNativeManagementTarget(undefined)} /> : null}
</div>
);
}
function SettingsPageBody(props: {
- onManageNativeHost(): void;
+ onManageNativeHost(target: { readonly id: string; readonly name: string }): void;
section: SettingsSection;
isLocalRuntimeHost: boolean;
settings: AppSettings;
diff --git a/crates/cli/src/args.rs b/crates/cli/src/args.rs
index d018dae960..b1166f3c7f 100644
--- a/crates/cli/src/args.rs
+++ b/crates/cli/src/args.rs
@@ -78,6 +78,9 @@ enum HostCommand {
root: Root,
#[arg(long)]
expected_host_epoch: Option<String>,
+ /// Coordinate with this exact connected client without interrupting other clients.
+ #[arg(long, requires = "expected_host_epoch")]
+ handoff_connection_id: Option<uuid::Uuid>,
/// Permit interrupting other clients and non-cooperative resources.
#[arg(long)]
allow_interrupt_active_tasks: bool,
@@ -151,11 +154,16 @@ impl Cli {
Command::Host(HostCommand::Retire {
root,
expected_host_epoch,
+ handoff_connection_id,
allow_interrupt_active_tasks,
}) => {
let mut client = crate::host_client::HostClient::connect(&root.root, None).await?;
let result = client
- .retire(expected_host_epoch.as_deref(), allow_interrupt_active_tasks)
+ .retire(
+ expected_host_epoch.as_deref(),
+ allow_interrupt_active_tasks,
+ handoff_connection_id,
+ )
.await?;
drop(client);
println!("{}", serde_json::to_string(&result)?);
diff --git a/crates/cli/src/deployment/update.rs b/crates/cli/src/deployment/update.rs
index ac2cadd084..b0b4ca62a4 100644
--- a/crates/cli/src/deployment/update.rs
+++ b/crates/cli/src/deployment/update.rs
@@ -212,7 +212,7 @@ pub(super) async fn retire(deployment: &Deployment) -> Result<Option<RootOwner>,
if let Ok(mut client) =
HostClient::connect(&deployment.root_path, Some(&deployment.generation())).await
{
- match client.retire(None, false).await? {
+ match client.retire(None, false, None).await? {
RetirementResult::ActiveTasks => return Ok(None),
RetirementResult::Prepared { .. } => {}
}
diff --git a/crates/cli/src/host_client.rs b/crates/cli/src/host_client.rs
index 66a6e064a9..396c6db286 100644
--- a/crates/cli/src/host_client.rs
+++ b/crates/cli/src/host_client.rs
@@ -185,6 +185,7 @@ impl HostClient {
&mut self,
expected_epoch: Option<&str>,
interrupt: bool,
+ handoff_connection_id: Option<uuid::Uuid>,
) -> Result<RetirementResult, HostError> {
if expected_epoch.is_some_and(|epoch| epoch != self.discovery.host_epoch) {
return Err("Host epoch does not match the retirement target".into());
@@ -193,6 +194,7 @@ impl HostClient {
expected_host_epoch: self.discovery.host_epoch.clone(),
allow_interrupt_active_tasks: interrupt,
allow_cooperative_handoff: Some(true),
+ handoff_connection_id: handoff_connection_id.map(|id| id.to_string()),
};
let result = decode_retirement_result(
&self
diff --git a/crates/protocol/src/host.rs b/crates/protocol/src/host.rs
index 4563c9834f..7e5e5b3de2 100644
--- a/crates/protocol/src/host.rs
+++ b/crates/protocol/src/host.rs
@@ -32,6 +32,9 @@ pub struct RetirementInput {
pub allow_interrupt_active_tasks: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub allow_cooperative_handoff: Option<bool>,
+ /// A trusted operator may coordinate retirement with this still-connected client.
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub handoff_connection_id: Option<String>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -45,7 +48,7 @@ pub fn decode_retirement_input(value: &Value) -> Result<RetirementInput> {
codec::shaped(
codec::record(value, "Host retirement")?,
&["expectedHostEpoch", "allowInterruptActiveTasks"],
- &["allowCooperativeHandoff"],
+ &["allowCooperativeHandoff", "handoffConnectionId"],
)?;
let boolean = |value: &Value| {
value
@@ -59,6 +62,10 @@ pub fn decode_retirement_input(value: &Value) -> Result<RetirementInput> {
.get("allowCooperativeHandoff")
.map(boolean)
.transpose()?,
+ handoff_connection_id: value
+ .get("handoffConnectionId")
+ .map(|value| codec::string(value, "Handoff connection ID", 128))
+ .transpose()?,
})
}
diff --git a/crates/runtime-host/src/server.rs b/crates/runtime-host/src/server.rs
index fc96afb63b..f17ef2bee0 100644
--- a/crates/runtime-host/src/server.rs
+++ b/crates/runtime-host/src/server.rs
@@ -104,7 +104,7 @@ impl Default for HostOptions {
pub struct Host {
options: HostOptions,
retirement: Arc<Mutex<retirement::Phase>>,
- accepted_connections: AtomicUsize,
+ accepted_connections: Mutex<std::collections::HashSet<Uuid>>,
started: std::time::Instant,
accepted_connection_revision: AtomicU64,
executions: Arc<crate::execution::Executions>,
@@ -227,7 +227,7 @@ impl Host {
let host = Arc::new(Self {
options,
retirement: interactions.retirement.clone(),
- accepted_connections: AtomicUsize::new(0),
+ accepted_connections: Mutex::default(),
started: std::time::Instant::now(),
accepted_connection_revision: AtomicU64::new(0),
shells: executions.shells.clone(),
@@ -349,3 +349,13 @@ impl Drop for ConnectionCount<'_> {
self.0.fetch_sub(1, Ordering::SeqCst);
}
}
+
+struct AcceptedConnection<'a> {
+ connections: &'a Mutex<std::collections::HashSet<Uuid>>,
+ id: Uuid,
+}
+impl Drop for AcceptedConnection<'_> {
+ fn drop(&mut self) {
+ self.connections.lock().unwrap().remove(&self.id);
+ }
+}
diff --git a/crates/runtime-host/src/server/diagnostics.rs b/crates/runtime-host/src/server/diagnostics.rs
index 05c5215f15..0c6a261e8c 100644
--- a/crates/runtime-host/src/server/diagnostics.rs
+++ b/crates/runtime-host/src/server/diagnostics.rs
@@ -19,7 +19,7 @@
use super::Host;
use maka_protocol::host::{Diagnostics, Platform, Residency, Status};
-use std::{collections::VecDeque, sync::atomic::Ordering};
+use std::collections::VecDeque;
pub(super) struct Activity {
connections: usize,
@@ -56,8 +56,14 @@ impl Activity {
impl Host {
pub(super) fn activity(&self) -> Activity {
+ self.activity_except(None)
+ }
+
+ pub(super) fn activity_except(&self, handoff_connection: Option<uuid::Uuid>) -> Activity {
+ let connections = self.accepted_connections.lock().unwrap();
Activity {
- connections: self.accepted_connections.load(Ordering::SeqCst),
+ connections: connections.len()
+ - usize::from(handoff_connection.is_some_and(|id| connections.contains(&id))),
commands: self.commands.len(),
executions: self.executions.active_count(),
shells: self.shells.active_count(),
diff --git a/crates/runtime-host/src/server/dispatch.rs b/crates/runtime-host/src/server/dispatch.rs
index 75e754c8be..93a576ae17 100644
--- a/crates/runtime-host/src/server/dispatch.rs
+++ b/crates/runtime-host/src/server/dispatch.rs
@@ -42,7 +42,10 @@ impl Host {
) -> Result<Outcome, HostError> {
if operation == Operation::HostUpgradePrepare {
return match self
- .prepare_retirement(maka_protocol::host::decode_retirement_input(&input)?)
+ .prepare_retirement(
+ maka_protocol::host::decode_retirement_input(&input)?,
+ connection_id,
+ )
.await
{
Ok(result) => Ok(Outcome::success(serde_json::to_value(result)?)),
diff --git a/crates/runtime-host/src/server/handshake.rs b/crates/runtime-host/src/server/handshake.rs
index cfb650fc71..5506a88ba0 100644
--- a/crates/runtime-host/src/server/handshake.rs
+++ b/crates/runtime-host/src/server/handshake.rs
@@ -17,7 +17,7 @@
* under the License.
*/
-use super::{ConnectionCount, Host, HostError, LifecycleMode, authority::Authority};
+use super::{AcceptedConnection, Host, HostError, LifecycleMode, authority::Authority};
use maka_protocol::{
COMPATIBILITY_EPOCH, COMPOSITION_ID,
handshake::{ClientHello, HostHandshake, Lifecycle, ProtocolRange, Replacement},
@@ -29,7 +29,7 @@ use uuid::Uuid;
type Admission<'a> = (
HostHandshake,
- Option<ConnectionCount<'a>>,
+ Option<AcceptedConnection<'a>>,
Option<DropGuard>,
);
@@ -54,7 +54,7 @@ impl Host {
let generation_mismatch =
ephemeral && hello.generation.is_some() && hello.generation != self.options.generation;
let local_owner = matches!(authority, Authority::LocalOwner);
- let settled = self.accepted_connections.load(Ordering::SeqCst) == 0
+ let settled = self.accepted_connections.lock().unwrap().is_empty()
&& self.requests.is_empty()
&& self.executions.active_count() == 0
&& self.shells.active_count() == 0;
@@ -103,7 +103,10 @@ impl Host {
None,
));
};
- self.accepted_connections.fetch_add(1, Ordering::SeqCst);
+ self.accepted_connections
+ .lock()
+ .unwrap()
+ .insert(connection_id);
Ok((
HostHandshake::Accepted {
root_id: self.root_id().into(),
@@ -116,7 +119,10 @@ impl Host {
state: Lifecycle::Ready,
cooperative_handoff: Some(true),
},
- Some(ConnectionCount(&self.accepted_connections)),
+ Some(AcceptedConnection {
+ connections: &self.accepted_connections,
+ id: connection_id,
+ }),
None,
))
}
@@ -124,7 +130,7 @@ impl Host {
fn activity_snapshot(&self, version: Option<u8>) -> Value {
let activity = self.activity();
let mut result = json!({
- "connections": self.accepted_connections.load(Ordering::SeqCst),
+ "connections": self.accepted_connections.lock().unwrap().len(),
"activeOperations": self.commands.len(),
"processUptimeSeconds": self.started.elapsed().as_secs(),
"residencies": activity.residencies(),
diff --git a/crates/runtime-host/src/server/retirement.rs b/crates/runtime-host/src/server/retirement.rs
index 8c4dbdccd6..1be2d494d6 100644
--- a/crates/runtime-host/src/server/retirement.rs
+++ b/crates/runtime-host/src/server/retirement.rs
@@ -61,7 +61,14 @@ impl Host {
pub(super) async fn prepare_retirement(
&self,
input: RetirementInput,
+ requesting_connection: uuid::Uuid,
) -> Result<RetirementResult, OperationError> {
+ let handoff_connection = input
+ .handoff_connection_id
+ .as_deref()
+ .map(str::parse::<uuid::Uuid>)
+ .transpose()
+ .map_err(|_| failure(Code::OperationConflict, "Invalid handoff connection"))?;
let admission = self.executions.lock_admission().await;
let preparation = {
let mut phase = self.retirement.lock().unwrap_or_else(|e| e.into_inner());
@@ -74,7 +81,16 @@ impl Host {
"Host lifetime changed or retirement already began",
));
}
- let activity = self.activity();
+ if handoff_connection.is_some_and(|id| {
+ id == requesting_connection
+ || !self.accepted_connections.lock().unwrap().contains(&id)
+ }) {
+ return Err(failure(
+ Code::OperationConflict,
+ "Handoff connection is no longer attached",
+ ));
+ }
+ let activity = self.activity_except(handoff_connection);
if input.allow_interrupt_active_tasks || !activity.blocks_retirement(1) {
*phase = Phase::Retiring;
self.draining.cancel();
@@ -90,12 +106,13 @@ impl Host {
}
};
drop(admission);
- self.cooperate(preparation).await
+ self.cooperate(preparation, handoff_connection).await
}
async fn cooperate(
&self,
mut preparation: Preparation<'_>,
+ handoff_connection: Option<uuid::Uuid>,
) -> Result<RetirementResult, OperationError> {
let handoff_id = uuid::Uuid::new_v4().to_string();
let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
@@ -103,7 +120,11 @@ impl Host {
let mut reserved = BTreeSet::new();
loop {
let admission = self.executions.lock_admission().await;
- if self.draining.is_cancelled() || self.activity().blocks_cooperation(1) {
+ if self.draining.is_cancelled()
+ || self
+ .activity_except(handoff_connection)
+ .blocks_cooperation(1)
+ {
return Ok(RetirementResult::ActiveTasks);
}
let Some(runs) = self.executions.cooperative_runs() else {
@@ -142,7 +163,11 @@ impl Host {
// fence. A callback either retains command residency here or
// observes Retiring before it can enter the dispatcher.
let mut phase = self.retirement.lock().unwrap_or_else(|e| e.into_inner());
- if self.draining.is_cancelled() || self.activity().blocks_cooperation(1) {
+ if self.draining.is_cancelled()
+ || self
+ .activity_except(handoff_connection)
+ .blocks_cooperation(1)
+ {
return Ok(RetirementResult::ActiveTasks);
}
// All admitted owners are held or have completed. Commit is
diff --git a/crates/runtime-host/tests/integration/retirement.rs b/crates/runtime-host/tests/integration/retirement.rs
index 39d46c3f59..072018f6fb 100644
--- a/crates/runtime-host/tests/integration/retirement.rs
+++ b/crates/runtime-host/tests/integration/retirement.rs
@@ -70,6 +70,31 @@ async fn receive(frames: &mut mpsc::UnboundedReceiver<Value>) -> Value {
.unwrap()
}
+async fn attach(
+ host: &std::sync::Arc<Host>,
+) -> (
+ mpsc::UnboundedSender<Value>,
+ Value,
+ tokio::task::JoinHandle<Result<(), maka_runtime_host::server::HostError>>,
+) {
+ let (requests, reader) = mpsc::unbounded_channel();
+ let (frames, mut responses) = mpsc::unbounded_channel();
+ let task = tokio::spawn(host.clone().local_owner_connection(
+ Reader(reader),
+ Writer {
+ frames,
+ gate: None,
+ fail: false,
+ },
+ ));
+ requests.send(hello()).unwrap();
+ let handshake = receive(&mut responses).await;
+ assert_eq!(handshake["kind"], "accepted");
+ // Keep the writer open until the request side leaves.
+ tokio::spawn(async move { while responses.recv().await.is_some() {} });
+ (requests, handshake, task)
+}
+
async fn short_connections_reset_idle_expiry(host: &std::sync::Arc<Host>) {
let connect = || async {
let (requests, reader) = mpsc::unbounded_channel();
@@ -164,8 +189,34 @@ async fn retirement_fences_admission_and_keeps_root_until_receipt_flushed_or_aba
requests.send(hello()).unwrap();
let handshake = receive(&mut responses).await;
assert_eq!(handshake["state"], "ready");
- requests.send(json!({"requestId":"retire", "operation":"host.upgrade.prepare",
- "input":{"expectedHostEpoch":handshake["hostEpoch"], "allowInterruptActiveTasks":false}}))
+ let (desktop, desktop_identity, desktop_task) = attach(&host).await;
+ let (unrelated, _, unrelated_task) = attach(&host).await;
+ let mut input = json!({"expectedHostEpoch":handshake["hostEpoch"],
+ "allowInterruptActiveTasks":false, "allowCooperativeHandoff":true,
+ "handoffConnectionId":desktop_identity["connectionId"]});
+ requests
+ .send(json!({"requestId":"blocked", "operation":"host.upgrade.prepare", "input":input}))
+ .unwrap();
+ assert_eq!(
+ receive(&mut responses).await["result"]["kind"],
+ "active_tasks"
+ );
+ drop(unrelated);
+ unrelated_task.await.unwrap().unwrap();
+ input["handoffConnectionId"] = uuid::Uuid::new_v4().to_string().into();
+ requests
+ .send(json!({"requestId":"stale", "operation":"host.upgrade.prepare", "input":input}))
+ .unwrap();
+ assert_eq!(
+ receive(&mut responses).await["error"]["code"],
+ "operation_conflict"
+ );
+ input["handoffConnectionId"] = desktop_identity["connectionId"].clone();
+ requests
+ .send(
+ json!({"requestId":"retire", "operation":"host.upgrade.prepare",
+ "input":input}),
+ )
.unwrap();
tokio::time::timeout(Duration::from_secs(5), entered.cancelled())
.await
@@ -202,6 +253,8 @@ async fn retirement_fences_admission_and_keeps_root_until_receipt_flushed_or_aba
}
drop(requests);
assert_eq!(connection.await.unwrap().is_err(), fail);
+ drop(desktop);
+ desktop_task.await.unwrap().unwrap();
drop(host);
tokio::time::timeout(Duration::from_secs(5), server)
.await
diff --git a/docs/rust-runtime.md b/docs/rust-runtime.md
index ff35d5753c..a3e90d4bd6 100644
--- a/docs/rust-runtime.md
+++ b/docs/rust-runtime.md
@@ -82,6 +82,9 @@ start a service or change account policy.
Desktop reuses managed native Hosts and activates pinned code when needed;
its own generation and exit do not govern their lifetime. Pausing local launches
waits for outstanding activations before handing off the Root.
+Desktop can manage local native deployments and existing SSH/WSL native-operator
+profiles. Remote lifecycle commands use SSH/WSL OS authority, not WebSocket
+credentials. Stopped or uncertain targets stay paused until startup is confirmed.
Deployment authority lives in account-level
SQLite, outside the State Root; startup checks it before database migrations.
On Windows, supervised installations use the sibling `maka-service.exe`, a
diff --git a/docs/rust-runtime.zh-CN.md b/docs/rust-runtime.zh-CN.md
index d50a28d87a..5d416e0be6 100644
--- a/docs/rust-runtime.zh-CN.md
+++ b/docs/rust-runtime.zh-CN.md
@@ -70,6 +70,9 @@ Linux inode 未变并保留 Root ID。不要用它接管复制的根或旧版数
Windows 要求交互式用户会话。单独安装不会启动服务或修改账户策略。
Desktop 复用原生托管 Host，需要时激活固定版本；自身版本变化和退出不控制托管 Host 的寿命。
暂停本地启动后，会等待已发出的激活收尾，再交接 Root。
+Desktop 可管理本机原生部署及已有的 SSH／WSL 原生 operator 配置。远程生命周期命令使用
+SSH／WSL 的操作系统权限，不扩大 WebSocket 凭据权限；停止或结果未确认的目标保持暂停，
+直到启动得到确认。
部署权威保存在 State Root 之外的账户级 SQLite 中，启动时先校验再执行数据库迁移。
Windows 托管服务使用同目录的 `maka-service.exe`，它是同一 Host 的无窗口入口，须与 `maka.exe` 一起分发。
按需激活要求启动环境允许脱离 Windows Job；应直接运行已构建的程序，不要通过 `cargo run` 激活。
diff --git a/tests/fixtures/client-native-managed-entry.mjs b/tests/fixtures/client-native-managed-entry.mjs
index 7006b822ff..6dbc0e1b7b 100644
--- a/tests/fixtures/client-native-managed-entry.mjs
+++ b/tests/fixtures/client-native-managed-entry.mjs
@@ -18,16 +18,15 @@
*/
import assert from 'node:assert/strict';
-import { execFile } from 'node:child_process';
-import { parseArgs, promisify } from 'node:util';
+import { parseArgs } from 'node:util';
import { createNativeRuntimeHostCandidateLaunchBarrier } from '../../apps/desktop/src/main/native-runtime-host.js';
import { readNativeRuntimeHostDeployment } from '../../apps/desktop/src/main/native-runtime-host-deployment.js';
import { createNativeRuntimeHostManagement } from '../../apps/desktop/src/main/native-runtime-host-management.js';
+import { runNativeRuntimeHostCommand } from '../../apps/desktop/src/main/native-runtime-host-command.js';
import { decodeNativeRuntimeHostDeploymentStatus } from '../../apps/desktop/src/shared/native-runtime-host-deployment.js';
import { connectExistingRuntimeHost } from '../../packages/runtime-host/src/client/connection.js';
-import { prepareConnectedRuntimeHostRetirement } from '../../packages/runtime-host/src/client/host-retirement.js';
+import { connectRuntimeHostProfile } from '../../packages/runtime-host/src/client/host-profile.js';
-const run = promisify(execFile);
const { values } = parseArgs({
options: {
'native-managed': { type: 'string' },
@@ -35,11 +34,27 @@ const { values } = parseArgs({
'root-id': { type: 'string' },
revoked: { type: 'boolean', default: false },
management: { type: 'boolean', default: false },
+ ssh: { type: 'string' },
},
});
const executable = values['native-managed'];
assert(executable && values.root && values['root-id']);
-const observed = await readNativeRuntimeHostDeployment(executable, values['root-id']);
+const operator = values.ssh
+ ? {
+ kind: 'ssh',
+ destination: values.ssh,
+ operator: { kind: 'native', platform: 'posix', executablePath: executable },
+ }
+ : { kind: 'local', executable };
+const execute = (args, readOnly = false) => runNativeRuntimeHostCommand(operator, args, readOnly);
+const read = async () =>
+ values.ssh
+ ? decodeNativeRuntimeHostDeploymentStatus(
+ JSON.parse(await execute(['status', '--root-id', values['root-id']], true)),
+ values['root-id'],
+ )
+ : readNativeRuntimeHostDeployment(executable, values['root-id']);
+const observed = await read();
assert.equal(observed.kind, 'installed');
assert.throws(() => decodeNativeRuntimeHostDeploymentStatus(observed, 'f'.repeat(64)));
assert.throws(() =>
@@ -65,23 +80,19 @@ const request = {
closeOnLauncherExit: true,
};
const connections = [];
+let pairing;
async function stop() {
const deadline = Date.now() + 5000;
for (;;) {
- const { stdout } = await run(
- executable,
- [
- 'host',
- 'stop',
- '--root-id',
- values['root-id'],
- '--expected-deployment-id',
- expected.deploymentId,
- '--expected-revision',
- String(expected.configRevision),
- ],
- { windowsHide: true },
- );
+ const stdout = await execute([
+ 'stop',
+ '--root-id',
+ values['root-id'],
+ '--expected-deployment-id',
+ expected.deploymentId,
+ '--expected-revision',
+ String(expected.configRevision),
+ ]);
const outcome = JSON.parse(stdout);
if (outcome.kind === 'stopped') return;
assert.equal(outcome.kind, 'active_tasks');
@@ -141,8 +152,17 @@ try {
} finally {
barrier.release();
for (const connection of connections) await connection.close();
+ if (pairing)
+ await execute([
+ 'access',
+ 'revoke',
+ '--root',
+ values.root,
+ '--credential-id',
+ pairing.credentialId,
+ ]);
if (!values.revoked) {
- const current = await readNativeRuntimeHostDeployment(executable, values['root-id']);
+ const current = await read();
if (current.kind === 'installed' && current.deployment.admission === undefined) {
expected = current.deployment;
await stop();
@@ -154,7 +174,7 @@ async function verifyManagement() {
let owned;
let resumes = 0;
const management = createNativeRuntimeHostManagement({
- executable: observed.deployment.executable,
+ operator: values.ssh ? operator : { kind: 'local', executable: observed.deployment.executable },
rootId: values['root-id'],
rootPath: values.root,
change: async (run) =>
@@ -163,15 +183,11 @@ async function verifyManagement() {
resumeOnSuccess() {
resumes++;
},
- async retire(deployment) {
+ async retire(deployment, hostEpoch, prepareRemote) {
if (!owned) return true;
assert.equal(deployment.deploymentId, expected.deploymentId);
- const retired = await prepareConnectedRuntimeHostRetirement(
- owned,
- 'refuse_active_work',
- 5000,
- );
- if (retired.kind === 'active_tasks') return false;
+ assert.equal(hostEpoch, owned.hostEpoch);
+ if (!(await prepareRemote(owned.connectionId))) return false;
await owned.close();
owned = undefined;
return true;
@@ -181,7 +197,37 @@ async function verifyManagement() {
const started = await management.run({ action: 'start' });
assert.equal(started.outcome.kind, 'ready');
assert.equal(resumes, 1);
+ if (values.ssh)
+ pairing = JSON.parse(
+ await execute([
+ 'access',
+ 'prepare',
+ '--root',
+ values.root,
+ '--principal',
+ 'native-management-test',
+ ]),
+ );
const connect = async () => {
+ if (pairing) {
+ const connection = await connectRuntimeHostProfile({
+ profile: {
+ id: 'native-test',
+ name: 'Native test',
+ kind: 'remote',
+ rootId: values['root-id'],
+ transport: {
+ kind: 'ssh',
+ destination: values.ssh,
+ activation: { kind: 'ssh_operator', operator: operator.operator },
+ },
+ },
+ credential: pairing.credential,
+ clientInstanceId: 'native-management-test',
+ });
+ connections.push(connection);
+ return connection;
+ }
const result = await connectExistingRuntimeHost({
rootPath: values.root,
protocol: request.protocol,
@@ -192,6 +238,18 @@ async function verifyManagement() {
return result.connection;
};
owned = await connect();
+ if (pairing) {
+ await owned.request('access.credential.finalize', {});
+ await owned.close();
+ owned = await connect();
+ await assert.rejects(
+ owned.request('host.upgrade.prepare', {
+ expectedHostEpoch: owned.hostEpoch,
+ allowInterruptActiveTasks: false,
+ }),
+ { code: 'unauthorized' },
+ );
+ }
const other = await connect();
const epoch = owned.hostEpoch;
await assert.rejects(
diff --git a/tests/fixtures/client.mjs b/tests/fixtures/client.mjs
index 794a30487a..642ba7a832 100644
--- a/tests/fixtures/client.mjs
+++ b/tests/fixtures/client.mjs
@@ -54,19 +54,22 @@ await withSourceBundle(fileURLToPath(new URL(entry, import.meta.url)), (bundle,
const child = spawnSync(process.execPath, [bundle, ...process.argv.slice(2)], {
stdio: 'inherit',
// The parent owns cleanup even if a failed fixture retains handles.
- timeout: process.argv.includes('--live-provider-workspace')
- ? 330000
- : process.argv.includes('--oauth-execution-workspace')
- ? 150000
- : process.argv.includes('--message-queue-workspace')
- ? 55000
- : process.argv.includes('--large-output-workspace')
- ? 120000
- : process.argv.includes('--bash-workspace') ||
- process.argv.includes('--onboarding-workspace') ||
- (process.argv.includes('--native-managed') && process.argv.includes('--management'))
- ? 45000
- : 15000,
+ timeout: process.argv.includes('--ssh')
+ ? 180000
+ : process.argv.includes('--live-provider-workspace')
+ ? 330000
+ : process.argv.includes('--oauth-execution-workspace')
+ ? 150000
+ : process.argv.includes('--message-queue-workspace')
+ ? 55000
+ : process.argv.includes('--large-output-workspace')
+ ? 120000
+ : process.argv.includes('--bash-workspace') ||
+ process.argv.includes('--onboarding-workspace') ||
+ (process.argv.includes('--native-managed') &&
+ process.argv.includes('--management'))
+ ? 45000
+ : 15000,
});
if (child.error) throw child.error;
if (child.signal) throw new Error(`Original-client subprocess terminated by ${child.signal}`);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org