Multi-Machine AI Agent Deployment

Org Status: 🟢 Active Cloudflare: N/A Last Audited: 2026-04-28

You have AI agents running in the cloud. You have local machines with GPUs, screens, and shells. The cloud can reason but cannot act. The machines can act but cannot coordinate. The missing piece is a deployment and coordination layer that registers machines, discovers capabilities, and routes work to the right node — installed with a single command.

This article covers the complete architecture for deploying a distributed AI agent platform across multiple machines, from the install script that bootstraps a new node, through the registration protocol that advertises its capabilities, to the pipeline system that chains nodes into workflows.

What you’ll learn:

• How to design a curl | sh installer that detects OS/arch, installs dependencies, configures services, and registers with a coordinator
• Machine registration and heartbeat protocols — how nodes announce what they can do
• Capability discovery patterns from Consul, Kubernetes, mDNS, and UPnP
• Node/Hub/Chain architecture for building multi-step agent pipelines
• Security: key exchange, mTLS, token rotation, and the zero-trust machine identity problem
• Real-world patterns from Ray, Dask, vLLM, and Temporal for distributed AI coordination
• A complete working example with TypeScript types, protocol messages, and install scripts

1. The Problem
2. Architecture Overview
3. Core Concepts
4. The Install Script
5. Machine Registration Protocol
6. Capability Discovery
7. Node / Hub / Chain Architecture
8. Security
9. Patterns
10. Small Examples
11. Real-World Multi-Machine AI Systems
12. Comparisons
13. Anti-Patterns
14. References

Building a distributed AI agent platform means solving three problems simultaneously:

1. Installation is fragmented. You have a cloud control plane (Cloudflare Workers, Durable Objects), local machine connectors (TypeScript daemon over WebSocket), a macOS HUD app, and various worker processes. Each one has its own install procedure, its own configuration, its own auth setup. Adding a new machine means 20 minutes of manual steps that nobody documents.

2. Machines don’t know about each other. Machine A has a GPU and can run inference. Machine B has a screen and can display status. Machine C has filesystem access and can execute code. But there is no registry, no discovery, no way for the cloud coordinator to say “I need a machine with GPU capability” and get routed to the right one.

3. Pipelines are hardcoded. When you need machine A to run inference, machine B to validate the output, and machine C to deploy the result, you wire it up manually. There is no abstraction for “chain these three capabilities together, routing each step to whichever machine can handle it.”

These problems compound. Without automated installation, you can’t easily add machines. Without capability discovery, you can’t route work. Without pipeline abstractions, you can’t compose multi-step workflows. The result is a system that works on one machine but collapses the moment you try to scale to two.

What changes if you get this right

A new machine joins your agent network with a single command:

curl -sfL https://install.atlas.dev | sh -s -- \
  --role node \
  --hub https://api-mom.garywu.dev \
  --token "$ATLAS_TOKEN"

Within 30 seconds, the machine has:

• Detected its OS, architecture, and available resources (CPU cores, memory, GPU)
• Downloaded and installed the Atlas connector binary
• Generated a machine identity keypair
• Registered with the central coordinator, advertising its capabilities
• Started heartbeating every 60 seconds
• Configured itself as a launchd service (macOS) or systemd unit (Linux)
• Begun accepting work dispatched from the cloud

Other machines in the network immediately see the new node and can route tasks to it. The cloud coordinator’s capability map updates in real time. Pipelines that were waiting for a “gpu-inference” capability suddenly have a machine to run on.

The system has three layers:

┌─────────────────────────────────────────────────────────────────┐
│  CLOUD LAYER (Cloudflare Workers + Durable Objects)             │
│                                                                 │
│  ┌──────────────┐  ┌──────────────┐  ┌────────────────────────┐ │
│  │ Atlas Center  │  │ Board Agents │  │ Pipeline Engine        │ │
│  │ (Registry DO) │  │ (C-suite DOs)│  │ (Workflow Orchestrator)│ │
│  │               │  │              │  │                        │ │
│  │ - machine map │  │ - perceive() │  │ - DAG execution        │ │
│  │ - capability  │  │ - reason()   │  │ - step routing         │ │
│  │   index       │  │ - dispatch() │  │ - retry/checkpoint     │ │
│  │ - presence    │  │              │  │                        │ │
│  └──────┬───────┘  └──────┬───────┘  └─────────┬──────────────┘ │
│         │                 │                     │               │
│         └─────────────┬───┘─────────────────────┘               │
│                       │                                         │
│               ┌───────┴────────┐                                │
│               │   API Mom      │  ← central proxy/registry      │
│               │ (cost routing) │                                │
│               └───────┬────────┘                                │
└───────────────────────┼─────────────────────────────────────────┘
                        │
               ═══ WebSocket ═══  (persistent, bidirectional)
                        │
┌───────────────────────┼─────────────────────────────────────────┐
│  LOCAL LAYER (per machine)                                      │
│               ┌───────┴────────┐                                │
│               │  Connector     │  ← installed by install script │
│               │                │                                │
│               │ - register()   │  → POST /v1/machines/register  │
│               │ - heartbeat()  │  → POST /v1/machines/heartbeat │
│               │ - capabilities │  → hud-notify, gpu-inference,  │
│               │ - ws handler   │    code-exec, macos-notify...  │
│               └──┬──────────┬──┘                                │
│                  │          │                                    │
│           ┌──────┴──┐  ┌───┴────────┐                           │
│           │  HUD    │  │ Workers    │                            │
│           │ (notch) │  │ (GPU, etc) │                            │
│           └─────────┘  └────────────┘                            │
└─────────────────────────────────────────────────────────────────┘

Three machine roles:

Machine Identity

Every machine in the network has a unique identity — a combination of a machine ID, a cryptographic keypair, and a set of metadata that describes what the machine is and what it can do.

interface MachineIdentity {
  /** Unique machine identifier (typically hostname or UUID) */
  machineId: string;

  /** Human-readable label */
  label: string;

  /** Ed25519 public key for this machine */
  publicKey: string;

  /** Machine role in the network */
  role: "node" | "hub" | "hybrid";

  /** When this identity was created */
  createdAt: string;

  /** Network addresses where this machine can be reached */
  addresses: {
    /** Local network address (mDNS or static IP) */
    local?: string;
    /** Tailscale/WireGuard mesh address */
    mesh?: string;
    /** Public address (if directly reachable) */
    public?: string;
  };
}

The machine ID is generated during installation and persisted to ~/.atlas/machine-id. It does not change across restarts, reinstalls, or network changes. The keypair is generated fresh on first install and stored in ~/.atlas/keys/.

Key insight: Machine identity must survive reboots, network changes, and even OS upgrades. Tying identity to hardware (like a MAC address) is fragile. Tying it to a hostname is ambiguous. The safest approach is a generated UUID stored on disk, paired with a keypair that proves ownership.

Capabilities

A capability is a named, typed unit of work that a machine can perform. It has a name, an input schema, an output schema, and metadata about resource requirements.

interface CapabilitySpec {
  /** Unique capability name (e.g., "gpu-inference", "hud-notify") */
  name: string;

  /** Human-readable description */
  description: string;

  /** Zod-compatible JSON schema for the input */
  inputSchema: Record<string, unknown>;

  /** Zod-compatible JSON schema for the output */
  outputSchema?: Record<string, unknown>;

  /** Resource requirements to execute this capability */
  resources?: {
    /** Minimum CPU cores needed */
    minCpuCores?: number;
    /** Minimum memory in GB */
    minMemoryGb?: number;
    /** Requires GPU */
    gpu?: boolean;
    /** Requires display (for HUD, notifications) */
    display?: boolean;
    /** Requires filesystem access */
    filesystem?: boolean;
    /** Requires network access */
    network?: boolean;
    /** Estimated execution time in seconds */
    estimatedDurationSeconds?: number;
  };

  /** Tags for filtering and routing */
  tags?: string[];

  /** Maximum concurrent executions on this machine */
  maxConcurrency?: number;
}

A minimal working example of registering capabilities:

import * as os from "node:os";
import { z } from "zod";

const capabilities: CapabilitySpec[] = [
  {
    name: "hud-notify",
    description: "Write a message to the local HUD status queue",
    inputSchema: {
      type: "object",
      properties: {
        severity: { type: "string", enum: ["green", "yellow", "red"] },
        message: { type: "string" },
      },
      required: ["severity", "message"],
    },
    resources: { display: true },
    maxConcurrency: 10,
  },
  {
    name: "jane-status",
    description: "Return local machine status (CPU, memory, uptime)",
    inputSchema: { type: "object" },
    resources: {},
    maxConcurrency: 50,
  },
];

// Capabilities are declared statically but resource availability
// is checked dynamically at registration time
function getAvailableCapabilities(): CapabilitySpec[] {
  const hasDisplay = os.platform() === "darwin"; // macOS has a display
  return capabilities.filter((cap) => {
    if (cap.resources?.display && !hasDisplay) return false;
    if (cap.resources?.gpu && !detectGpu()) return false;
    return true;
  });
}

Key insight: Capabilities are not the same as “what software is installed.” A machine might have ollama installed but be out of VRAM. Capabilities must be dynamically validated at registration time and continuously re-evaluated during heartbeats. A capability that was available 60 seconds ago might not be available now.

Registration

Registration is the process by which a machine announces itself to the hub, providing its identity, capabilities, and resource snapshot. The hub stores this in its machine registry and makes it available for capability routing.

interface MachineRegistration {
  /** Machine identity */
  machineId: string;
  label: string;

  /** WebSocket endpoint for dispatching work */
  endpoint: string;

  /** Capabilities this machine offers */
  capabilities: CapabilitySpec[];

  /** Current resource snapshot */
  resources: ResourceSnapshot;

  /** How long this registration is valid (seconds) */
  ttlSeconds: number;

  /** Protocol version */
  protocolVersion: string;
}

interface ResourceSnapshot {
  platform: string;
  arch: string;
  cpus: number;
  memoryGb: number;
  freeMemoryGb: number;
  loadAvg: [number, number, number];
  uptimeSeconds: number;
  gpus?: GpuInfo[];
  diskFreeGb?: number;
}

interface GpuInfo {
  name: string;
  memoryMb: number;
  freeMemoryMb: number;
  utilization: number;
}

Registration is idempotent. Calling register with the same machine ID updates the existing registration rather than creating a duplicate. The TTL ensures that if a machine crashes without deregistering, its registration expires and it is removed from the capability index.

Key insight: Registration must be idempotent and TTL-based. Machines crash. Networks partition. Graceful deregistration (DELETE on shutdown) is best-effort. The TTL is the real garbage collector. Set it to 3x your heartbeat interval — if you heartbeat every 60s, set TTL to 180s.

Heartbeat

The heartbeat is a periodic signal from a machine to the hub, confirming liveness and updating resource state. It is lighter than a full registration — it refreshes the TTL and updates dynamic fields (free memory, load average, GPU utilization) without resending the full capability list.

interface HeartbeatPayload {
  machineId: string;
  capabilities: string[]; // Just the names, not full specs
  resources: ResourceSnapshot;
  busy: boolean;
  currentTasks?: string[]; // IDs of tasks currently executing
  version: string;
}

// Heartbeat response from the hub
interface HeartbeatResponse {
  /** Whether the hub still recognizes this machine */
  registered: boolean;
  /** If false, machine should re-register */
  reRegister?: boolean;
  /** Pending configuration updates */
  config?: Partial<ConnectorConfig>;
  /** Commands to execute (e.g., "upgrade", "restart") */
  commands?: HubCommand[];
}

The heartbeat response is also a command channel. The hub can use it to tell a machine to upgrade, change its configuration, or even deregister. This avoids the need for a separate push channel for administrative commands (though the WebSocket connection provides one too).

Design Principles

The best install scripts in the ecosystem share common patterns. K3s installs a complete Kubernetes distribution with curl -sfL https://get.k3s.io | sh -. Homebrew bootstraps a package manager. mise and proto install runtime version managers. Tailscale installs a mesh VPN.

What makes them work:

1. Detect everything, assume nothing. OS, architecture, init system, shell, existing installations.
2. Single entry point, multiple roles. The same script installs a server or an agent depending on environment variables.
3. Idempotent. Running the script twice produces the same result. It detects existing installations and skips or upgrades.
4. Verifiable. Binary checksums are validated before installation.
5. Service integration. The script configures the installed binary as a system service (launchd, systemd, openrc) so it survives reboots.
6. Minimal dependencies. Only curl/wget, tar/unzip, and a shell. Everything else is bundled.

K3s Install Script Anatomy

The K3s install script is the gold standard for curl | sh distributed system installers. Its structure:

verify_system()          → detect init system (systemd/openrc)
setup_verify_arch()      → map uname -m to binary suffix
setup_env()              → determine role (server/agent from K3S_URL)
download_and_verify()    → fetch binary + SHA256 checksum, verify
setup_selinux()          → configure SELinux policies
create_env_file()        → write /etc/systemd/system/k3s.env
create_service_file()    → write systemd unit or openrc init script
create_killall()         → write k3s-killall.sh
create_uninstall()       → write k3s-uninstall.sh
systemd_enable_start()   → enable and start the service

Key decisions:

• Role is determined by a single env var (K3S_URL being set means “agent mode”)
• Version pinning via INSTALL_K3S_VERSION or auto-detect latest stable
• Skip download with INSTALL_K3S_SKIP_DOWNLOAD for air-gapped installs
• Force restart with INSTALL_K3S_FORCE_RESTART

Atlas Connector Install Script

Here is a complete install script for the Atlas connector, following the patterns from k3s and Tailscale:

#!/bin/sh
#
#

set -eu


ATLAS_REPO="garywu/atlas"
ATLAS_API="https://api.github.com/repos/${ATLAS_REPO}/releases"
ATLAS_DEFAULT_HUB="https://api-mom.garywu.dev"
ATLAS_CONFIG_DIR="${ATLAS_CONFIG_DIR:-${HOME}/.atlas}"
ATLAS_INSTALL_DIR="${ATLAS_INSTALL_DIR:-/usr/local/bin}"
ATLAS_LOG_DIR="${ATLAS_CONFIG_DIR}/logs"


RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
BLUE='\033[0;34m'
NC='\033[0m' # No Color

info()  { printf "${BLUE}[atlas]${NC} %s\n" "$*"; }
ok()    { printf "${GREEN}[atlas]${NC} %s\n" "$*"; }
warn()  { printf "${YELLOW}[atlas]${NC} %s\n" "$*" >&2; }
fatal() { printf "${RED}[atlas]${NC} %s\n" "$*" >&2; exit 1; }


detect_os() {
  OS="$(uname -s | tr '[:upper:]' '[:lower:]')"
  case "$OS" in
    darwin)  OS="darwin" ;;
    linux)   OS="linux" ;;
    *)       fatal "Unsupported OS: $OS" ;;
  esac
}

detect_arch() {
  ARCH="$(uname -m)"
  case "$ARCH" in
    x86_64|amd64)   ARCH="amd64" ;;
    aarch64|arm64)   ARCH="arm64" ;;
    armv7l|armhf)    ARCH="arm" ;;
    *)               fatal "Unsupported architecture: $ARCH" ;;
  esac
}

detect_init_system() {
  if [ -d /run/systemd/system ]; then
    INIT_SYSTEM="systemd"
  elif command -v launchctl >/dev/null 2>&1; then
    INIT_SYSTEM="launchd"
  elif command -v openrc-run >/dev/null 2>&1; then
    INIT_SYSTEM="openrc"
  else
    INIT_SYSTEM="none"
    warn "No supported init system detected — skipping service setup"
  fi
}


check_dependencies() {
  for cmd in curl tar; do
    if ! command -v "$cmd" >/dev/null 2>&1; then
      fatal "Required command not found: $cmd"
    fi
  done

  # Check for Node.js (required for the connector)
  if ! command -v node >/dev/null 2>&1; then
    warn "Node.js not found — attempting to install via mise"
    install_node_runtime
  fi

  NODE_VERSION="$(node --version 2>/dev/null || echo 'none')"
  info "Node.js version: $NODE_VERSION"
}

install_node_runtime() {
  if command -v mise >/dev/null 2>&1; then
    info "Installing Node.js via mise..."
    mise install node@lts
    mise use --global node@lts
  elif command -v brew >/dev/null 2>&1; then
    info "Installing Node.js via Homebrew..."
    brew install node
  elif command -v apt-get >/dev/null 2>&1; then
    info "Installing Node.js via apt..."
    curl -fsSL https://deb.nodesource.com/setup_lts.x | sh -
    apt-get install -y nodejs
  else
    fatal "Cannot install Node.js — please install it manually"
  fi
}


parse_args() {
  ROLE="${ATLAS_ROLE:-node}"
  HUB_URL="${ATLAS_HUB_URL:-${ATLAS_DEFAULT_HUB}}"
  TOKEN="${ATLAS_TOKEN:-}"
  MACHINE_ID="${ATLAS_MACHINE_ID:-$(hostname)}"
  LABEL=""
  CHANNEL="stable"
  VERSION=""
  SKIP_SERVICE=false
  SKIP_START=false
  UNINSTALL=false

  while [ $# -gt 0 ]; do
    case "$1" in
      --role)        ROLE="$2";       shift 2 ;;
      --hub)         HUB_URL="$2";    shift 2 ;;
      --token)       TOKEN="$2";      shift 2 ;;
      --machine-id)  MACHINE_ID="$2"; shift 2 ;;
      --label)       LABEL="$2";      shift 2 ;;
      --channel)     CHANNEL="$2";    shift 2 ;;
      --version)     VERSION="$2";    shift 2 ;;
      --skip-service) SKIP_SERVICE=true; shift ;;
      --skip-start)  SKIP_START=true; shift ;;
      --uninstall)   UNINSTALL=true;  shift ;;
      *)             fatal "Unknown option: $1" ;;
    esac
  done

  if [ -z "$LABEL" ]; then
    LABEL="$(hostname) (${OS}/${ARCH})"
  fi
}


resolve_version() {
  if [ -n "$VERSION" ]; then
    info "Using pinned version: $VERSION"
    return
  fi

  info "Resolving latest ${CHANNEL} version..."
  VERSION=$(curl -sfL "${ATLAS_API}/latest" | \
    grep '"tag_name"' | \
    sed -E 's/.*"tag_name": *"([^"]+)".*/\1/')

  if [ -z "$VERSION" ]; then
    fatal "Could not determine latest version"
  fi

  info "Latest version: $VERSION"
}


download_binary() {
  BINARY_NAME="atlas-connector-${OS}-${ARCH}"
  DOWNLOAD_URL="https://github.com/${ATLAS_REPO}/releases/download/${VERSION}/${BINARY_NAME}.tar.gz"
  CHECKSUM_URL="${DOWNLOAD_URL}.sha256"

  TMPDIR="$(mktemp -d)"
  trap 'rm -rf "$TMPDIR"' EXIT

  info "Downloading ${BINARY_NAME} ${VERSION}..."
  curl -sfL -o "${TMPDIR}/connector.tar.gz" "$DOWNLOAD_URL" || \
    fatal "Download failed: $DOWNLOAD_URL"

  info "Verifying checksum..."
  curl -sfL -o "${TMPDIR}/connector.sha256" "$CHECKSUM_URL" || \
    warn "Checksum file not available — skipping verification"

  if [ -f "${TMPDIR}/connector.sha256" ]; then
    cd "$TMPDIR"
    if command -v sha256sum >/dev/null 2>&1; then
      echo "$(cat connector.sha256)  connector.tar.gz" | sha256sum -c - || \
        fatal "Checksum verification failed"
    elif command -v shasum >/dev/null 2>&1; then
      echo "$(cat connector.sha256)  connector.tar.gz" | shasum -a 256 -c - || \
        fatal "Checksum verification failed"
    fi
    ok "Checksum verified"
  fi

  info "Extracting..."
  tar xzf "${TMPDIR}/connector.tar.gz" -C "$TMPDIR"
}


install_binary() {
  info "Installing to ${ATLAS_INSTALL_DIR}/"

  # Ensure install directory exists
  if [ ! -w "$ATLAS_INSTALL_DIR" ]; then
    if command -v sudo >/dev/null 2>&1; then
      sudo mkdir -p "$ATLAS_INSTALL_DIR"
      sudo cp "${TMPDIR}/atlas-connector" "${ATLAS_INSTALL_DIR}/"
      sudo chmod +x "${ATLAS_INSTALL_DIR}/atlas-connector"
    else
      fatal "Cannot write to ${ATLAS_INSTALL_DIR} — run as root or set ATLAS_INSTALL_DIR"
    fi
  else
    mkdir -p "$ATLAS_INSTALL_DIR"
    cp "${TMPDIR}/atlas-connector" "${ATLAS_INSTALL_DIR}/"
    chmod +x "${ATLAS_INSTALL_DIR}/atlas-connector"
  fi

  ok "Installed atlas-connector to ${ATLAS_INSTALL_DIR}/atlas-connector"
}


setup_config() {
  mkdir -p "$ATLAS_CONFIG_DIR"
  mkdir -p "$ATLAS_LOG_DIR"
  mkdir -p "${ATLAS_CONFIG_DIR}/keys"

  # Generate machine ID file if not present
  MACHINE_ID_FILE="${ATLAS_CONFIG_DIR}/machine-id"
  if [ ! -f "$MACHINE_ID_FILE" ]; then
    if command -v uuidgen >/dev/null 2>&1; then
      uuidgen | tr '[:upper:]' '[:lower:]' > "$MACHINE_ID_FILE"
    else
      cat /proc/sys/kernel/random/uuid > "$MACHINE_ID_FILE" 2>/dev/null || \
        head -c 16 /dev/urandom | od -An -tx1 | tr -d ' \n' | \
        sed 's/\(.\{8\}\)\(.\{4\}\)\(.\{4\}\)\(.\{4\}\)/\1-\2-\3-\4-/' \
        > "$MACHINE_ID_FILE"
    fi
    info "Generated machine ID: $(cat "$MACHINE_ID_FILE")"
  else
    info "Existing machine ID: $(cat "$MACHINE_ID_FILE")"
  fi

  # Generate Ed25519 keypair if not present
  KEY_FILE="${ATLAS_CONFIG_DIR}/keys/machine.key"
  PUB_FILE="${ATLAS_CONFIG_DIR}/keys/machine.pub"
  if [ ! -f "$KEY_FILE" ]; then
    info "Generating Ed25519 keypair..."
    if command -v ssh-keygen >/dev/null 2>&1; then
      ssh-keygen -t ed25519 -f "$KEY_FILE" -N "" -C "atlas-connector@$(hostname)" \
        >/dev/null 2>&1
      ok "Keypair generated"
    elif command -v openssl >/dev/null 2>&1; then
      openssl genpkey -algorithm Ed25519 -out "$KEY_FILE" 2>/dev/null
      openssl pkey -in "$KEY_FILE" -pubout -out "$PUB_FILE" 2>/dev/null
      ok "Keypair generated (OpenSSL)"
    else
      warn "No key generation tool available — skipping keypair generation"
    fi
    chmod 600 "$KEY_FILE" 2>/dev/null || true
  fi

  # Write config file
  CONFIG_FILE="${ATLAS_CONFIG_DIR}/connector.json"
  cat > "$CONFIG_FILE" <<EOF
{
  "machineId": "$(cat "$MACHINE_ID_FILE")",
  "label": "${LABEL}",
  "role": "${ROLE}",
  "hubUrl": "${HUB_URL}",
  "protocolVersion": "1.0",
  "heartbeatIntervalMs": 60000,
  "registrationTtlSeconds": 180,
  "capabilities": {
    "autoDetect": true,
    "additional": []
  }
}
EOF

  ok "Configuration written to $CONFIG_FILE"
}


setup_service() {
  if [ "$SKIP_SERVICE" = true ]; then
    info "Skipping service setup (--skip-service)"
    return
  fi

  case "$INIT_SYSTEM" in
    launchd)  setup_launchd ;;
    systemd)  setup_systemd ;;
    openrc)   setup_openrc ;;
    *)        warn "Cannot configure service — no supported init system" ;;
  esac
}

setup_launchd() {
  PLIST_PATH="${HOME}/Library/LaunchAgents/com.atlas.connector.plist"
  info "Writing launchd plist: $PLIST_PATH"

  cat > "$PLIST_PATH" <<EOF
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN"
  "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
    <key>Label</key>
    <string>com.atlas.connector</string>

    <key>ProgramArguments</key>
    <array>
        <string>${ATLAS_INSTALL_DIR}/atlas-connector</string>
    </array>

    <key>WorkingDirectory</key>
    <string>${ATLAS_CONFIG_DIR}</string>

    <key>RunAtLoad</key>
    <true/>

    <key>KeepAlive</key>
    <dict>
        <key>SuccessfulExit</key>
        <false/>
    </dict>

    <key>ThrottleInterval</key>
    <integer>10</integer>

    <key>StandardOutPath</key>
    <string>${ATLAS_LOG_DIR}/connector.log</string>

    <key>StandardErrorPath</key>
    <string>${ATLAS_LOG_DIR}/connector.err.log</string>

    <key>EnvironmentVariables</key>
    <dict>
        <key>PATH</key>
        <string>/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin</string>
        <key>ATLAS_CONFIG_DIR</key>
        <string>${ATLAS_CONFIG_DIR}</string>
        <key>ATLAS_TOKEN</key>
        <string>${TOKEN}</string>
    </dict>
</dict>
</plist>
EOF

  ok "launchd service configured"
}

setup_systemd() {
  UNIT_FILE="/etc/systemd/system/atlas-connector.service"
  info "Writing systemd unit: $UNIT_FILE"

  sudo tee "$UNIT_FILE" >/dev/null <<EOF
[Unit]
Description=Atlas Connector
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
ExecStart=${ATLAS_INSTALL_DIR}/atlas-connector
WorkingDirectory=${ATLAS_CONFIG_DIR}
Restart=on-failure
RestartSec=10
StandardOutput=journal
StandardError=journal
Environment=ATLAS_CONFIG_DIR=${ATLAS_CONFIG_DIR}
Environment=ATLAS_TOKEN=${TOKEN}
Environment=PATH=/usr/local/bin:/usr/bin:/bin

NoNewPrivileges=true
ProtectSystem=strict
ProtectHome=read-only
ReadWritePaths=${ATLAS_CONFIG_DIR}
PrivateTmp=true

[Install]
WantedBy=multi-user.target
EOF

  sudo systemctl daemon-reload
  sudo systemctl enable atlas-connector
  ok "systemd service configured and enabled"
}

setup_openrc() {
  INIT_SCRIPT="/etc/init.d/atlas-connector"
  info "Writing OpenRC init script: $INIT_SCRIPT"

  sudo tee "$INIT_SCRIPT" >/dev/null <<'INITEOF'
#!/sbin/openrc-run
description="Atlas Connector"
command="${ATLAS_INSTALL_DIR}/atlas-connector"
command_background=true
pidfile="/var/run/atlas-connector.pid"
output_log="${ATLAS_LOG_DIR}/connector.log"
error_log="${ATLAS_LOG_DIR}/connector.err.log"

depend() {
    need net
    after firewall
}
INITEOF

  sudo chmod +x "$INIT_SCRIPT"
  sudo rc-update add atlas-connector default
  ok "OpenRC service configured"
}


register_with_hub() {
  if [ -z "$TOKEN" ]; then
    warn "No token provided — skipping initial registration"
    warn "Set ATLAS_TOKEN or use --token to register with the hub"
    return
  fi

  info "Registering with hub at ${HUB_URL}..."

  RESPONSE=$(curl -sf -X POST \
    -H "Content-Type: application/json" \
    -H "Authorization: Bearer ${TOKEN}" \
    -d "{
      \"machine_id\": \"$(cat "${ATLAS_CONFIG_DIR}/machine-id")\",
      \"label\": \"${LABEL}\",
      \"role\": \"${ROLE}\",
      \"resources\": {
        \"platform\": \"${OS}\",
        \"arch\": \"${ARCH}\",
        \"cpus\": $(nproc 2>/dev/null || sysctl -n hw.ncpu 2>/dev/null || echo 1)
      },
      \"ttl_seconds\": 180
    }" \
    "${HUB_URL}/v1/machines/register" 2>&1) || {
    warn "Registration failed — connector will retry on startup"
    return
  }

  ok "Registered with hub"
}


start_service() {
  if [ "$SKIP_START" = true ]; then
    info "Skipping service start (--skip-start)"
    return
  fi

  case "$INIT_SYSTEM" in
    launchd)
      launchctl load "$PLIST_PATH" 2>/dev/null || true
      launchctl start com.atlas.connector 2>/dev/null || true
      ok "Service started via launchd"
      ;;
    systemd)
      sudo systemctl start atlas-connector
      ok "Service started via systemd"
      ;;
    openrc)
      sudo rc-service atlas-connector start
      ok "Service started via OpenRC"
      ;;
    *)
      info "Start the connector manually:"
      info "  atlas-connector"
      ;;
  esac
}


do_uninstall() {
  info "Uninstalling Atlas connector..."

  # Stop service
  case "$INIT_SYSTEM" in
    launchd)
      launchctl unload "${HOME}/Library/LaunchAgents/com.atlas.connector.plist" 2>/dev/null || true
      rm -f "${HOME}/Library/LaunchAgents/com.atlas.connector.plist"
      ;;
    systemd)
      sudo systemctl stop atlas-connector 2>/dev/null || true
      sudo systemctl disable atlas-connector 2>/dev/null || true
      sudo rm -f /etc/systemd/system/atlas-connector.service
      sudo systemctl daemon-reload
      ;;
    openrc)
      sudo rc-service atlas-connector stop 2>/dev/null || true
      sudo rc-update del atlas-connector 2>/dev/null || true
      sudo rm -f /etc/init.d/atlas-connector
      ;;
  esac

  # Deregister from hub
  if [ -f "${ATLAS_CONFIG_DIR}/machine-id" ] && [ -n "${TOKEN:-}" ]; then
    MID="$(cat "${ATLAS_CONFIG_DIR}/machine-id")"
    curl -sf -X DELETE \
      -H "Authorization: Bearer ${TOKEN}" \
      "${HUB_URL}/v1/machines/${MID}" 2>/dev/null || true
    info "Deregistered from hub"
  fi

  # Remove binary
  rm -f "${ATLAS_INSTALL_DIR}/atlas-connector"

  # Ask about config
  printf "Remove configuration directory %s? [y/N] " "$ATLAS_CONFIG_DIR"
  read -r answer
  if [ "$answer" = "y" ] || [ "$answer" = "Y" ]; then
    rm -rf "$ATLAS_CONFIG_DIR"
    ok "Configuration removed"
  else
    info "Configuration preserved at $ATLAS_CONFIG_DIR"
  fi

  ok "Atlas connector uninstalled"
  exit 0
}


main() {
  info "Atlas Connector Installer"
  info "========================="

  detect_os
  detect_arch
  detect_init_system
  parse_args "$@"

  info "OS: ${OS}  Arch: ${ARCH}  Init: ${INIT_SYSTEM}  Role: ${ROLE}"

  if [ "$UNINSTALL" = true ]; then
    do_uninstall
  fi

  check_dependencies
  resolve_version
  download_binary
  install_binary
  setup_config
  setup_service
  register_with_hub
  start_service

  ok "========================================="
  ok "Atlas Connector installed successfully!"
  ok "========================================="
  ok ""
  ok "  Machine ID:  $(cat "${ATLAS_CONFIG_DIR}/machine-id")"
  ok "  Role:        ${ROLE}"
  ok "  Hub:         ${HUB_URL}"
  ok "  Config:      ${ATLAS_CONFIG_DIR}/connector.json"
  ok "  Logs:        ${ATLAS_LOG_DIR}/connector.log"
  ok ""
  ok "Manage the service:"
  case "$INIT_SYSTEM" in
    launchd)
      ok "  launchctl start com.atlas.connector"
      ok "  launchctl stop com.atlas.connector"
      ok "  tail -f ${ATLAS_LOG_DIR}/connector.log"
      ;;
    systemd)
      ok "  sudo systemctl status atlas-connector"
      ok "  sudo journalctl -u atlas-connector -f"
      ;;
    openrc)
      ok "  sudo rc-service atlas-connector status"
      ok "  tail -f ${ATLAS_LOG_DIR}/connector.log"
      ;;
  esac
}

main "$@"

What this script does differently from most installers

1. Generates a persistent machine identity. Not just installing a binary — creating a cryptographic identity that will represent this machine in the network.

2. Auto-detects and installs Node.js via whichever package manager is available (mise, Homebrew, apt).

3. Configures the init system automatically. Detects launchd (macOS), systemd (Linux), or OpenRC and writes the appropriate service configuration with security hardening.

4. Registers with the hub during installation. The machine is immediately discoverable by other nodes in the network.

5. Includes a complete uninstaller that deregisters from the hub, stops the service, removes the binary, and optionally cleans up configuration.

Key insight: The best install scripts do not just copy a binary. They establish identity, configure supervision, and integrate the new component into an existing system. The install script is the onboarding protocol.

Protocol Messages

The connector communicates with the hub using a typed message protocol over WebSocket. Here is the complete type system:

// ── Messages from Hub → Machine (Server Messages) ──────────────────────────

type ServerMessage =
  | DispatchRequest
  | PingMessage
  | PongMessage
  | UpdateCommand
  | ConfigPush
  | CapabilityQuery;

interface DispatchRequest {
  type: "request";
  /** Unique request ID for correlation */
  id: string;
  /** Capability to invoke */
  capability: string;
  /** Input payload for the capability handler */
  input: unknown;
  /** Optional: which pipeline this request belongs to */
  pipelineId?: string;
  /** Optional: step index within the pipeline */
  stepIndex?: number;
  /** Timeout in milliseconds */
  timeoutMs?: number;
}

interface PingMessage {
  type: "ping";
  timestamp: number;
}

interface PongMessage {
  type: "pong";
  timestamp: number;
}

interface UpdateCommand {
  type: "update";
  /** Target version to upgrade to */
  version?: string;
  /** URL to download the update from */
  url?: string;
  /** Whether to restart after update */
  restart?: boolean;
}

interface ConfigPush {
  type: "config-push";
  /** Partial config to merge */
  config: Partial<{
    heartbeatIntervalMs: number;
    capabilities: string[];
    logLevel: "debug" | "info" | "warn" | "error";
  }>;
}

interface CapabilityQuery {
  type: "capability-query";
  /** Request ID */
  id: string;
  /** Capability name to check */
  capability: string;
}

// ── Messages from Machine → Hub (Client Messages) ──────────────────────────

type ClientMessage =
  | ResponseMessage
  | StatusReport
  | PingMessage
  | PongMessage
  | CapabilityAdvertisement
  | TaskProgress;

interface ResponseMessage {
  type: "response";
  /** Correlates with the request ID */
  id: string;
  /** HTTP-style status code */
  status: number;
  /** Result payload */
  body: unknown;
}

interface StatusReport {
  type: "status-report";
  /** Machine ID */
  machine: string;
  /** Whether the machine is currently processing a task */
  busy: boolean;
  /** Connector version */
  version: string;
  /** Current resource snapshot */
  resources?: ResourceSnapshot;
  /** IDs of tasks currently executing */
  activeTasks?: string[];
}

interface CapabilityAdvertisement {
  type: "capability-advertisement";
  /** Full capability specs (sent on connect and when capabilities change) */
  capabilities: CapabilitySpec[];
}

interface TaskProgress {
  type: "task-progress";
  /** Request ID this progress is for */
  id: string;
  /** Progress percentage (0-100) */
  progress: number;
  /** Human-readable status message */
  message?: string;
  /** Partial result (for streaming) */
  partial?: unknown;
}

Registration Flow

The full registration flow has six phases:

Machine                                Hub (Atlas Center DO)
  │                                        │
  │  1. POST /v1/machines/register         │
  │  ──────────────────────────────────►   │
  │  { machineId, capabilities,            │
  │    resources, ttlSeconds }             │
  │                                        │  Store in machine registry
  │  ◄──────────────────────────────────   │  Update capability index
  │  { registered: true, hubWsUrl }        │
  │                                        │
  │  2. WebSocket connect                  │
  │  ══════════════════════════════════►   │
  │  Authorization: Bearer <token>         │
  │                                        │
  │  3. status-report (initial)            │
  │  ──────────────────────────────────►   │
  │  { machine, busy: false, version }     │  Mark machine as "connected"
  │                                        │
  │  4. capability-advertisement           │
  │  ──────────────────────────────────►   │
  │  { capabilities: [...full specs] }     │  Merge into capability index
  │                                        │
  │  5. heartbeat loop (every 60s)         │
  │  ──────────────────────────────────►   │
  │  POST /v1/machines/heartbeat           │  Refresh TTL
  │  { machineId, resources }              │  Update resource snapshot
  │  ◄──────────────────────────────────   │
  │  { registered: true, commands: [] }    │
  │                                        │
  │  6. dispatch (when work arrives)       │
  │  ◄══════════════════════════════════   │
  │  { type: "request", capability,        │
  │    input, id }                         │
  │  ══════════════════════════════════►   │
  │  { type: "response", id, status,       │
  │    body }                              │
  │                                        │

Hub-Side Registry (Durable Object)

The hub maintains a machine registry as a Durable Object. Here is the core data structure and routing logic:

interface MachineRegistry {
  /** All registered machines, keyed by machine ID */
  machines: Map<string, RegisteredMachine>;

  /** Capability index: capability name → set of machine IDs */
  capabilityIndex: Map<string, Set<string>>;

  /** Active WebSocket connections, keyed by machine ID */
  connections: Map<string, WebSocket>;
}

interface RegisteredMachine {
  machineId: string;
  label: string;
  role: "node" | "hub" | "hybrid";
  capabilities: CapabilitySpec[];
  resources: ResourceSnapshot;
  registeredAt: number;
  lastHeartbeat: number;
  ttlSeconds: number;
  busy: boolean;
  activeTasks: string[];
  version: string;
  connected: boolean;
}

class AtlasCenterDO {
  private registry: MachineRegistry = {
    machines: new Map(),
    capabilityIndex: new Map(),
    connections: new Map(),
  };

  /**
   * Find machines that can handle a given capability.
   * Returns machines sorted by suitability (least busy, most resources).
   */
  findMachinesForCapability(
    capabilityName: string,
    requirements?: Partial<CapabilitySpec["resources"]>
  ): RegisteredMachine[] {
    const machineIds = this.registry.capabilityIndex.get(capabilityName);
    if (!machineIds) return [];

    const now = Date.now();
    const candidates: RegisteredMachine[] = [];

    for (const id of machineIds) {
      const machine = this.registry.machines.get(id);
      if (!machine) continue;

      // Check TTL — skip expired registrations
      const expiresAt = machine.lastHeartbeat + machine.ttlSeconds * 1000;
      if (now > expiresAt) {
        this.deregisterMachine(id);
        continue;
      }

      // Check if machine is connected
      if (!machine.connected) continue;

      // Check resource requirements
      if (requirements) {
        if (requirements.gpu && !machine.resources.gpus?.length) continue;
        if (
          requirements.minMemoryGb &&
          machine.resources.freeMemoryGb < requirements.minMemoryGb
        )
          continue;
        if (
          requirements.minCpuCores &&
          machine.resources.cpus < requirements.minCpuCores
        )
          continue;
      }

      // Check concurrency limits
      const capSpec = machine.capabilities.find(
        (c) => c.name === capabilityName
      );
      if (
        capSpec?.maxConcurrency &&
        machine.activeTasks.length >= capSpec.maxConcurrency
      )
        continue;

      candidates.push(machine);
    }

    // Sort: prefer not-busy, then by free memory, then by load
    return candidates.sort((a, b) => {
      if (a.busy !== b.busy) return a.busy ? 1 : -1;
      const aLoad = a.resources.loadAvg?.[0] ?? 0;
      const bLoad = b.resources.loadAvg?.[0] ?? 0;
      return aLoad - bLoad;
    });
  }

  /**
   * Dispatch a request to the best available machine for a capability.
   */
  async dispatchToCapability(
    capabilityName: string,
    input: unknown,
    options?: { timeoutMs?: number; pipelineId?: string; stepIndex?: number }
  ): Promise<{ machineId: string; result: unknown }> {
    const candidates = this.findMachinesForCapability(capabilityName);

    if (candidates.length === 0) {
      throw new Error(
        `No machines available for capability "${capabilityName}"`
      );
    }

    const target = candidates[0];
    const ws = this.registry.connections.get(target.machineId);

    if (!ws) {
      throw new Error(
        `Machine "${target.machineId}" is registered but not connected`
      );
    }

    const requestId = crypto.randomUUID();

    return new Promise((resolve, reject) => {
      const timeout = setTimeout(() => {
        reject(
          new Error(
            `Timeout waiting for response from "${target.machineId}" ` +
              `(capability: ${capabilityName})`
          )
        );
      }, options?.timeoutMs ?? 30_000);

      // Store the resolver so the WebSocket message handler can call it
      this.pendingRequests.set(requestId, {
        resolve: (result: unknown) => {
          clearTimeout(timeout);
          resolve({ machineId: target.machineId, result });
        },
        reject: (error: Error) => {
          clearTimeout(timeout);
          reject(error);
        },
      });

      const message: DispatchRequest = {
        type: "request",
        id: requestId,
        capability: capabilityName,
        input,
        pipelineId: options?.pipelineId,
        stepIndex: options?.stepIndex,
        timeoutMs: options?.timeoutMs,
      };

      ws.send(JSON.stringify(message));
    });
  }

  private pendingRequests = new Map<
    string,
    {
      resolve: (result: unknown) => void;
      reject: (error: Error) => void;
    }
  >();

  /**
   * Remove a machine from the registry and update the capability index.
   */
  private deregisterMachine(machineId: string): void {
    const machine = this.registry.machines.get(machineId);
    if (!machine) return;

    // Remove from capability index
    for (const cap of machine.capabilities) {
      const machines = this.registry.capabilityIndex.get(cap.name);
      if (machines) {
        machines.delete(machineId);
        if (machines.size === 0) {
          this.registry.capabilityIndex.delete(cap.name);
        }
      }
    }

    // Close WebSocket if still open
    const ws = this.registry.connections.get(machineId);
    if (ws) {
      try {
        ws.close(1000, "deregistered");
      } catch {
        // ignore
      }
      this.registry.connections.delete(machineId);
    }

    this.registry.machines.delete(machineId);
  }
}

Connector-Side Handler

Here is how the connector processes incoming dispatch requests. This is a production-quality implementation based on the real Atlas connector:

import * as os from "node:os";
import WebSocket from "ws";
import { z } from "zod";

// ── Capability Registry ─────────────────────────────────────────────────────

type CapabilityResult = {
  success: boolean;
  data?: unknown;
  error?: string;
};

type CapabilityHandler = (input: unknown) => Promise<CapabilityResult>;

const CAPABILITY_HANDLERS: Map<string, CapabilityHandler> = new Map();

/**
 * Register a capability handler. Called at startup.
 */
function registerCapability(
  name: string,
  handler: CapabilityHandler
): void {
  CAPABILITY_HANDLERS.set(name, handler);
}

// ── Built-in Capabilities ───────────────────────────────────────────────────

registerCapability("jane-status", async () => ({
  success: true,
  data: {
    hostname: os.hostname(),
    platform: os.platform(),
    arch: os.arch(),
    uptimeSeconds: os.uptime(),
    cpu: {
      model: os.cpus()[0]?.model ?? "unknown",
      cores: os.cpus().length,
      loadAvg: os.loadavg(),
    },
    memory: {
      totalGb: +(os.totalmem() / 1e9).toFixed(1),
      freeGb: +(os.freemem() / 1e9).toFixed(1),
      usedPct: +(
        ((os.totalmem() - os.freemem()) / os.totalmem()) *
        100
      ).toFixed(1),
    },
  },
}));

registerCapability("hud-notify", async (input) => {
  const schema = z.object({
    severity: z.enum(["green", "yellow", "red"]),
    message: z.string(),
    source: z.string().default("connector"),
  });

  const parsed = schema.safeParse(input);
  if (!parsed.success) {
    return { success: false, error: parsed.error.message };
  }

  // Write to the HUD status queue file
  const fs = await import("node:fs/promises");
  const path = await import("node:path");
  const queuePath = path.join(os.homedir(), ".atlas", "status-queue.json");

  const msg = {
    id: crypto.randomUUID(),
    ...parsed.data,
    created: new Date().toISOString(),
  };

  let queue = { messages: [] as unknown[] };
  try {
    const raw = await fs.readFile(queuePath, "utf-8");
    queue = JSON.parse(raw);
  } catch {
    // File doesn't exist — start fresh
  }

  queue.messages.push(msg);
  await fs.writeFile(queuePath, JSON.stringify(queue, null, 2));

  return { success: true, data: { id: msg.id } };
});

// ── Request Handler ─────────────────────────────────────────────────────────

async function handleDispatch(
  ws: WebSocket,
  request: DispatchRequest,
  machineId: string
): Promise<void> {
  const handler = CAPABILITY_HANDLERS.get(request.capability);

  if (!handler) {
    ws.send(
      JSON.stringify({
        type: "response",
        id: request.id,
        status: 404,
        body: {
          error: "unknown_capability",
          message: `Capability "${request.capability}" not registered`,
          available: Array.from(CAPABILITY_HANDLERS.keys()),
        },
      })
    );
    return;
  }

  // Report busy
  ws.send(
    JSON.stringify({
      type: "status-report",
      machine: machineId,
      busy: true,
      version: "0.1.0",
    })
  );

  let result: CapabilityResult;
  const startTime = Date.now();

  try {
    // Apply timeout if specified
    if (request.timeoutMs) {
      result = await Promise.race([
        handler(request.input),
        new Promise<CapabilityResult>((_, reject) =>
          setTimeout(
            () => reject(new Error("Capability execution timed out")),
            request.timeoutMs!
          )
        ),
      ]);
    } else {
      result = await handler(request.input);
    }
  } catch (err) {
    result = {
      success: false,
      error: `Handler threw: ${err instanceof Error ? err.message : String(err)}`,
    };
  }

  const durationMs = Date.now() - startTime;

  // Send response
  ws.send(
    JSON.stringify({
      type: "response",
      id: request.id,
      status: result.success ? 200 : 500,
      body: { ...result, durationMs },
    })
  );

  // Report not busy
  ws.send(
    JSON.stringify({
      type: "status-report",
      machine: machineId,
      busy: false,
      version: "0.1.0",
    })
  );
}

How Systems Discover What Machines Can Do

Capability discovery is the process of building and maintaining an index of “what can be done and where.” Different systems take fundamentally different approaches.

Pattern 1: Central Registry (Consul, etcd)

Consul and etcd use a central key-value store as the source of truth. Machines register themselves, and consumers query the registry.

// Consul-style service registration
interface ConsulServiceRegistration {
  ID: string;
  Name: string;
  Tags: string[];
  Port: number;
  Meta: Record<string, string>;
  Check: {
    HTTP?: string;
    TCP?: string;
    Interval: string;
    Timeout: string;
    DeregisterCriticalServiceAfter: string;
  };
}

// Register a service with Consul
async function registerWithConsul(
  consulAddr: string,
  service: ConsulServiceRegistration
): Promise<void> {
  await fetch(`${consulAddr}/v1/agent/service/register`, {
    method: "PUT",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify(service),
  });
}

// Query capabilities from Consul
async function findCapableServices(
  consulAddr: string,
  capability: string
): Promise<ConsulServiceRegistration[]> {
  const res = await fetch(
    `${consulAddr}/v1/catalog/service/atlas-connector?tag=${capability}`
  );
  return res.json();
}

Trade-offs:

• Central point of truth — no split-brain
• Requires the registry to be available (SPoF without clustering)
• Health checks are poll-based (Consul) or TTL-based (etcd)
• Good for services that rarely change their capabilities

Pattern 2: Gossip Protocol (Consul Serf, Swim)

Consul itself uses a gossip protocol (Serf) for cluster membership and failure detection. Nodes communicate directly with each other, spreading membership information epidemically.

// Gossip-style peer discovery
interface GossipPeer {
  machineId: string;
  address: string;
  capabilities: string[];
  generation: number; // Lamport clock for conflict resolution
  status: "alive" | "suspect" | "dead";
  lastSeen: number;
}

class GossipMembership {
  private peers: Map<string, GossipPeer> = new Map();
  private self: GossipPeer;
  private fanout = 3; // Number of peers to gossip to per round

  constructor(selfPeer: GossipPeer) {
    this.self = selfPeer;
    this.peers.set(selfPeer.machineId, selfPeer);
  }

  /**
   * Gossip round: pick random peers and exchange membership lists.
   * Each round, every node talks to `fanout` other nodes.
   * Information propagates in O(log N) rounds.
   */
  async gossipRound(): Promise<void> {
    const alivePeers = Array.from(this.peers.values()).filter(
      (p) => p.status === "alive" && p.machineId !== this.self.machineId
    );

    // Pick random peers to gossip with
    const targets = this.pickRandom(alivePeers, this.fanout);

    for (const target of targets) {
      try {
        // Send our membership list, receive theirs
        const response = await this.sendGossip(target.address, {
          sender: this.self,
          members: Array.from(this.peers.values()),
        });

        // Merge received membership info
        for (const remotePeer of response.members) {
          this.mergePeer(remotePeer);
        }
      } catch {
        // Mark peer as suspect
        target.status = "suspect";
      }
    }
  }

  private mergePeer(remote: GossipPeer): void {
    const local = this.peers.get(remote.machineId);
    if (!local || remote.generation > local.generation) {
      this.peers.set(remote.machineId, remote);
    }
  }

  private pickRandom<T>(arr: T[], n: number): T[] {
    const shuffled = [...arr].sort(() => Math.random() - 0.5);
    return shuffled.slice(0, n);
  }

  private async sendGossip(
    address: string,
    payload: { sender: GossipPeer; members: GossipPeer[] }
  ): Promise<{ members: GossipPeer[] }> {
    const res = await fetch(`${address}/gossip`, {
      method: "POST",
      body: JSON.stringify(payload),
    });
    return res.json();
  }
}

Pattern 3: mDNS / Bonjour (Zero-Configuration Discovery)

For local network discovery, mDNS (multicast DNS) and DNS-SD (DNS Service Discovery) let machines find each other without any central registry. On macOS, this is Bonjour. On Linux, it is Avahi.

import { Bonjour } from "bonjour-service";

const bonjour = new Bonjour();

// Advertise this machine's capabilities on the local network
function advertiseCapabilities(
  machineId: string,
  capabilities: string[],
  port: number
): void {
  bonjour.publish({
    name: `atlas-connector-${machineId}`,
    type: "atlas-connector",
    protocol: "tcp",
    port,
    txt: {
      machineId,
      capabilities: capabilities.join(","),
      version: "0.1.0",
      role: "node",
    },
  });
}

// Discover other Atlas connectors on the local network
function discoverPeers(
  onFound: (peer: {
    machineId: string;
    host: string;
    port: number;
    capabilities: string[];
  }) => void
): void {
  bonjour.find({ type: "atlas-connector" }, (service) => {
    const txt = service.txt as Record<string, string>;
    onFound({
      machineId: txt.machineId ?? service.name,
      host: service.host,
      port: service.port,
      capabilities: (txt.capabilities ?? "").split(",").filter(Boolean),
    });
  });
}

// Usage
advertiseCapabilities("jane-macbook", ["hud-notify", "jane-status"], 7654);

discoverPeers((peer) => {
  console.log(
    `Found peer: ${peer.machineId} at ${peer.host}:${peer.port}`,
    `capabilities: ${peer.capabilities.join(", ")}`
  );
});

Key insight: mDNS is perfect for LAN-only discovery but does not cross network boundaries. For a system like Atlas that spans local machines and cloud, you need both: mDNS for discovering peers on the same network, and a central registry (the hub) for cross-network coordination. Peers discovered via mDNS should still register with the hub.

Pattern 4: Kubernetes Labels and Taints

Kubernetes uses node labels, taints, and tolerations for capability-based scheduling. Labels describe what a node has. Taints describe what a node rejects.

// Kubernetes-inspired label-based capability matching
interface MachineLabels {
  // Standard labels
  "atlas.dev/platform": string;      // "darwin", "linux"
  "atlas.dev/arch": string;          // "arm64", "amd64"
  "atlas.dev/role": string;          // "node", "hub"

  // Capability labels
  "atlas.dev/gpu": string;           // "true", "false"
  "atlas.dev/gpu-model"?: string;    // "m2-ultra", "rtx-4090"
  "atlas.dev/gpu-vram"?: string;     // "64", "24" (GB)
  "atlas.dev/display": string;       // "true", "false"
  "atlas.dev/inference": string;     // "ollama", "vllm", "none"

  // Custom labels
  [key: string]: string | undefined;
}

// Taints: machine rejects work unless the request tolerates it
interface MachineTaint {
  key: string;
  value: string;
  effect: "NoSchedule" | "PreferNoSchedule" | "NoExecute";
}

// Selector: request specifies what it needs
interface CapabilitySelector {
  /** Required labels (AND logic) */
  matchLabels?: Record<string, string>;
  /** Label expressions (more flexible matching) */
  matchExpressions?: {
    key: string;
    operator: "In" | "NotIn" | "Exists" | "DoesNotExist" | "Gt" | "Lt";
    values?: string[];
  }[];
  /** Tolerate specific taints */
  tolerations?: {
    key: string;
    operator: "Equal" | "Exists";
    value?: string;
    effect?: MachineTaint["effect"];
  }[];
}

// Example: find a machine with GPU and display for inference + visualization
const inferenceSelector: CapabilitySelector = {
  matchLabels: {
    "atlas.dev/gpu": "true",
    "atlas.dev/display": "true",
  },
  matchExpressions: [
    {
      key: "atlas.dev/gpu-vram",
      operator: "Gt",
      values: ["16"], // At least 16GB VRAM
    },
    {
      key: "atlas.dev/inference",
      operator: "In",
      values: ["ollama", "vllm"],
    },
  ],
};

function matchesMachine(
  machine: { labels: MachineLabels; taints?: MachineTaint[] },
  selector: CapabilitySelector
): boolean {
  // Check matchLabels
  if (selector.matchLabels) {
    for (const [key, value] of Object.entries(selector.matchLabels)) {
      if ((machine.labels as Record<string, string>)[key] !== value) {
        return false;
      }
    }
  }

  // Check matchExpressions
  if (selector.matchExpressions) {
    for (const expr of selector.matchExpressions) {
      const machineValue = (machine.labels as Record<string, string>)[
        expr.key
      ];
      switch (expr.operator) {
        case "In":
          if (!expr.values?.includes(machineValue ?? "")) return false;
          break;
        case "NotIn":
          if (expr.values?.includes(machineValue ?? "")) return false;
          break;
        case "Exists":
          if (machineValue === undefined) return false;
          break;
        case "DoesNotExist":
          if (machineValue !== undefined) return false;
          break;
        case "Gt":
          if (Number(machineValue) <= Number(expr.values?.[0])) return false;
          break;
        case "Lt":
          if (Number(machineValue) >= Number(expr.values?.[0])) return false;
          break;
      }
    }
  }

  // Check taints vs tolerations
  if (machine.taints) {
    for (const taint of machine.taints) {
      if (taint.effect === "NoSchedule" || taint.effect === "NoExecute") {
        const tolerated = selector.tolerations?.some(
          (t) =>
            t.key === taint.key &&
            (t.operator === "Exists" || t.value === taint.value)
        );
        if (!tolerated) return false;
      }
    }
  }

  return true;
}

Comparison: Discovery Approaches

For Atlas, the answer is a hybrid: mDNS for LAN peer discovery + cloud hub for global coordination. Machines on the same network find each other instantly via mDNS. All machines register with the cloud hub for cross-network routing and pipeline orchestration.

Node: The Worker

A node is a machine that registers capabilities and executes dispatched work. It does not make routing decisions — it receives requests and returns results.

class AtlasNode {
  private config: ConnectorConfig;
  private ws: WebSocket | null = null;
  private capabilities: Map<string, CapabilityHandler> = new Map();
  private reconnectAttempt = 0;
  private heartbeatTimer: ReturnType<typeof setInterval> | null = null;

  constructor(config: ConnectorConfig) {
    this.config = config;
  }

  /**
   * Register a capability handler on this node.
   */
  addCapability(name: string, handler: CapabilityHandler): void {
    this.capabilities.set(name, handler);
  }

  /**
   * Start the node: register with hub, connect WebSocket, begin heartbeat.
   */
  async start(): Promise<void> {
    await this.register();
    this.connect();
  }

  private async register(): Promise<void> {
    const capabilities = Array.from(this.capabilities.keys()).map((name) => ({
      name,
      spec: { description: `Capability: ${name}` },
    }));

    const body = {
      machine_id: this.config.machineId,
      label: this.config.machineLabel,
      endpoint: this.config.runnerWsUrl,
      capabilities,
      resources: {
        platform: os.platform(),
        arch: os.arch(),
        cpus: os.cpus().length,
        memory_gb: +(os.totalmem() / 1e9).toFixed(1),
      },
      ttl_seconds: this.config.registrationTtlSeconds,
    };

    const res = await fetch(
      `${this.config.apiMomUrl}/v1/machines/register`,
      {
        method: "POST",
        headers: {
          "Content-Type": "application/json",
          Authorization: `Bearer ${this.config.secret}`,
        },
        body: JSON.stringify(body),
      }
    );

    if (!res.ok) {
      throw new Error(`Registration failed: ${res.status}`);
    }
  }

  private connect(): void {
    this.ws = new WebSocket(this.config.runnerWsUrl, {
      headers: { Authorization: `Bearer ${this.config.secret}` },
    });

    this.ws.on("open", () => {
      this.reconnectAttempt = 0;
      this.startHeartbeat();
    });

    this.ws.on("message", (raw) => {
      const msg = JSON.parse(raw.toString());
      if (msg.type === "request") {
        this.handleRequest(msg);
      } else if (msg.type === "ping") {
        this.ws?.send(JSON.stringify({ type: "pong" }));
      }
    });

    this.ws.on("close", () => {
      this.stopHeartbeat();
      this.scheduleReconnect();
    });
  }

  private async handleRequest(msg: DispatchRequest): Promise<void> {
    const handler = this.capabilities.get(msg.capability);
    if (!handler) {
      this.ws?.send(
        JSON.stringify({
          type: "response",
          id: msg.id,
          status: 404,
          body: { error: "unknown_capability" },
        })
      );
      return;
    }

    const result = await handler(msg.input);
    this.ws?.send(
      JSON.stringify({
        type: "response",
        id: msg.id,
        status: result.success ? 200 : 500,
        body: result,
      })
    );
  }

  private startHeartbeat(): void {
    this.heartbeatTimer = setInterval(async () => {
      await fetch(`${this.config.apiMomUrl}/v1/machines/heartbeat`, {
        method: "POST",
        headers: {
          "Content-Type": "application/json",
          Authorization: `Bearer ${this.config.secret}`,
        },
        body: JSON.stringify({
          machine_id: this.config.machineId,
          capabilities: Array.from(this.capabilities.keys()),
          resources: {
            platform: os.platform(),
            cpus: os.cpus().length,
            memory_gb: +(os.totalmem() / 1e9).toFixed(1),
            free_memory_gb: +(os.freemem() / 1e9).toFixed(1),
            load_avg: os.loadavg(),
          },
        }),
      }).catch(() => {});
    }, this.config.heartbeatIntervalMs);
  }

  private stopHeartbeat(): void {
    if (this.heartbeatTimer) {
      clearInterval(this.heartbeatTimer);
      this.heartbeatTimer = null;
    }
  }

  private scheduleReconnect(): void {
    const delay = Math.min(1000 * 2 ** this.reconnectAttempt, 60_000);
    const jitter = delay * (0.8 + Math.random() * 0.4);
    this.reconnectAttempt++;
    setTimeout(() => this.connect(), jitter);
  }
}