Spread fleet replicas across extra machines and show message RTT percentiles

machines.json plus Add-machine UI place copies on the least-loaded host.
Monitor samples processing RTT and displays min, avg, p50, and p90 for
capacity planning. Remote hosts run src/agent.js.

packages/verae-fleet/src/agent.js

@@ -0,0 +1,86 @@
+#!/usr/bin/env node
+/**
+ * Remote host agent: spawn/kill fleet workers so the control plane can spread replicas.
+ *
+ *   FLEET_AGENT_PORT=3851 node src/agent.js
+ *
+ * Bind loopback or a private interface. Do not publish NATS.
+ */
+import http from 'node:http';
+import { spawn } from 'node:child_process';
+import path from 'node:path';
+import { fileURLToPath } from 'node:url';
+
+const WORKER = path.join(path.dirname(fileURLToPath(import.meta.url)), 'worker.js');
+const PORT = Number(process.env.FLEET_AGENT_PORT || 3851);
+const BIND = process.env.FLEET_AGENT_BIND || '127.0.0.1';
+const children = new Map();
+
+function json(res, code, obj) {
+  res.writeHead(code, { 'content-type': 'application/json' });
+  res.end(JSON.stringify(obj));
+}
+
+async function readBody(req) {
+  const chunks = [];
+  for await (const c of req) chunks.push(c);
+  const raw = Buffer.concat(chunks).toString('utf8');
+  return raw ? JSON.parse(raw) : {};
+}
+
+export function startAgent({ port = PORT, bind = BIND } = {}) {
+  const server = http.createServer(async (req, res) => {
+    const url = new URL(req.url, `http://${bind}:${port}`);
+    try {
+      if (req.method === 'GET' && url.pathname === '/health') {
+        return json(res, 200, {
+          ok: true,
+          role: 'fleet-agent',
+          workers: [...children.keys()],
+        });
+      }
+      if (req.method === 'POST' && url.pathname === '/spawn') {
+        const body = await readBody(req);
+        const instance = body.instance;
+        if (!instance) return json(res, 400, { error: 'instance required' });
+        if (children.has(instance)) {
+          return json(res, 200, { pid: children.get(instance).pid, instance, existing: true });
+        }
+        const child = spawn(process.execPath, [WORKER], {
+          env: { ...process.env, ...(body.env || {}) },
+          stdio: ['ignore', 'pipe', 'pipe'],
+        });
+        children.set(instance, child);
+        child.on('exit', () => children.delete(instance));
+        return json(res, 200, { pid: child.pid, instance });
+      }
+      if (req.method === 'POST' && url.pathname === '/kill') {
+        const body = await readBody(req);
+        const child = children.get(body.instance);
+        if (child) {
+          child.kill('SIGTERM');
+          children.delete(body.instance);
+        }
+        return json(res, 200, { killed: Boolean(child), instance: body.instance });
+      }
+      json(res, 404, { error: 'not found' });
+    } catch (err) {
+      json(res, 400, { error: err.message });
+    }
+  });
+  server.listen(port, bind, () => {
+    process.stdout.write(`fleet-agent http://${bind}:${port}/\n`);
+  });
+  return {
+    server,
+    async close() {
+      for (const c of children.values()) c.kill('SIGTERM');
+      children.clear();
+      await new Promise((r) => server.close(r));
+    },
+  };
+}
+
+if (process.argv[1] && path.basename(process.argv[1]) === 'agent.js') {
+  startAgent();
+}

packages/verae-fleet/src/cli.js

@@ -35,7 +35,9 @@ function help() {
   reconcile                    force floor check
 
 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
 `);
 }
 

packages/verae-fleet/src/health.js

@@ -8,10 +8,21 @@ import net from 'node:net';
 import fs from 'node:fs';
 import path from 'node:path';
 
-export function httpProbe(port, pathname = '/health', timeoutMs = 800) {
+/**
+ * @param {number|{ host?: string, port: number, path?: string, timeoutMs?: number }} portOrOpts
+ */
+export function httpProbe(portOrOpts, pathname = '/health', timeoutMs = 800) {
+  const opts =
+    typeof portOrOpts === 'object'
+      ? portOrOpts
+      : { port: portOrOpts, path: pathname, timeoutMs };
+  const host = opts.host || '127.0.0.1';
+  const port = opts.port;
+  const pathName = opts.path || pathname;
+  const tmo = opts.timeoutMs || timeoutMs;
   return new Promise((resolve) => {
     const req = http.get(
-      { host: '127.0.0.1', port, path: pathname, timeout: timeoutMs },
+      { host, port, path: pathName, timeout: tmo },
       (res) => {
         const chunks = [];
         res.on('data', (c) => chunks.push(c));
@@ -57,10 +68,26 @@ export function tcpProbe(host, port, timeoutMs = 400) {
   });
 }
 
-export function postControl(port, pathname) {
+/**
+ * @param {number|{ host?: string, port: number, path?: string, body?: object }} portOrOpts
+ * @param {string} [pathname]
+ */
+export function postControl(portOrOpts, pathname = '/') {
+  const opts = typeof portOrOpts === 'object' ? portOrOpts : { port: portOrOpts, path: pathname };
+  const host = opts.host || '127.0.0.1';
+  const port = opts.port;
+  const pathName = opts.path || pathname;
+  const body = opts.body ? JSON.stringify(opts.body) : '';
   return new Promise((resolve, reject) => {
     const req = http.request(
-      { host: '127.0.0.1', port, path: pathname, method: 'POST', timeout: 800 },
+      {
+        host,
+        port,
+        path: pathName,
+        method: 'POST',
+        timeout: opts.timeoutMs || 2000,
+        headers: body ? { 'content-type': 'application/json', 'content-length': Buffer.byteLength(body) } : {},
+      },
       (res) => {
         const chunks = [];
         res.on('data', (c) => chunks.push(c));
@@ -74,6 +101,7 @@ export function postControl(port, pathname) {
       },
     );
     req.on('error', reject);
+    if (body) req.write(body);
     req.end();
   });
 }

packages/verae-fleet/src/load.js

@@ -6,6 +6,7 @@
 import fs from 'node:fs';
 import path from 'node:path';
 import { fileURLToPath } from 'node:url';
+import { loadMachines } from './machines.js';
 
 const HERE = path.dirname(fileURLToPath(import.meta.url));
 export const FLEET_ROOT = path.resolve(HERE, '..');
@@ -17,7 +18,7 @@ export function readJson(file) {
 /**
  * @param {string} [root]
  */
-export function loadFleet(root = FLEET_ROOT) {
+export function loadFleet(root = FLEET_ROOT, opts = {}) {
   const fleetPath = path.join(root, 'fleet.json');
   const fleet = readJson(fleetPath);
   const dir = path.join(root, 'services');
@@ -44,11 +45,17 @@ export function loadFleet(root = FLEET_ROOT) {
       managed: central.managed === false ? false : spec.managed !== false,
     };
   }
+  const overlay =
+    opts.overlay === false ? null : path.join(root, 'data', 'runtime', 'machines-overlay.json');
+  const machines = loadMachines(root, overlay);
   return {
     root,
     fleetPath: path.relative(root, fleetPath) || 'fleet.json',
+    machinesPath: 'machines.json',
+    machinesOverlay: overlay,
     control: fleet.control,
     nats: fleet.nats,
+    machines,
     services,
     unmanaged,
   };

packages/verae-fleet/src/machines.js

@@ -0,0 +1,92 @@
+/**
+ * Host catalog: where replicas may run. Least-loaded placement.
+ * @module machines
+ */
+
+import fs from 'node:fs';
+import path from 'node:path';
+
+export function defaultMachines() {
+  return [
+    {
+      id: 'local',
+      title: 'Control plane (this host)',
+      kind: 'local',
+      host: '127.0.0.1',
+      enabled: true,
+      capacity: 32,
+      roles: ['*'],
+    },
+  ];
+}
+
+export function loadMachines(root, overlayPath) {
+  const file = path.join(root, 'machines.json');
+  let list = defaultMachines();
+  if (fs.existsSync(file)) {
+    const raw = JSON.parse(fs.readFileSync(file, 'utf8'));
+    if (Array.isArray(raw.machines) && raw.machines.length) list = raw.machines;
+  }
+  if (overlayPath && fs.existsSync(overlayPath)) {
+    const extra = JSON.parse(fs.readFileSync(overlayPath, 'utf8'));
+    const incoming = Array.isArray(extra) ? extra : extra.machines || [];
+    for (const m of incoming) upsertMachine(list, m);
+  }
+  return list.map(normalizeMachine);
+}
+
+export function normalizeMachine(m) {
+  return {
+    id: String(m.id || '').trim(),
+    title: m.title || m.id,
+    kind: m.kind === 'agent' ? 'agent' : 'local',
+    host: m.host || '127.0.0.1',
+    agentPort: Number(m.agentPort || 3851),
+    enabled: m.enabled !== false,
+    capacity: Number(m.capacity || 8),
+    roles: Array.isArray(m.roles) && m.roles.length ? m.roles : ['*'],
+    notes: m.notes || '',
+  };
+}
+
+export function upsertMachine(list, spec) {
+  const m = normalizeMachine(spec);
+  if (!m.id) throw new Error('machine id required');
+  const i = list.findIndex((x) => x.id === m.id);
+  if (i >= 0) list[i] = { ...list[i], ...m };
+  else list.push(m);
+  return m;
+}
+
+export function canHost(machine, serviceId, role) {
+  if (!machine?.enabled) return false;
+  const roles = machine.roles || ['*'];
+  if (roles.includes('*')) return true;
+  return roles.includes(serviceId) || (role && roles.includes(role));
+}
+
+/**
+ * Pick the enabled machine with the most free capacity (fewest running / capacity).
+ * @param {object[]} machines
+ * @param {Array<{ machine?: string }>} instances
+ * @param {string} serviceId
+ * @param {string} [role]
+ */
+export function pickMachine(machines, instances, serviceId, role) {
+  const eligible = machines.filter((m) => canHost(m, serviceId, role));
+  if (!eligible.length) return null;
+  const scored = eligible.map((m) => {
+    const running = instances.filter((i) => i.machine === m.id && i.pid).length;
+    const free = m.capacity - running;
+    const load = m.capacity <= 0 ? 1 : running / m.capacity;
+    return { m, running, free, load };
+  });
+  scored.sort((a, b) => a.load - b.load || a.running - b.running || a.m.id.localeCompare(b.m.id));
+  const best = scored.find((s) => s.free > 0);
+  return best ? best.m : null;
+}
+
+export function saveOverlay(overlayPath, machines) {
+  fs.mkdirSync(path.dirname(overlayPath), { recursive: true });
+  fs.writeFileSync(overlayPath, JSON.stringify({ machines }, null, 2));
+}

packages/verae-fleet/src/rtt.js

@@ -0,0 +1,57 @@
+/**
+ * Ring buffer of message-processing RTTs (ms) for planning percentiles.
+ * @module rtt
+ */
+
+export class RttWindow {
+  /**
+   * @param {number} [cap=400]
+   */
+  constructor(cap = 400) {
+    this.cap = cap;
+    /** @type {number[]} */
+    this.samples = [];
+  }
+
+  add(ms) {
+    const n = Number(ms);
+    if (!Number.isFinite(n) || n < 0) return;
+    this.samples.push(n);
+    if (this.samples.length > this.cap) this.samples.splice(0, this.samples.length - this.cap);
+  }
+
+  stats() {
+    return summarizeRtt(this.samples);
+  }
+}
+
+/**
+ * @param {number[]} samples
+ */
+export function summarizeRtt(samples) {
+  const xs = samples.filter((n) => Number.isFinite(n) && n >= 0).slice().sort((a, b) => a - b);
+  if (!xs.length) {
+    return { count: 0, minMs: null, avgMs: null, p50Ms: null, p90Ms: null };
+  }
+  const sum = xs.reduce((a, b) => a + b, 0);
+  return {
+    count: xs.length,
+    minMs: round1(xs[0]),
+    avgMs: round1(sum / xs.length),
+    p50Ms: round1(percentile(xs, 0.5)),
+    p90Ms: round1(percentile(xs, 0.9)),
+  };
+}
+
+function percentile(sorted, p) {
+  if (sorted.length === 1) return sorted[0];
+  const idx = (sorted.length - 1) * p;
+  const lo = Math.floor(idx);
+  const hi = Math.ceil(idx);
+  if (lo === hi) return sorted[lo];
+  return sorted[lo] + (sorted[hi] - sorted[lo]) * (idx - lo);
+}
+
+function round1(n) {
+  return Math.round(n * 10) / 10;
+}

packages/verae-fleet/src/server.js

@@ -37,6 +37,22 @@ export function startControlServer(sup, mon) {
       if (req.method === 'GET' && url.pathname === '/api/status') {
         return json(res, 200, { ...sup.status(), monitor: mon?.last || null });
       }
+      if (req.method === 'GET' && url.pathname === '/api/machines') {
+        return json(res, 200, { machines: sup.status().machines });
+      }
+      if (req.method === 'POST' && url.pathname === '/api/machines') {
+        const chunks = [];
+        for await (const c of req) chunks.push(c);
+        const spec = JSON.parse(Buffer.concat(chunks).toString('utf8') || '{}');
+        const m = sup.addMachine(spec);
+        return json(res, 200, { machine: m, machines: sup.status().machines });
+      }
+      const mach = url.pathname.match(/^\/api\/machines\/([^/]+)\/(enable|disable)$/);
+      if (mach && req.method === 'POST') {
+        const [, id, op] = mach;
+        const m = sup.setMachineEnabled(id, op === 'enable');
+        return json(res, 200, { machine: m, machines: sup.status().machines });
+      }
       if (req.method === 'POST' && url.pathname === '/api/reconcile') {
         const actions = await sup.reconcile();
         return json(res, 200, { actions, status: sup.status() });

packages/verae-fleet/src/supervisor.js

@@ -10,6 +10,8 @@ import { fileURLToPath } from 'node:url';
 import { loadFleet } from './load.js';
 import { httpProbe, postControl } from './health.js';
 import { classifyService } from './classify.js';
+import { pickMachine, upsertMachine, saveOverlay } from './machines.js';
+import { summarizeRtt } from './rtt.js';
 
 const WORKER = path.join(path.dirname(fileURLToPath(import.meta.url)), 'worker.js');
 
@@ -27,6 +29,30 @@ export class Supervisor {
     this.instances = new Map();
     this.events = [];
     this._seq = 0;
+    if (!Array.isArray(this.loaded.machines) || !this.loaded.machines.length) {
+      this.loaded.machines = [
+        { id: 'local', title: 'local', kind: 'local', host: '127.0.0.1', enabled: true, capacity: 32, roles: ['*'] },
+      ];
+    }
+  }
+
+  machines() {
+    return this.loaded.machines;
+  }
+
+  addMachine(spec) {
+    const m = upsertMachine(this.loaded.machines, spec);
+    if (this.loaded.machinesOverlay) saveOverlay(this.loaded.machinesOverlay, this.loaded.machines);
+    this.log('machine-add', { id: m.id, host: m.host, kind: m.kind });
+    return m;
+  }
+
+  setMachineEnabled(id, enabled) {
+    const m = this.loaded.machines.find((x) => x.id === id);
+    if (!m) throw new Error(`unknown machine ${id}`);
+    m.enabled = Boolean(enabled);
+    if (this.loaded.machinesOverlay) saveOverlay(this.loaded.machinesOverlay, this.loaded.machines);
+    return m;
   }
 
   log(type, detail) {
@@ -72,43 +98,76 @@ export class Supervisor {
     }
     const instance = `${serviceId}-${index}`;
     if (this.instances.has(instance)) return this.instances.get(instance);
+    const machine =
+      opts.machine ||
+      pickMachine(this.machines(), [...this.instances.values()], serviceId, spec.role || serviceId);
+    if (!machine) {
+      this.log('skip', { service: serviceId, reason: 'no-machine-capacity' });
+      return null;
+    }
     const healthPort = (spec.ports?.healthBase || 19000) + index;
     const stateDir = path.join(this.stateRoot, instance);
     fs.mkdirSync(stateDir, { recursive: true });
-    const child = spawn(process.execPath, [WORKER], {
-      env: {
-        ...process.env,
-        FLEET_ROLE: spec.role || serviceId,
-        FLEET_SERVICE: serviceId,
-        FLEET_INSTANCE: instance,
-        FLEET_HEALTH_PORT: String(healthPort),
-        FLEET_STATE_DIR: stateDir,
-      },
-      stdio: ['ignore', 'pipe', 'pipe'],
-    });
+    const env = {
+      ...process.env,
+      FLEET_ROLE: spec.role || serviceId,
+      FLEET_SERVICE: serviceId,
+      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_MESSAGE_HZ: process.env.FLEET_MESSAGE_HZ || '8',
+    };
+    let child = null;
+    let pid = null;
+    if (machine.kind === 'agent') {
+      try {
+        const spawned = await postControl({
+          host: machine.host,
+          port: machine.agentPort,
+          path: '/spawn',
+          body: { env, instance },
+          timeoutMs: 4000,
+        });
+        pid = spawned.pid || null;
+      } catch (err) {
+        this.log('skip', { service: serviceId, reason: 'agent-spawn-failed', machine: machine.id, error: err.message });
+        return null;
+      }
+    } else {
+      child = spawn(process.execPath, [WORKER], { env, stdio: ['ignore', 'pipe', 'pipe'] });
+      pid = child.pid;
+    }
     const rec = {
       id: instance,
       service: serviceId,
       index,
-      pid: child.pid,
+      machine: machine.id,
+      host: machine.host,
+      pid,
       healthPort,
       stateDir,
       child,
+      kind: machine.kind,
       paused: false,
       startedAt: Date.now(),
       restarts: 0,
       lastError: null,
+      rttSamples: [],
     };
-    child.stderr?.on('data', () => {});
-    child.on('exit', (code, signal) => {
-      rec.exitCode = code;
-      rec.signal = signal;
-      rec.pid = null;
-      this.log('exit', { instance, service: serviceId, code, signal });
-    });
+    if (child) {
+      child.stderr?.on('data', () => {});
+      child.on('exit', (code, signal) => {
+        rec.exitCode = code;
+        rec.signal = signal;
+        rec.pid = null;
+        this.log('exit', { instance, service: serviceId, code, signal });
+      });
+    }
     this.instances.set(instance, rec);
-    this.log('start', { instance, service: serviceId, pid: rec.pid, healthPort });
-    const h = await waitForHealth(healthPort, 4000);
+    this.log('start', { instance, service: serviceId, machine: machine.id, pid: rec.pid, healthPort });
+    const h = await waitForHealth(rec.host, healthPort, 4000);
     rec.healthy = h.ok;
     rec.lastProbe = h;
     return rec;
@@ -118,7 +177,16 @@ export class Supervisor {
     const rec = this.instances.get(instanceId);
     if (!rec) return null;
     rec.stopping = true;
-    if (rec.child && rec.pid) {
+    if (rec.kind === 'agent') {
+      const m = this.machines().find((x) => x.id === rec.machine);
+      if (m) {
+        try {
+          await postControl({ host: m.host, port: m.agentPort, path: '/kill', body: { instance: instanceId } });
+        } catch {
+          /* ignore */
+        }
+      }
+    } else if (rec.child && rec.pid) {
       rec.child.kill('SIGTERM');
       await waitExit(rec.child, 2000);
     }
@@ -131,7 +199,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(rec.healthPort, '/pause');
+    await postControl({ host: rec.host || '127.0.0.1', port: rec.healthPort, path: '/pause' });
     rec.paused = true;
     this.log('pause', { instance: instanceId, service: rec.service });
     if (replace) await this.reconcile(rec.service);
@@ -141,7 +209,7 @@ export class Supervisor {
   async resumeInstance(instanceId) {
     const rec = this.instances.get(instanceId);
     if (!rec) throw new Error(`no instance ${instanceId}`);
-    await postControl(rec.healthPort, '/resume');
+    await postControl({ host: rec.host || '127.0.0.1', port: rec.healthPort, path: '/resume' });
     rec.paused = false;
     this.log('resume', { instance: instanceId, service: rec.service });
     return rec;
@@ -150,11 +218,9 @@ export class Supervisor {
   async restartInstance(instanceId) {
     const rec = this.instances.get(instanceId);
     if (!rec) throw new Error(`no instance ${instanceId}`);
-    const { service, index } = rec;
-    rec.child?.kill('SIGTERM');
-    await waitExit(rec.child, 2000);
-    this.instances.delete(instanceId);
-    const next = await this.startOne(service, { index });
+    const { service, index, machine } = rec;
+    await this.stopInstance(instanceId, { replace: false });
+    const next = await this.startOne(service, { index, machine: this.machines().find((m) => m.id === machine) });
     if (next) next.restarts = (rec.restarts || 0) + 1;
     this.log('restart', { instance: instanceId, service });
     return next;
@@ -163,16 +229,35 @@ export class Supervisor {
   async markUnhealthy(instanceId) {
     const rec = this.instances.get(instanceId);
     if (!rec) throw new Error(`no instance ${instanceId}`);
-    await postControl(rec.healthPort, '/unhealthy');
+    await postControl({ host: rec.host || '127.0.0.1', port: rec.healthPort, path: '/unhealthy' });
     return rec;
   }
 
   async probe(rec) {
     const spec = this.spec(rec.service);
-    const r = await httpProbe(rec.healthPort, spec.health?.path || '/health', spec.health?.timeoutMs || 800);
+    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,
+    });
     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' });
+        if (Number.isFinite(msg.rttMs)) {
+          rec.rttSamples = rec.rttSamples || [];
+          rec.rttSamples.push(msg.rttMs);
+          if (rec.rttSamples.length > 200) rec.rttSamples.splice(0, rec.rttSamples.length - 200);
+        }
+        rec.rtt = msg.rtt || r.body?.rtt || summarizeRtt(rec.rttSamples);
+      } catch {
+        rec.rtt = r.body?.rtt || summarizeRtt(rec.rttSamples || []);
+      }
+    }
     return r;
   }
 
@@ -279,17 +364,33 @@ export class Supervisor {
         instances: inst.map((i) => ({
           id: i.id,
           pid: i.pid,
+          machine: i.machine || 'local',
+          host: i.host || '127.0.0.1',
           healthPort: i.healthPort,
           paused: i.paused,
           healthy: i.healthy,
           restarts: i.restarts,
+          rtt: i.rtt || summarizeRtt(i.rttSamples || []),
         })),
       };
+      const allSamples = inst.flatMap((i) => i.rttSamples || []);
+      services[spec.id].rtt = summarizeRtt(allSamples);
       services[spec.id].health = classifyService(services[spec.id]);
     }
+    const machines = this.machines().map((m) => {
+      const inst = [...this.instances.values()].filter((i) => i.machine === m.id);
+      const samples = inst.flatMap((i) => i.rttSamples || []);
+      return {
+        ...m,
+        running: inst.filter((i) => i.pid).length,
+        available: inst.filter((i) => i.pid && !i.paused && i.healthy !== false).length,
+        rtt: summarizeRtt(samples),
+      };
+    });
     return {
       control: this.loaded.control,
       nats: this.loaded.nats,
+      machines,
       services,
       events: this.events.slice(-50),
     };
@@ -314,10 +415,10 @@ function waitExit(child, ms) {
   });
 }
 
-async function waitForHealth(port, ms) {
+async function waitForHealth(host, port, ms) {
   const deadline = Date.now() + ms;
   while (Date.now() < deadline) {
-    const r = await httpProbe(port, '/health', 300);
+    const r = await httpProbe({ host: host || '127.0.0.1', port, path: '/health', timeoutMs: 300 });
     if (r.ok || r.statusCode === 503) return r;
     await new Promise((r) => setTimeout(r, 50));
   }

packages/verae-fleet/src/worker.js

@@ -6,14 +6,20 @@
 import http from 'node:http';
 import fs from 'node:fs';
 import path from 'node:path';
+import { createHash, randomBytes } from 'node:crypto';
 import { fileURLToPath } from 'node:url';
+import { RttWindow } from './rtt.js';
 
 const role = process.env.FLEET_ROLE || 'unknown';
 const service = process.env.FLEET_SERVICE || role;
 const instance = process.env.FLEET_INSTANCE || `${service}-0`;
+const machine = process.env.FLEET_MACHINE || 'local';
 const port = Number(process.env.FLEET_HEALTH_PORT || 0);
+const bind = process.env.FLEET_HEALTH_BIND || '127.0.0.1';
 const stateDir = process.env.FLEET_STATE_DIR || path.join(process.cwd(), 'data', instance);
 const startedAt = new Date().toISOString();
+const rtt = new RttWindow(400);
+const hz = Number(process.env.FLEET_MESSAGE_HZ || 6);
 
 fs.mkdirSync(stateDir, { recursive: true });
 
@@ -43,6 +49,24 @@ if (role === 'tree-node' || role === 'archive-worm') {
   }
 }
 
+function processMessage() {
+  const t0 = process.hrtime.bigint();
+  if (paused) return null;
+  const payload = randomBytes(32);
+  if (archive) {
+    const sha = createHash('sha256').update(payload).digest('hex');
+    archive.put({ sha256: sha, kind: role === 'tree-node' ? 'tree' : 'publicMeta', record: { n: puts } });
+    archive.query(sha);
+    puts += 1;
+  } else {
+    createHash('sha256').update(payload).digest('hex');
+    puts += 1;
+  }
+  const ms = Number(process.hrtime.bigint() - t0) / 1e6;
+  rtt.add(ms);
+  return ms;
+}
+
 function snapshot() {
   return {
     ok: !paused && !forceUnhealthy,
@@ -51,11 +75,13 @@ function snapshot() {
     role,
     service,
     instance,
+    machine,
     port,
     pid: process.pid,
     startedAt,
     puts,
     archiveId: archive?.archiveId || null,
+    rtt: rtt.stats(),
   };
 }
 
@@ -91,17 +117,31 @@ const server = http.createServer((req, res) => {
     beat();
     return json(200, snapshot());
   }
+  if (url.pathname === '/message' && req.method === 'POST') {
+    const ms = processMessage();
+    return json(paused ? 503 : 200, { rttMs: ms, ...snapshot() });
+  }
+  if (url.pathname === '/metrics' && req.method === 'GET') {
+    return json(200, { rtt: rtt.stats(), samples: rtt.samples.slice(-80) });
+  }
   json(404, { error: 'not found' });
 });
 
-server.listen(port, '127.0.0.1', () => {
+server.listen(port, bind, () => {
   beat();
-  process.stdout.write(`fleet-worker ${instance} health http://127.0.0.1:${port}/health\n`);
+  process.stdout.write(`fleet-worker ${instance}@${machine} health http://${bind}:${port}/health\n`);
 });
 
 const iv = setInterval(beat, 400);
+const msgIv =
+  hz > 0
+    ? setInterval(() => {
+        if (!paused && !forceUnhealthy) processMessage();
+      }, Math.max(50, Math.round(1000 / hz)))
+    : null;
 function shutdown() {
   clearInterval(iv);
+  if (msgIv) clearInterval(msgIv);
   try {
     server.close();
   } catch {