Control remote fleet workers over SSH with user, host, and key path

NS1 is marchon@70.88.205.138 using ~/.ssh/id_ed25519. Private key
bytes stay off git. Spawn, health, and kill go through ssh; remote
workers bind loopback only.

packages/verae-fleet/src/cli.js

@@ -37,7 +37,8 @@ function help() {
 Central spec:  fleet.json
 Machines:      machines.json  (add hosts to spread replicas)
 Per service:   services/<id>.json
-Remote host:   FLEET_AGENT_PORT=3851 node src/agent.js
+Remote SSH:    machines.json user + host + identityFile (path only)
+  ssh-check [id]               test SSH login (default ns1)
 `);
 }
 
@@ -74,6 +75,19 @@ async function main() {
     }
     print(`\ncentral replica spec: ${loaded.fleetPath}`);
     print(`tree-node floor:      min=${loaded.services['tree-node'].min} keepFloor=${loaded.services['tree-node'].keepFloor}`);
+    print('\nmachines:');
+    for (const m of loaded.machines) {
+      const ssh = m.kind === 'ssh' ? `${m.user}@${m.host} key=${m.identityFile}` : m.host;
+      print(`  ${m.id.padEnd(12)} ${m.kind.padEnd(6)} ${m.enabled ? 'on ' : 'off'} cap=${m.capacity} ${ssh}`);
+    }
+    return;
+  }
+  if (cmd === 'ssh-check') {
+    const { sshCheck } = await import('./ssh.js');
+    const id = target || 'ns1';
+    const m = loaded.machines.find((x) => x.id === id);
+    if (!m) throw new Error(`unknown machine ${id}`);
+    print(await sshCheck(m));
     return;
   }
   if (cmd === 'serve' || cmd === 'monitor') {

packages/verae-fleet/src/machines.js

@@ -32,15 +32,29 @@ export function loadMachines(root, overlayPath) {
     const incoming = Array.isArray(extra) ? extra : extra.machines || [];
     for (const m of incoming) upsertMachine(list, m);
   }
+  const secretsPath = path.join(root, 'machines.secrets.json');
+  if (fs.existsSync(secretsPath)) {
+    const secrets = JSON.parse(fs.readFileSync(secretsPath, 'utf8'));
+    const byId = secrets.machines || secrets;
+    for (const m of list) {
+      const extra = byId[m.id];
+      if (extra && typeof extra === 'object') Object.assign(m, extra);
+    }
+  }
   return list.map(normalizeMachine);
 }
 
 export function normalizeMachine(m) {
+  const kind = m.kind === 'agent' || m.kind === 'ssh' ? m.kind : 'local';
   return {
     id: String(m.id || '').trim(),
     title: m.title || m.id,
-    kind: m.kind === 'agent' ? 'agent' : 'local',
+    kind,
     host: m.host || '127.0.0.1',
+    user: m.user || (kind === 'ssh' ? 'marchon' : ''),
+    sshPort: Number(m.sshPort || 22),
+    identityFile: m.identityFile || (kind === 'ssh' ? '~/.ssh/id_ed25519' : ''),
+    remoteDir: m.remoteDir || '~/verae-fleet-runtime',
     agentPort: Number(m.agentPort || 3851),
     enabled: m.enabled !== false,
     capacity: Number(m.capacity || 8),
@@ -72,8 +86,9 @@ export function canHost(machine, serviceId, role) {
  * @param {string} serviceId
  * @param {string} [role]
  */
-export function pickMachine(machines, instances, serviceId, role) {
-  const eligible = machines.filter((m) => canHost(m, serviceId, role));
+export function pickMachine(machines, instances, serviceId, role, exclude = []) {
+  const skip = new Set(exclude);
+  const eligible = machines.filter((m) => canHost(m, serviceId, role) && !skip.has(m.id));
   if (!eligible.length) return null;
   const scored = eligible.map((m) => {
     const running = instances.filter((i) => i.machine === m.id && i.pid).length;

packages/verae-fleet/src/server.js

@@ -47,9 +47,13 @@ export function startControlServer(sup, mon) {
         const m = sup.addMachine(spec);
         return json(res, 200, { machine: m, machines: sup.status().machines });
       }
-      const mach = url.pathname.match(/^\/api\/machines\/([^/]+)\/(enable|disable)$/);
+      const mach = url.pathname.match(/^\/api\/machines\/([^/]+)\/(enable|disable|check)$/);
       if (mach && req.method === 'POST') {
         const [, id, op] = mach;
+        if (op === 'check') {
+          const check = await sup.checkMachine(id);
+          return json(res, check.ok ? 200 : 400, { check, machines: sup.status().machines });
+        }
         const m = sup.setMachineEnabled(id, op === 'enable');
         return json(res, 200, { machine: m, machines: sup.status().machines });
       }

packages/verae-fleet/src/ssh.js

@@ -0,0 +1,168 @@
+/**
+ * SSH control of remote fleet workers. Uses username + host + identity file path.
+ * Never logs or stores private-key material.
+ * @module ssh
+ */
+
+import { spawn } from 'node:child_process';
+import fs from 'node:fs';
+import os from 'node:os';
+import path from 'node:path';
+import { fileURLToPath } from 'node:url';
+
+const HERE = path.dirname(fileURLToPath(import.meta.url));
+
+export function expandHome(p) {
+  if (!p) return p;
+  if (p === '~') return os.homedir();
+  if (p.startsWith('~/')) return path.join(os.homedir(), p.slice(2));
+  return p;
+}
+
+export function shellQuote(s) {
+  return `'${String(s).replace(/'/g, `'\"'\"'`)}'`;
+}
+
+export function sshTarget(machine) {
+  return `${machine.user || 'marchon'}@${machine.host}`;
+}
+
+export function sshBaseArgs(machine) {
+  const ident = expandHome(machine.identityFile || '');
+  const args = [
+    '-o', 'BatchMode=yes',
+    '-o', 'StrictHostKeyChecking=accept-new',
+    '-o', 'IdentitiesOnly=yes',
+    '-o', 'ConnectTimeout=8',
+    '-p', String(machine.sshPort || 22),
+  ];
+  if (ident) {
+    if (!fs.existsSync(ident)) throw new Error(`identityFile not found: ${ident}`);
+    args.push('-i', ident);
+  }
+  return args;
+}
+
+export function scpBaseArgs(machine) {
+  const ident = expandHome(machine.identityFile || '');
+  const args = [
+    '-o', 'BatchMode=yes',
+    '-o', 'StrictHostKeyChecking=accept-new',
+    '-o', 'IdentitiesOnly=yes',
+    '-o', 'ConnectTimeout=8',
+    '-P', String(machine.sshPort || 22),
+  ];
+  if (ident) args.push('-i', ident);
+  return args;
+}
+
+export function sshExec(machine, remoteCommand, { timeoutMs = 20000 } = {}) {
+  return new Promise((resolve, reject) => {
+    const args = [...sshBaseArgs(machine), sshTarget(machine), remoteCommand];
+    const child = spawn('ssh', args, { stdio: ['ignore', 'pipe', 'pipe'] });
+    const out = [];
+    const err = [];
+    child.stdout.on('data', (c) => out.push(c));
+    child.stderr.on('data', (c) => err.push(c));
+    const t = setTimeout(() => {
+      child.kill('SIGKILL');
+      reject(new Error(`ssh timeout: ${machine.id || machine.host}`));
+    }, timeoutMs);
+    child.on('error', (e) => {
+      clearTimeout(t);
+      reject(e);
+    });
+    child.on('close', (code) => {
+      clearTimeout(t);
+      const stdout = Buffer.concat(out).toString('utf8').trim();
+      const stderr = Buffer.concat(err).toString('utf8').trim();
+      if (code !== 0) {
+        reject(new Error(stderr || stdout || `ssh exit ${code}`));
+        return;
+      }
+      resolve({ stdout, stderr, code });
+    });
+  });
+}
+
+export function scpTo(machine, localFiles, remoteDir) {
+  return new Promise((resolve, reject) => {
+    const dest = `${sshTarget(machine)}:${remoteDir}/`;
+    const args = [...scpBaseArgs(machine), ...localFiles, dest];
+    const child = spawn('scp', args, { stdio: ['ignore', 'pipe', 'pipe'] });
+    const err = [];
+    child.stderr.on('data', (c) => err.push(c));
+    const t = setTimeout(() => {
+      child.kill('SIGKILL');
+      reject(new Error('scp timeout'));
+    }, 30000);
+    child.on('close', (code) => {
+      clearTimeout(t);
+      if (code !== 0) reject(new Error(Buffer.concat(err).toString('utf8') || `scp exit ${code}`));
+      else resolve();
+    });
+    child.on('error', (e) => {
+      clearTimeout(t);
+      reject(e);
+    });
+  });
+}
+
+const synced = new Set();
+
+export async function ensureRemoteRuntime(machine) {
+  const remoteDir = machine.remoteDir || '~/verae-fleet-runtime';
+  const key = `${machine.id}:${remoteDir}`;
+  if (synced.has(key)) return remoteDir;
+  await sshExec(machine, `mkdir -p ${remoteDir}/src ${remoteDir}/data`);
+  await scpTo(machine, [path.join(HERE, 'worker.js'), path.join(HERE, 'rtt.js')], `${remoteDir}/src`);
+  synced.add(key);
+  return remoteDir;
+}
+
+export async function sshSpawnWorker(machine, { instance, env }) {
+  const remoteDir = await ensureRemoteRuntime(machine);
+  const fleetEnv = Object.entries(env)
+    .filter(([k]) => k.startsWith('FLEET_'))
+    .map(([k, v]) => `${k}=${shellQuote(v)}`)
+    .join(' ');
+  const cmd = `mkdir -p ${remoteDir}/data/${instance} && cd ${remoteDir} && ${fleetEnv} nohup node src/worker.js >data/${instance}/worker.log 2>&1 & echo $!`;
+  const { stdout } = await sshExec(machine, cmd);
+  const pid = Number(String(stdout).split('\n').pop());
+  if (!Number.isFinite(pid)) throw new Error(`remote spawn produced no pid: ${stdout}`);
+  return { pid, remoteDir };
+}
+
+export async function sshKillWorker(machine, pid, instance) {
+  const cmd = pid
+    ? `kill -TERM ${Number(pid)} 2>/dev/null || true`
+    : `pkill -f ${shellQuote(`FLEET_INSTANCE=${instance}`)} 2>/dev/null || true`;
+  await sshExec(machine, cmd);
+}
+
+export async function sshHttp(machine, method, urlPath, { port, body } = {}) {
+  const url = `http://127.0.0.1:${port}${urlPath}`;
+  let cmd = `curl -sS -m 3 -X ${method}`;
+  if (body) {
+    cmd += ` -H 'content-type: application/json' -d ${shellQuote(JSON.stringify(body))}`;
+  }
+  cmd += ` ${shellQuote(url)}`;
+  const { stdout } = await sshExec(machine, cmd, { timeoutMs: 8000 });
+  try {
+    return JSON.parse(stdout || '{}');
+  } catch {
+    return { raw: stdout };
+  }
+}
+
+export async function sshCheck(machine) {
+  const { stdout } = await sshExec(machine, 'echo OK; whoami; hostname; command -v node; node -v');
+  const lines = stdout.split('\n');
+  return {
+    ok: lines[0] === 'OK',
+    user: lines[1],
+    hostname: lines[2],
+    node: lines[3] ? `${lines[3]} ${lines[4] || ''}`.trim() : null,
+    identityFile: expandHome(machine.identityFile || ''),
+  };
+}

packages/verae-fleet/src/supervisor.js

@@ -12,6 +12,7 @@ import { httpProbe, postControl } from './health.js';
 import { classifyService } from './classify.js';
 import { pickMachine, upsertMachine, saveOverlay } from './machines.js';
 import { summarizeRtt } from './rtt.js';
+import { sshSpawnWorker, sshKillWorker, sshHttp, sshCheck } from './ssh.js';
 
 const WORKER = path.join(path.dirname(fileURLToPath(import.meta.url)), 'worker.js');
 
@@ -55,6 +56,35 @@ export class Supervisor {
     return m;
   }
 
+  async checkMachine(id) {
+    const m = this.machines().find((x) => x.id === id);
+    if (!m) throw new Error(`unknown machine ${id}`);
+    if (m.kind === 'ssh') return sshCheck(m);
+    if (m.kind === 'agent') {
+      const r = await httpProbe({ host: m.host, port: m.agentPort, path: '/health', timeoutMs: 2000 });
+      return { ok: r.ok, kind: 'agent', body: r.body };
+    }
+    return { ok: true, kind: 'local', host: m.host };
+  }
+
+  machineOf(rec) {
+    return this.machines().find((x) => x.id === rec.machine);
+  }
+
+  async controlInstance(rec, pathname, body) {
+    if (rec.kind === 'ssh') {
+      const m = this.machineOf(rec);
+      if (!m) throw new Error(`machine ${rec.machine} missing`);
+      return sshHttp(m, 'POST', pathname, { port: rec.healthPort, body });
+    }
+    return postControl({
+      host: rec.host || '127.0.0.1',
+      port: rec.healthPort,
+      path: pathname,
+      body,
+    });
+  }
+
   log(type, detail) {
     const ev = { t: new Date().toISOString(), type, ...detail };
     this.events.push(ev);
@@ -100,7 +130,13 @@ export class Supervisor {
     if (this.instances.has(instance)) return this.instances.get(instance);
     const machine =
       opts.machine ||
-      pickMachine(this.machines(), [...this.instances.values()], serviceId, spec.role || serviceId);
+      pickMachine(
+        this.machines(),
+        [...this.instances.values()],
+        serviceId,
+        spec.role || serviceId,
+        opts.excludeMachines || [],
+      );
     if (!machine) {
       this.log('skip', { service: serviceId, reason: 'no-machine-capacity' });
       return null;
@@ -115,13 +151,25 @@ export class Supervisor {
       FLEET_INSTANCE: instance,
       FLEET_MACHINE: machine.id,
       FLEET_HEALTH_PORT: String(healthPort),
-      FLEET_HEALTH_BIND: machine.kind === 'local' ? '127.0.0.1' : '0.0.0.0',
-      FLEET_STATE_DIR: stateDir,
+      FLEET_HEALTH_BIND: machine.kind === 'local' ? '127.0.0.1' : '127.0.0.1',
+      FLEET_STATE_DIR: machine.kind === 'ssh' ? `${machine.remoteDir || '~/verae-fleet-runtime'}/data/${instance}` : stateDir,
       FLEET_MESSAGE_HZ: process.env.FLEET_MESSAGE_HZ || '8',
     };
     let child = null;
     let pid = null;
-    if (machine.kind === 'agent') {
+    if (machine.kind === 'ssh') {
+      try {
+        const spawned = await sshSpawnWorker(machine, { instance, env });
+        pid = spawned.pid;
+      } catch (err) {
+        this.log('skip', { service: serviceId, reason: 'ssh-spawn-failed', machine: machine.id, error: err.message });
+        const exclude = [...(opts.excludeMachines || []), machine.id];
+        if (exclude.length < this.machines().length) {
+          return this.startOne(serviceId, { ...opts, excludeMachines: exclude });
+        }
+        return null;
+      }
+    } else if (machine.kind === 'agent') {
       try {
         const spawned = await postControl({
           host: machine.host,
@@ -150,6 +198,7 @@ export class Supervisor {
       stateDir,
       child,
       kind: machine.kind,
+      _machine: machine,
       paused: false,
       startedAt: Date.now(),
       restarts: 0,
@@ -167,7 +216,7 @@ export class Supervisor {
     }
     this.instances.set(instance, rec);
     this.log('start', { instance, service: serviceId, machine: machine.id, pid: rec.pid, healthPort });
-    const h = await waitForHealth(rec.host, healthPort, 4000);
+    const h = await waitForHealth(rec, 8000);
     rec.healthy = h.ok;
     rec.lastProbe = h;
     return rec;
@@ -177,7 +226,16 @@ export class Supervisor {
     const rec = this.instances.get(instanceId);
     if (!rec) return null;
     rec.stopping = true;
-    if (rec.kind === 'agent') {
+    if (rec.kind === 'ssh') {
+      const m = this.machines().find((x) => x.id === rec.machine);
+      if (m) {
+        try {
+          await sshKillWorker(m, rec.pid, instanceId);
+        } catch {
+          /* ignore */
+        }
+      }
+    } else if (rec.kind === 'agent') {
       const m = this.machines().find((x) => x.id === rec.machine);
       if (m) {
         try {
@@ -199,7 +257,7 @@ export class Supervisor {
   async pauseInstance(instanceId, { replace = true } = {}) {
     const rec = this.instances.get(instanceId);
     if (!rec) throw new Error(`no instance ${instanceId}`);
-    await postControl({ host: rec.host || '127.0.0.1', port: rec.healthPort, path: '/pause' });
+    await this.controlInstance(rec, '/pause');
     rec.paused = true;
     this.log('pause', { instance: instanceId, service: rec.service });
     if (replace) await this.reconcile(rec.service);
@@ -209,7 +267,7 @@ export class Supervisor {
   async resumeInstance(instanceId) {
     const rec = this.instances.get(instanceId);
     if (!rec) throw new Error(`no instance ${instanceId}`);
-    await postControl({ host: rec.host || '127.0.0.1', port: rec.healthPort, path: '/resume' });
+    await this.controlInstance(rec, '/resume');
     rec.paused = false;
     this.log('resume', { instance: instanceId, service: rec.service });
     return rec;
@@ -229,25 +287,36 @@ export class Supervisor {
   async markUnhealthy(instanceId) {
     const rec = this.instances.get(instanceId);
     if (!rec) throw new Error(`no instance ${instanceId}`);
-    await postControl({ host: rec.host || '127.0.0.1', port: rec.healthPort, path: '/unhealthy' });
+    await this.controlInstance(rec, '/unhealthy');
     return rec;
   }
 
   async probe(rec) {
     const spec = this.spec(rec.service);
-    const host = rec.host || '127.0.0.1';
-    const r = await httpProbe({
-      host,
-      port: rec.healthPort,
-      path: spec.health?.path || '/health',
-      timeoutMs: spec.health?.timeoutMs || 800,
-    });
+    const pathName = spec.health?.path || '/health';
+    let r;
+    if (rec.kind === 'ssh') {
+      const m = this.machineOf(rec);
+      try {
+        const body = await sshHttp(m, 'GET', pathName, { port: rec.healthPort });
+        r = { ok: body.ok !== false, paused: Boolean(body.paused), body, statusCode: body.ok === false ? 503 : 200 };
+      } catch (err) {
+        r = { ok: false, paused: false, error: err.message };
+      }
+    } else {
+      r = await httpProbe({
+        host: rec.host || '127.0.0.1',
+        port: rec.healthPort,
+        path: pathName,
+        timeoutMs: spec.health?.timeoutMs || 800,
+      });
+    }
     rec.paused = Boolean(r.paused);
     rec.lastProbe = r;
     rec.healthy = r.ok;
     if (r.ok && !rec.paused) {
       try {
-        const msg = await postControl({ host, port: rec.healthPort, path: '/message' });
+        const msg = await this.controlInstance(rec, '/message');
         if (Number.isFinite(msg.rttMs)) {
           rec.rttSamples = rec.rttSamples || [];
           rec.rttSamples.push(msg.rttMs);
@@ -415,12 +484,27 @@ function waitExit(child, ms) {
   });
 }
 
-async function waitForHealth(host, port, ms) {
+async function waitForHealth(rec, ms) {
   const deadline = Date.now() + ms;
   while (Date.now() < deadline) {
-    const r = await httpProbe({ host: host || '127.0.0.1', port, path: '/health', timeoutMs: 300 });
+    let r;
+    if (rec.kind === 'ssh' && rec._machine) {
+      try {
+        const body = await sshHttp(rec._machine, 'GET', '/health', { port: rec.healthPort });
+        r = { ok: body.ok !== false, statusCode: 200, paused: Boolean(body.paused), body };
+      } catch {
+        r = { ok: false };
+      }
+    } else {
+      r = await httpProbe({
+        host: rec.host || '127.0.0.1',
+        port: rec.healthPort,
+        path: '/health',
+        timeoutMs: 400,
+      });
+    }
     if (r.ok || r.statusCode === 503) return r;
-    await new Promise((r) => setTimeout(r, 50));
+    await new Promise((res) => setTimeout(res, 80));
   }
   return { ok: false, error: 'start-timeout' };
 }