diff --git a/.env.example b/.env.example index 13313f9..b91625d 100644 --- a/.env.example +++ b/.env.example @@ -81,6 +81,14 @@ NAVI_UI_MCP_HOST=127.0.0.1 NAVI_UI_MCP_PORT=8098 +# ── swarm / hive registry (see hive/README.md) ──────────────────────────────────── +# instance.json (generated by install.sh) holds this navi's name for the swarm. +# .swarm-key (generated by install.sh, gitignored) is the shared PSK. +# Leave HIVE_URL empty on a standalone machine — no announcements are made, +# and the registry being down never blocks anything. +#HIVE_URL=http://192.168.1.168:8087 +ANNOUNCE_INTERVAL_SEC=120 + # ── gnexus-auth OAuth ──────────────────────────────────────────────────────────── GNAUTH_BASE_URL=https://auth.your-domain.com GNAUTH_CLIENT_ID= diff --git a/.gitignore b/.gitignore index cab97ec..f210185 100644 --- a/.gitignore +++ b/.gitignore @@ -25,3 +25,7 @@ .python/ cpython-*.tar.gz docker-compose-linux-* +# swarm / instance identity +instance.json +.swarm-key +.swarm-key.previous diff --git a/deploy/env.template b/deploy/env.template index 6566625..0363f52 100644 --- a/deploy/env.template +++ b/deploy/env.template @@ -45,4 +45,9 @@ LOG_LEVEL=INFO SESSION_MESSAGES_WINDOW=1000 WS_REPLAY_BUFFER_SIZE=500 -CONTEXT_COMPRESSION_ENABLED=true \ No newline at end of file +CONTEXT_COMPRESSION_ENABLED=true +# ─── Swarm / hive registry (see hive/README.md) ───────────────────── +# Empty HIVE_URL = standalone machine: no announcements, nothing blocked. +# To join a swarm: copy .swarm-key from the hive host, uncomment, restart. +#HIVE_URL=http://192.168.1.168:8087 +ANNOUNCE_INTERVAL_SEC=120 diff --git a/deploy/install.sh b/deploy/install.sh index 6e1eca9..cdee0d4 100755 --- a/deploy/install.sh +++ b/deploy/install.sh @@ -173,6 +173,20 @@ ./.venv/bin/pip install -e . echo "Installed entry points: navi-server, navi-code" +# Instance identity (name shown to peers) + swarm key (shared PSK for the +# hive registry / peer protocol). A .swarm-key already in the repo root wins +# (scp'd from the hive host to join a swarm); otherwise a fresh one is +# generated — the machine stays standalone until HIVE_URL is set. +if [ ! -f instance.json ]; then + ./.venv/bin/python -c "from navi.identity import ensure_identity; ensure_identity()" + echo "Generated instance identity: $(./.venv/bin/python -c 'import json; print(json.load(open("instance.json"))["name"])')" +fi +if [ ! -f .swarm-key ]; then + ( umask 177; head -c 32 /dev/urandom | od -An -tx1 | tr -d ' \n' > .swarm-key ) + echo "Generated swarm key (.swarm-key). To join a swarm: copy the key from the hive host here, set HIVE_URL in .env, restart." +fi +chmod 600 .swarm-key 2>/dev/null || true + # ── 5. systemd unit ───────────────────────────────────────────────── say "Installing systemd unit (navi.service)" RUN_USER="$(id -un)" diff --git a/docs/config.md b/docs/config.md index d11860c..77ba49d 100644 --- a/docs/config.md +++ b/docs/config.md @@ -104,6 +104,21 @@ | `NAVI_HOST` | str | `127.0.0.1` | Bind address for the `navi-server` launcher (uvicorn). `127.0.0.1` = local-only; remote clients connect via SSH tunnel or reverse proxy. | | `NAVI_PORT` | int | `8099` | Bind port for the `navi-server` launcher. | +### Swarm / hive registry + +Navi instances can form a swarm: each announces itself to a tiny registry +service ("hive", see [`hive/README.md`](../hive/README.md)) running on the main +server; agents ask the book who is alive and where. The registry is never a +blocking dependency — with `HIVE_URL` empty, or with the hive down, everything +else works and the agent is simply told the book is unreachable. + +| Variable | Type | Default | Description | +|---|---|---|---| +| `HIVE_URL` | str | `""` | Hive registry address (e.g. `http://192.168.1.168:8087`). Empty = standalone navi, no announcements. | +| `ANNOUNCE_INTERVAL_SEC` | int | `120` | How often this navi re-announces itself to the hive. | +| `SWARM_KEY_FILE` | str | `.swarm-key` | Shared PSK file (gitignored, 0600, generated by `deploy/install.sh`). | +| `INSTANCE_FILE` | str | `instance.json` | This navi's identity for the swarm: generated name (adjective-animal) + instance id. Rename = edit the file, restart. | + See [`deploy/README.md`](../deploy/README.md) for the one-command server deployment (`bash deploy/install.sh`): dockerized PostgreSQL + systemd + `navi-code` in PATH, web UI and auth off by default. ## Logging diff --git a/docs/index.md b/docs/index.md index 71e30c7..4d3a63b 100644 --- a/docs/index.md +++ b/docs/index.md @@ -41,6 +41,7 @@ | [`permissions.md`](permissions.md) | Permission gate — authoritative backend confirmation for destructive tool calls (**design only, not yet implemented**) | | [`auth.md`](auth.md) | Auth: multi-user OAuth via gnexus-auth, or disable with `NAVI_AUTH_ENABLED=false` | | [`../deploy/README.md`](../deploy/README.md) | Server deployment — one command: dockerized postgres + systemd + navi-code, web UI off | +| [`../hive/README.md`](../hive/README.md) | hive — the swarm address book: machines announce themselves, agents ask who is alive | | [`config.md`](config.md) | All environment variables with types and defaults | | [`api.md`](api.md) | REST API endpoints + full WebSocket event schemas and sequences | | [`testing.md`](testing.md) | Test layout, fixtures, how to run the suite | diff --git a/hive/README.md b/hive/README.md new file mode 100644 index 0000000..a3f01c1 --- /dev/null +++ b/hive/README.md @@ -0,0 +1,61 @@ +# hive — the swarm address book + +A tiny standalone service for the **main server**. Every Navi instance in the +swarm announces itself here every couple of minutes (name, instance id, address, +machine info), and agents ask the book who is currently alive and where. The +free-form `meta` field is the seed of a future mini knowledge base about the +fleet (host, OS, cores, RAM — whatever the machine reports). + +The hive is **not** part of the Navi server and never starts by default — +install.sh installs nothing hive-related. It runs on one machine with a stable +address (e.g. the main server: local `192.168.1.168:8087`). + +## Run it + +On the main server, from the repo root (same clone, same venv — zero extra +dependencies): + +```bash +# quick check +.venv/bin/uvicorn hive.app:app --host 0.0.0.0 --port 8087 + +# as a systemd unit: +sed -e "s|EDIT_ME|$HOME|g" hive/hive.service | sudo tee /etc/systemd/system/hive.service +sudo systemctl daemon-reload +sudo systemctl enable --now hive +``` + +`--host 0.0.0.0` is required so other machines on the LAN can announce. + +## Auth + +Everything except `/health` requires the shared swarm key: an `X-Swarm-Key` +header matching `.swarm-key` in the working directory. `.swarm-key.previous` +is also accepted — that is the rotation window: put the new key in +`.swarm-key`, move the old one to `.swarm-key.previous`, then delete the old +file once every navi has picked up the new key. + +## Endpoints + +| Endpoint | Auth | What | +|---|---|---| +| `GET /health` | none | liveness + machine count | +| `POST /announce` | swarm key | upsert: `{name, instance_id, version, port, meta}` — the source IP is recorded as the machine's host (reliable on floating-IP networks), `port` comes from the payload | +| `GET /peers` | swarm key | `{peers: [{name, instance_id, host, port, address, online, last_seen, first_seen, meta}], ttl_sec}` — `online` = announced within the TTL | + +## Configuration (env vars) + +| Var | Default | Meaning | +|---|---|---| +| `HIVE_KEY_FILE` | `.swarm-key` | current swarm key file | +| `HIVE_DB` | `hive.db` | SQLite file | +| `HIVE_TTL_SEC` | `300` | a machine is `online` if it announced within this window | + +## Navi side + +A Navi announces itself when `.env` sets `HIVE_URL` (e.g. +`HIVE_URL=http://192.168.1.168:8087`) and `.swarm-key` exists next to +`.env` (install.sh generates one; copy the same file to every machine and the +hive host to form a swarm). No `HIVE_URL` — the navi is standalone and keeps +working; the registry being down is also non-blocking: the agent just gets a +context note that the book is unreachable. \ No newline at end of file diff --git a/hive/__init__.py b/hive/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/hive/__init__.py diff --git a/hive/app.py b/hive/app.py new file mode 100644 index 0000000..3ec6a61 --- /dev/null +++ b/hive/app.py @@ -0,0 +1,99 @@ +"""hive — the swarm address book. + +A tiny standalone service for the main server: Navi instances across the +network announce themselves here (name, uuid, address, machine info), and +agents ask this book who is currently alive and where. Later this grows into +a mini knowledge base about the fleet (the free-form ``meta`` field is the seed). + +Not part of the Navi server — it never starts by default. Run it manually on +the main server (see hive/README.md): + + .venv/bin/uvicorn hive.app:app --host 0.0.0.0 --port 8087 + +Auth: the shared swarm key (PSK). Every announce/peers request must carry the +X-Swarm-Key header matching ``.swarm-key`` (or ``.swarm-key.previous`` — the +rotation window). /health is open. +""" + +from __future__ import annotations + +import hmac +import os +from pathlib import Path + +from fastapi import FastAPI, Header, HTTPException, Request +from pydantic import BaseModel, Field + +from hive.db import HiveDB + +_KEY_FILE = Path(os.environ.get("HIVE_KEY_FILE", ".swarm-key")) +# Rotation window: the previous key keeps working until it is deleted. +_PREVIOUS_KEY_FILE = _KEY_FILE.with_name(_KEY_FILE.name + ".previous") +_DB_PATH = os.environ.get("HIVE_DB", "hive.db") +_TTL_SEC = int(os.environ.get("HIVE_TTL_SEC", "300")) + +app = FastAPI(title="navi hive", description="Swarm address book for Navi instances.") +_db = HiveDB(_DB_PATH) + + +def _load_keys() -> list[str]: + keys: list[str] = [] + for p in (_KEY_FILE, _PREVIOUS_KEY_FILE): + try: + value = p.read_text(encoding="utf-8").strip() + except OSError: + continue + if value: + keys.append(value) + return keys + + +def _check_key(x_swarm_key: str | None) -> None: + keys = _load_keys() + if not keys: + raise HTTPException(status_code=503, detail="hive has no swarm key configured") + if not x_swarm_key: + raise HTTPException(status_code=401, detail="X-Swarm-Key header required") + if not any(hmac.compare_digest(x_swarm_key, k) for k in keys): + raise HTTPException(status_code=403, detail="invalid swarm key") + + +class AnnouncePayload(BaseModel): + name: str = Field(min_length=1, max_length=64) + instance_id: str = Field(min_length=8, max_length=64) + version: str = Field(default="", max_length=32) + port: int = Field(ge=1, le=65535) + meta: dict = Field(default_factory=dict) + + +@app.get("/health") +async def health() -> dict: + return {"status": "ok", "hive": True, "machines": len(_db.list_peers(_TTL_SEC))} + + +@app.post("/announce") +async def announce( + payload: AnnouncePayload, + request: Request, + x_swarm_key: str | None = Header(default=None), +) -> dict: + _check_key(x_swarm_key) + # The source IP is the reliable address for a floating-IP network; the + # announced payload only supplies the port (and could not be trusted for + # the host anyway — anyone with the key may announce from anywhere). + host = request.client.host if request.client else None + _db.upsert( + instance_id=payload.instance_id, + name=payload.name, + host=host, + port=payload.port, + version=payload.version, + meta=payload.meta, + ) + return {"ok": True, "host": host} + + +@app.get("/peers") +async def peers(x_swarm_key: str | None = Header(default=None)) -> dict: + _check_key(x_swarm_key) + return {"peers": _db.list_peers(_TTL_SEC), "ttl_sec": _TTL_SEC} \ No newline at end of file diff --git a/hive/db.py b/hive/db.py new file mode 100644 index 0000000..31096b0 --- /dev/null +++ b/hive/db.py @@ -0,0 +1,103 @@ +"""SQLite storage for the hive (swarm address book). + +Single-writer, single-process, tiny — SQLite is deliberately enough here +(the hive is a notebook, not a Navi database). ``online`` is derived at read +time: a machine is considered present if it announced within the TTL window. +""" + +from __future__ import annotations + +import json +import sqlite3 +import time +from pathlib import Path + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS machines ( + instance_id TEXT PRIMARY KEY, + name TEXT NOT NULL, + host TEXT, + port INTEGER, + version TEXT, + meta TEXT, + first_seen REAL NOT NULL, + last_seen REAL NOT NULL +); +""" + + +class HiveDB: + def __init__(self, path: str | Path = "hive.db") -> None: + self._path = str(path) + self._conn = sqlite3.connect(self._path, check_same_thread=False) + self._conn.row_factory = sqlite3.Row + self._conn.execute("PRAGMA journal_mode=WAL") + self._conn.executescript(_SCHEMA) + self._conn.commit() + + def close(self) -> None: + self._conn.close() + + def upsert( + self, + *, + instance_id: str, + name: str, + host: str | None, + port: int, + version: str, + meta: dict, + ) -> None: + now = time.time() + self._conn.execute( + """ + INSERT INTO machines + (instance_id, name, host, port, version, meta, first_seen, last_seen) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(instance_id) DO UPDATE SET + name = excluded.name, + host = COALESCE(excluded.host, machines.host), + port = excluded.port, + version = excluded.version, + meta = excluded.meta, + last_seen = excluded.last_seen + """, + ( + instance_id, + name, + host, + port, + version, + json.dumps(meta), + now, + now, + ), + ) + self._conn.commit() + + def list_peers(self, ttl_sec: int) -> list[dict]: + now = time.time() + rows = self._conn.execute( + "SELECT * FROM machines ORDER BY name COLLATE NOCASE" + ).fetchall() + peers = [] + for row in rows: + try: + meta = json.loads(row["meta"] or "{}") + except json.JSONDecodeError: + meta = {} + peers.append( + { + "name": row["name"], + "instance_id": row["instance_id"], + "host": row["host"], + "port": row["port"], + "address": f"{row['host']}:{row['port']}" if row["host"] else None, + "version": row["version"], + "online": (now - row["last_seen"]) <= ttl_sec, + "first_seen": row["first_seen"], + "last_seen": row["last_seen"], + "meta": meta, + } + ) + return peers \ No newline at end of file diff --git a/hive/hive.service b/hive/hive.service new file mode 100644 index 0000000..a1e0757 --- /dev/null +++ b/hive/hive.service @@ -0,0 +1,17 @@ +[Unit] +Description=Navi hive (swarm address book) +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple +# Edit before installing: the user that owns the repo clone and the clone path. +User=EDIT_ME +WorkingDirectory=EDIT_ME/navi-1 +Environment=PYTHONIOENCODING=utf-8 +ExecStart=EDIT_ME/navi-1/.venv/bin/uvicorn hive.app:app --host 0.0.0.0 --port 8087 +Restart=always +RestartSec=3 + +[Install] +WantedBy=multi-user.target \ No newline at end of file diff --git a/navi/api/routes/health.py b/navi/api/routes/health.py index e62489b..d1a0da4 100644 --- a/navi/api/routes/health.py +++ b/navi/api/routes/health.py @@ -1,7 +1,6 @@ from fastapi import APIRouter from navi.config import settings -from navi.llm.ollama import OllamaBackend router = APIRouter(tags=["health"]) @@ -12,9 +11,22 @@ return { "status": "ok", "embed": embed_status, + "hive": _hive_status(), } +def _hive_status() -> dict: + """Swarm registry (hive) reachability as seen by the announce loop.""" + if not settings.hive_url: + return {"configured": False} + from navi.swarm import get_announcer + + announcer = get_announcer() + if announcer is None: + return {"configured": False, "error": "swarm key missing - announce disabled"} + return announcer.get_status() + + @router.get("/health/embed") async def health_embed() -> dict: return await _check_embed() diff --git a/navi/config.py b/navi/config.py index 1746138..b34b8b2 100644 --- a/navi/config.py +++ b/navi/config.py @@ -127,6 +127,14 @@ navi_ui_mcp_host: str = "127.0.0.1" navi_ui_mcp_port: int = 8098 + # Swarm identity + hive registry (see navi/identity.py and hive/). + # instance_file/swarm_key_file live in the working directory, gitignored. + # HIVE_URL empty = this navi does not announce itself anywhere (standalone). + instance_file: str = "instance.json" + swarm_key_file: str = ".swarm-key" + hive_url: str = "" + announce_interval_sec: int = 120 + # Auth session cookie encryption (Fernet key, 32-byte base64) navi_auth_encryption_key: str = "" navi_auth_cookie_name: str = "navi_auth_session" diff --git a/navi/context_providers/hive_status.py b/navi/context_providers/hive_status.py new file mode 100644 index 0000000..f5608c4 --- /dev/null +++ b/navi/context_providers/hive_status.py @@ -0,0 +1,33 @@ +"""Reports hive registry (swarm address book) reachability into LLM context. + +Silent while the registry is reachable (token economy); when it is down the +agent is told explicitly — this system is supposed to be able to fix its own +network, so it must notice a fallen registry instead of silently ignoring it. +""" + +from navi.config import settings + +name = "hive_status" +description = "Injects hive registry reachability (swarm address book) status." +global_provider = True + + +async def get_context() -> str | None: + if not settings.hive_url: + return None # standalone navi — no registry, nothing to report + from navi.swarm import get_announcer + + announcer = get_announcer() + if announcer is None: + return None # HIVE_URL set but key missing — announce disabled; already warned in logs + status = announcer.get_status() + if status.get("reachable") is True: + return None # healthy: stay out of the context + if status.get("reachable") is None: + return None # hasn't ticked yet (startup moment); not a failure + return ( + f"[System] Hive registry ({status['url']}) is unreachable " + f"since {status['down_since']}: {status['last_error']}. " + "Other navi agents cannot discover this machine while the registry is down. " + "If the current task depends on the swarm, investigate the registry host." + ) \ No newline at end of file diff --git a/navi/core/context_builder.py b/navi/core/context_builder.py index 60271cc..e3bb65f 100644 --- a/navi/core/context_builder.py +++ b/navi/core/context_builder.py @@ -98,6 +98,9 @@ return cached parts: list[str] = [] + identity = self._instance_identity() + if identity is not None: + parts.append(identity) persona = _config.settings.navi_persona.strip() if persona: parts.append(persona) @@ -130,6 +133,24 @@ else: self._system_prompt_cache.pop(profile_id, None) + def _instance_identity(self) -> str | None: + """The 'you are navi on host X' block for the system prompt.""" + import socket + + try: + from navi.identity import load_identity + + ident = load_identity(_config.settings.instance_file) + except Exception: + return None + lines = [ + f"You are navi '{ident.name}' (instance {ident.instance_id[:8]}) " + f"running on host {socket.gethostname()}.", + "Other navi instances in the swarm know this machine by that name — " + "identify yourself as '" + ident.name + "' when talking to remote peers.", + ] + return "\n".join(lines) + async def _memory_msg(self, user_id: str | None = None) -> "Message | None": if self._memory is None: return None diff --git a/navi/identity.py b/navi/identity.py new file mode 100644 index 0000000..d21e7fb --- /dev/null +++ b/navi/identity.py @@ -0,0 +1,94 @@ +"""Instance identity — a stable, human-friendly name for this Navi install. + +Every deployed Navi carries ``instance.json`` in its working directory +(gitignored): ``{"name": "quiet-otter", "instance_id": "", "created_at": "..."}``. +The name is generated once (adjective + animal — readable, speakable, collision- +free at household-network scale) and is meant to be renamed by hand if the owner +wants. The instance_id is the technical identity for audit and peer pinning. + +Rename = edit the file, restart the server. Nothing else reads the name as +long-lived state, so this is safe. +""" + +from __future__ import annotations + +import json +import uuid as _uuid +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path + +_ADJECTIVES = [ + "quiet", "amber", "hollow", "silver", "brisk", "cobalt", "dawn", "ember", + "frost", "gentle", "harbor", "ivory", "jasper", "keen", "lucid", "mellow", + "nimble", "opal", "pale", "quartz", "rapid", "soft", "tidal", "umber", + "velvet", "wan", "crimson", "golden", "misty", "noble", "olive", "plum", + "royal", "slate", "topaz", "vivid", "wild", "yellow", "zesty", "bold", + "calm", "deep", "eager", "flint", "gray", "hasty", "iron", "jolly", + "kind", "lunar", +] + +_ANIMALS = [ + "otter", "falcon", "heron", "moth", "badger", "cormorant", "dingo", "elk", + "fox", "grouse", "hare", "ibex", "jackal", "koi", "lynx", "marten", + "newt", "osprey", "puffin", "quail", "raven", "stoat", "tapir", "urchin", + "vole", "wren", "yak", "zephyr", "beagle", "caribou", "dolphin", "ermine", + "ferret", "gecko", "haddock", "ibis", "jaguar", "kestrel", "lemur", "mink", + "narwhal", "ocelot", "penguin", "quokka", "robin", "seal", "thrush", + "umbra", "vicuna", "wallaby", +] + +IDENTITY_FILE = "instance.json" + + +@dataclass(frozen=True) +class InstanceIdentity: + name: str + instance_id: str + created_at: str + + +def generate_name() -> str: + """Random adjective-animal pair, e.g. ``quiet-otter``.""" + import secrets + + return f"{secrets.choice(_ADJECTIVES)}-{secrets.choice(_ANIMALS)}" + + +def load_identity(path: str | Path = IDENTITY_FILE) -> InstanceIdentity: + """Read instance.json. Raises if the file is missing or malformed.""" + data = json.loads(Path(path).read_text(encoding="utf-8")) + return InstanceIdentity( + name=str(data["name"]), + instance_id=str(data["instance_id"]), + created_at=str(data.get("created_at", "")), + ) + + +def ensure_identity(path: str | Path = IDENTITY_FILE) -> InstanceIdentity: + """Load the identity, creating the file on first run (install / boot).""" + p = Path(path) + if p.exists(): + return load_identity(p) + identity = InstanceIdentity( + name=generate_name(), + instance_id=str(_uuid.uuid4()), + created_at=datetime.now(timezone.utc).isoformat(), + ) + p.write_text( + json.dumps( + { + "name": identity.name, + "instance_id": identity.instance_id, + "created_at": identity.created_at, + }, + indent=2, + ) + + "\n", + encoding="utf-8", + ) + try: + p.chmod(0o644) + except OSError: + pass + return identity \ No newline at end of file diff --git a/navi/main.py b/navi/main.py index bfd909d..13b321d 100644 --- a/navi/main.py +++ b/navi/main.py @@ -161,9 +161,35 @@ ) pending_sweep_task = asyncio.create_task(pending_sweep_loop(container.session_store)) + # Swarm announce loop — tell the hive registry (address book) this navi + # exists. Never a blocking dependency: no key/no HIVE_URL = no task at all, + # and the loop itself survives any number of failed ticks. + from navi.identity import ensure_identity + from navi.swarm import build_announcer_from_settings, set_announcer + + _identity = ensure_identity(settings.instance_file) + log.info( + "startup.instance_identity", + name=_identity.name, + instance_id=_identity.instance_id, + ) + announce_task: asyncio.Task | None = None + if settings.hive_url: + announcer = build_announcer_from_settings() + if announcer is not None: + set_announcer(announcer) + announce_task = asyncio.create_task(announcer.run()) + yield # Shutdown + if announce_task is not None: + announce_task.cancel() + try: + await announce_task + except asyncio.CancelledError: + pass + set_announcer(None) scheduler_task.cancel() try: await scheduler_task diff --git a/navi/swarm.py b/navi/swarm.py new file mode 100644 index 0000000..891a4f2 --- /dev/null +++ b/navi/swarm.py @@ -0,0 +1,198 @@ +"""Swarm announce loop — this Navi's heartbeat to the hive registry. + +The hive (see ``hive/``) is the swarm's address book: a tiny standalone service +on the main server. Each Navi announces itself (name, uuid, address, machine +info) every ANNOUNCE_INTERVAL_SEC seconds. + +Design rules: + - The hive is NEVER a blocking dependency. Startup, agent turns and all + tools work with the hive down; the loop just keeps trying on its next tick. + - The Navi must KNOW about reachability, though: state transitions are logged + (up→down, down→up — not every tick), /health carries the status, and the + ``hive_status`` context provider tells the agent when the registry is down + so it can act (this system is supposed to fix its own network). +""" + +from __future__ import annotations + +import asyncio +import os +import platform +import socket +from datetime import datetime, timezone +from pathlib import Path + +import httpx +import structlog + +from navi import __version__ +from navi.identity import InstanceIdentity + +log = structlog.get_logger() + + +def read_swarm_key(path: str | Path) -> str | None: + """The swarm PSK from a key file, or None if the file is missing/empty. + + A ``.swarm-key.previous`` file next to the main key is NOT read here — the + rotation window applies to the receiving side (hive/peer endpoints accept + both); the announcer always sends the current key. + """ + try: + value = Path(path).read_text(encoding="utf-8").strip() + except OSError: + return None + return value or None + + +def _machine_meta() -> dict: + """Basic machine facts for the address book (the mini knowledge base).""" + meta: dict = { + "hostname": socket.gethostname(), + "os": f"{platform.system()} {platform.release()}", + "cpu_cores": os.cpu_count(), + } + try: # Linux-only, cheap; absent fields are fine + with open("/proc/meminfo") as f: + for line in f: + if line.startswith("MemTotal:"): + meta["ram_mb"] = round(int(line.split()[1]) / 1024) + break + except OSError: + pass + return meta + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +class HiveAnnouncer: + """Periodic announce loop with reachable/unreachable state tracking.""" + + def __init__( + self, + hive_url: str, + key: str, + identity: InstanceIdentity, + port: int, + interval: int = 120, + client: httpx.AsyncClient | None = None, + ) -> None: + self._hive_url = hive_url.rstrip("/") + self._key = key + self._identity = identity + self._port = port + self._interval = interval + self._client = client # injectable for tests; created lazily otherwise + # state + self._reachable: bool | None = None # None = never announced yet + self._last_success: str | None = None + self._last_error: str | None = None + self._down_since: str | None = None + self._consecutive_failures = 0 + + @property + def payload(self) -> dict: + return { + "name": self._identity.name, + "instance_id": self._identity.instance_id, + "version": __version__, + "port": self._port, + "meta": _machine_meta(), + } + + async def announce_once(self) -> bool: + """One announce attempt; updates and logs state transitions.""" + client = self._client or httpx.AsyncClient(timeout=5.0) + try: + resp = await client.post( + f"{self._hive_url}/announce", + json=self.payload, + headers={"X-Swarm-Key": self._key}, + ) + resp.raise_for_status() + except Exception as e: + self._consecutive_failures += 1 + self._last_error = f"{type(e).__name__}: {e}" + if self._reachable is not False: + self._down_since = _now_iso() + log.warning( + "hive.announce_failed", + hive=self._hive_url, + error=self._last_error, + hint="registry unreachable - other agents cannot discover this navi", + ) + self._reachable = False + return False + finally: + if self._client is None: + await client.aclose() + + if self._reachable is not True: + log.info("hive.announce_recovered" if self._reachable is False else "hive.announce_ok", hive=self._hive_url) + self._reachable = True + self._last_success = _now_iso() + self._last_error = None + self._down_since = None + self._consecutive_failures = 0 + return True + + async def run(self) -> None: + """Loop forever; the first announce fires immediately at startup.""" + log.info("hive.announce_started", hive=self._hive_url, interval=self._interval) + while True: + try: + await self.announce_once() + except asyncio.CancelledError: + raise + except Exception: # never die — the loop outlives any single error + log.exception("hive.announce_loop_error") + await asyncio.sleep(self._interval) + + def get_status(self) -> dict: + return { + "configured": True, + "url": self._hive_url, + "reachable": self._reachable, + "last_success": self._last_success, + "last_error": self._last_error, + "down_since": self._down_since, + "consecutive_failures": self._consecutive_failures, + } + + +# ── process-wide instance (lifespan creates it; provider/health read it) ── + +_announcer: HiveAnnouncer | None = None + + +def set_announcer(announcer: HiveAnnouncer | None) -> None: + global _announcer + _announcer = announcer + + +def get_announcer() -> HiveAnnouncer | None: + return _announcer + + +def build_announcer_from_settings() -> HiveAnnouncer | None: + """Assemble the announcer from config; None (with a warning) if misconfigured.""" + from navi.config import settings + from navi.identity import ensure_identity + + key = read_swarm_key(settings.swarm_key_file) + if not key: + log.warning( + "hive.key_missing", + key_file=settings.swarm_key_file, + hint="HIVE_URL is set but the swarm key file is absent - announce disabled", + ) + return None + return HiveAnnouncer( + hive_url=settings.hive_url, + key=key, + identity=ensure_identity(settings.instance_file), + port=settings.navi_port, + interval=settings.announce_interval_sec, + ) \ No newline at end of file diff --git a/tests/unit/test_hive.py b/tests/unit/test_hive.py new file mode 100644 index 0000000..fa0e778 --- /dev/null +++ b/tests/unit/test_hive.py @@ -0,0 +1,128 @@ +"""hive — the swarm address book service (see hive/app.py). + +The app module reads env config at import time, so each test reimports it with +fresh env into its own key file / database. +""" + +import importlib +import sqlite3 + +import pytest +from fastapi.testclient import TestClient + + +@pytest.fixture() +def hive(tmp_path, monkeypatch): + monkeypatch.setenv("HIVE_KEY_FILE", str(tmp_path / ".swarm-key")) + monkeypatch.setenv("HIVE_DB", str(tmp_path / "hive.db")) + monkeypatch.setenv("HIVE_TTL_SEC", "300") + (tmp_path / ".swarm-key").write_text("secret-key-123") + import hive.app as hive_app + + module = importlib.reload(hive_app) + yield module + module._db.close() + + +def _headers(key: str = "secret-key-123") -> dict: + return {"X-Swarm-Key": key} + + +def _announce(client, name="quiet-otter", instance_id="11111111-2222-3333-4444-555555555555", port=8099): + return client.post( + "/announce", + json={"name": name, "instance_id": instance_id, "version": "0.1.0", "port": port, + "meta": {"hostname": "gpu-box", "os": "Linux 6.1", "cpu_cores": 16, "ram_mb": 65536}}, + headers=_headers(), + ) + + +def test_health_open_without_key(hive): + with TestClient(hive.app) as client: + resp = client.get("/health") + assert resp.status_code == 200 + assert resp.json()["hive"] is True + + +def test_announce_requires_key(hive): + with TestClient(hive.app) as client: + assert client.post("/announce", json={"name": "x", "instance_id": "12345678", "port": 8099}).status_code == 401 + assert client.post( + "/announce", + json={"name": "x", "instance_id": "12345678", "port": 8099}, + headers={"X-Swarm-Key": "wrong"}, + ).status_code == 403 + + +def test_peers_requires_key(hive): + with TestClient(hive.app) as client: + assert client.get("/peers").status_code == 401 + assert client.get("/peers", headers={"X-Swarm-Key": "wrong"}).status_code == 403 + + +def test_announce_upsert_and_peers_listing(hive): + with TestClient(hive.app) as client: + assert _announce(client).status_code == 200 + # same instance_id again = upsert (renamed, different port), not a second row + assert _announce(client, name="renamed-otter", port=8123).status_code == 200 + # a second machine + assert _announce(client, name="amber-falcon", instance_id="99999999-2222-3333-4444-555555555555").status_code == 200 + + resp = client.get("/peers", headers=_headers()) + assert resp.status_code == 200 + body = resp.json() + peers = body["peers"] + assert len(peers) == 2 + assert [p["name"] for p in peers] == ["amber-falcon", "renamed-otter"] # sorted by name + otter = peers[1] + assert otter["port"] == 8123 + assert otter["address"] == f"{otter['host']}:8123" + assert otter["online"] is True + assert otter["meta"]["hostname"] == "gpu-box" + assert otter["meta"]["ram_mb"] == 65536 + assert otter["version"] == "0.1.0" + assert body["ttl_sec"] == 300 + + +def test_offline_after_ttl(hive, tmp_path): + with TestClient(hive.app) as client: + _announce(client) + # backdate the last announcement beyond the TTL — machine went quiet + conn = sqlite3.connect(tmp_path / "hive.db") + conn.execute("UPDATE machines SET last_seen = last_seen - 400") + conn.commit() + conn.close() + peers = client.get("/peers", headers=_headers()).json()["peers"] + assert peers[0]["online"] is False + + +def test_previous_key_still_accepted(hive, tmp_path): + # rotation window: .swarm-key.previous keeps working + (tmp_path / ".swarm-key.previous").write_text("old-key-42") + import hive.app as module + + importlib.reload(module) + with TestClient(module.app) as client: + resp = client.post( + "/announce", + json={"name": "x", "instance_id": "12345678", "port": 8099}, + headers={"X-Swarm-Key": "old-key-42"}, + ) + assert resp.status_code == 200 + module._db.close() + + +def test_hive_without_key_file_refuses(hive, tmp_path, monkeypatch): + monkeypatch.setenv("HIVE_KEY_FILE", str(tmp_path / ".no-key")) + monkeypatch.setenv("HIVE_DB", str(tmp_path / "hive2.db")) + import hive.app as module + + importlib.reload(module) + with TestClient(module.app) as client: + resp = client.post( + "/announce", + json={"name": "x", "instance_id": "12345678", "port": 8099}, + headers={"X-Swarm-Key": "whatever"}, + ) + assert resp.status_code == 503 + module._db.close() \ No newline at end of file diff --git a/tests/unit/test_identity.py b/tests/unit/test_identity.py new file mode 100644 index 0000000..9959afe --- /dev/null +++ b/tests/unit/test_identity.py @@ -0,0 +1,57 @@ +"""Instance identity: generation, persistence, rename-by-edit.""" + +import json + +import pytest + +from navi.identity import ( + InstanceIdentity, + ensure_identity, + generate_name, + load_identity, +) + + +def test_generated_name_is_adjective_animal(): + for _ in range(50): + name = generate_name() + assert "-" in name + adj, animal = name.split("-", 1) + assert adj.isalpha() and animal.isalpha() + assert adj == adj.lower() + + +def test_generated_names_vary(): + names = {generate_name() for _ in range(80)} + assert len(names) > 1 + + +def test_ensure_creates_file_and_is_stable(tmp_path): + path = tmp_path / "instance.json" + first = ensure_identity(path) + assert path.exists() + second = load_identity(path) + assert first.name == second.name + assert first.instance_id == second.instance_id + # idempotent: ensure on an existing file never regenerates + third = ensure_identity(path) + assert third.instance_id == first.instance_id + + +def test_rename_by_editing_file(tmp_path): + path = tmp_path / "instance.json" + ensure_identity(path) + data = json.loads(path.read_text()) + data["name"] = "renamed-by-hand" + path.write_text(json.dumps(data)) + assert load_identity(path).name == "renamed-by-hand" + + +def test_load_missing_file_raises(tmp_path): + with pytest.raises(Exception): + load_identity(tmp_path / "absent.json") + + +def test_identity_dataclass_fields(): + ident = InstanceIdentity(name="quiet-otter", instance_id="uuid-x", created_at="t") + assert ident.name == "quiet-otter" \ No newline at end of file diff --git a/tests/unit/test_swarm.py b/tests/unit/test_swarm.py new file mode 100644 index 0000000..8595804 --- /dev/null +++ b/tests/unit/test_swarm.py @@ -0,0 +1,151 @@ +"""Swarm announce loop: PSK reading, state transitions, context provider.""" + +import httpx +import pytest + +from navi.identity import InstanceIdentity +from navi.swarm import HiveAnnouncer, read_swarm_key + +IDENT = InstanceIdentity(name="quiet-otter", instance_id="11111111-2222-3333-4444-555555555555", created_at="") + + +def _announcer(transport) -> HiveAnnouncer: + client = httpx.AsyncClient(transport=transport, base_url="http://hive.test") + return HiveAnnouncer( + hive_url="http://hive.test", + key="secret", + identity=IDENT, + port=8099, + interval=120, + client=client, + ) + + +def test_read_swarm_key(tmp_path): + missing = read_swarm_key(tmp_path / "absent") + assert missing is None + empty = tmp_path / "empty" + empty.write_text(" \n") + assert read_swarm_key(empty) is None + good = tmp_path / ".swarm-key" + good.write_text(" deadbeef \n") + assert read_swarm_key(good) == "deadbeef" + + +@pytest.mark.asyncio +async def test_announce_success_marks_reachable(): + seen = {} + + def handler(request: httpx.Request) -> httpx.Response: + import json + + seen["url"] = str(request.url) + seen["key"] = request.headers.get("X-Swarm-Key") + seen["json"] = json.loads(request.content) + return httpx.Response(200, json={"ok": True}) + + announcer = _announcer(httpx.MockTransport(handler)) + ok = await announcer.announce_once() + assert ok is True + status = announcer.get_status() + assert status["reachable"] is True + assert status["down_since"] is None + assert status["last_error"] is None + # request shape + assert seen["url"] == "http://hive.test/announce" + assert seen["key"] == "secret" + assert seen["json"]["name"] == "quiet-otter" + assert seen["json"]["port"] == 8099 + assert seen["json"]["instance_id"] == IDENT.instance_id + assert seen["json"]["meta"]["hostname"] + + +@pytest.mark.asyncio +async def test_announce_failure_sets_down_state_and_keeps_it(): + calls = {"n": 0} + + def handler(request: httpx.Request) -> httpx.Response: + calls["n"] += 1 + raise httpx.ConnectError("connection refused") + + announcer = _announcer(httpx.MockTransport(handler)) + assert await announcer.announce_once() is False + status = announcer.get_status() + assert status["reachable"] is False + assert status["down_since"] is not None + assert "ConnectError" in status["last_error"] + assert status["consecutive_failures"] == 1 + + # second failure: down_since does not shift (no re-log of transition) + await announcer.announce_once() + assert announcer.get_status()["down_since"] == status["down_since"] + assert announcer.get_status()["consecutive_failures"] == 2 + + +@pytest.mark.asyncio +async def test_announce_recovers_after_failure(): + fail = {"on": True} + + def handler(request: httpx.Request) -> httpx.Response: + if fail["on"]: + raise httpx.ConnectError("down") + return httpx.Response(200, json={"ok": True}) + + announcer = _announcer(httpx.MockTransport(handler)) + await announcer.announce_once() + assert announcer.get_status()["reachable"] is False + + fail["on"] = False + assert await announcer.announce_once() is True + status = announcer.get_status() + assert status["reachable"] is True + assert status["down_since"] is None + assert status["consecutive_failures"] == 0 + + +@pytest.mark.asyncio +async def test_hive_status_provider_silent_when_healthy_or_unconfigured(monkeypatch): + from types import SimpleNamespace + + from navi.context_providers import hive_status as provider + from navi import swarm + + # unconfigured: no context at all + monkeypatch.setattr(provider, "settings", SimpleNamespace(hive_url="")) + assert await provider.get_context() is None + + # configured + healthy: silent (token economy) + monkeypatch.setattr(provider, "settings", SimpleNamespace(hive_url="http://hive.test")) + + class FakeAnnouncer: + def get_status(self): + return {"reachable": True, "url": "http://hive.test", "down_since": None, "last_error": None} + + monkeypatch.setattr(swarm, "get_announcer", lambda: FakeAnnouncer()) + assert await provider.get_context() is None + + +@pytest.mark.asyncio +async def test_hive_status_provider_reports_outage(monkeypatch): + from types import SimpleNamespace + + from navi.context_providers import hive_status as provider + from navi import swarm + + monkeypatch.setattr(provider, "settings", SimpleNamespace(hive_url="http://hive.test")) + + class FakeAnnouncer: + def get_status(self): + return { + "reachable": False, + "url": "http://hive.test", + "down_since": "2026-09-10T12:00:00+00:00", + "last_error": "ConnectError: connection refused", + } + + monkeypatch.setattr(swarm, "get_announcer", lambda: FakeAnnouncer()) + text = await provider.get_context() + assert text is not None + assert "unreachable" in text + assert "http://hive.test" in text + assert "connection refused" in text \ No newline at end of file