diff --git a/hive/README.md b/hive/README.md index e6d1453..da3b429 100644 --- a/hive/README.md +++ b/hive/README.md @@ -39,9 +39,60 @@ | 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 /health` | none | liveness + machine count + current config version | +| `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. The reply may carry `config: {fetch, version, hash}` — that is how an instance that was offline learns it is behind | | `GET /peers` | swarm key | `{peers: [{name, instance_id, host, port, address, online, last_seen, first_seen, meta}], ttl_sec}` — `online` = announced within the TTL | +| `GET /config` | swarm key | the layers for the asking instance (`?name=&instance_id=&tags=a,b`); `404` = nothing ever published; `layers: []` = "no overlay, use your local config" | +| `POST /config/ack` | swarm key | `{instance_id, version, hash, status, restart_required, error}` — the instance's report after applying | +| `PUT /config` | admin | publish `{layers, note, force}`; an identical document is a no-op unless `force` (which re-notifies without a new version) | +| `GET /config/info` | admin | the current document as published | +| `GET /config/history` | admin | versions, newest first (`?limit=`) | +| `GET /config/version/{id}` | admin | one version, layers included | +| `POST /config/rollback/{id}` | admin | publish an old document as a **new** version and ping the fleet | +| `GET /config/status` | admin | per instance: what it reports running, the effective hash for the current layers, whether it is behind, and its last ack | + +## Central config + +The hive stores one versioned config document for the fleet (see the +"Central config" section of `docs/config.md` for the document shape, +validation rules and the fail scenarios). Operational notes: + +- **Publish**: + + ```bash + curl -X PUT http://127.0.0.1:8087/config \ + -H "X-Admin-Key: $(cat .hive-admin-key)" \ + -H 'Content-Type: application/json' \ + -d '{"layers":[{"name":"global","config":{"settings":{"llm_complete_timeout":90}}}],"note":"timeouts"}' + ``` + + `layers: []` is a real publish: it clears the fleet overlay and every + instance returns to its local config. + +- **Watch it land**: `GET /config/status` after a publish. `behind` names the + machines still on an older effective hash; each has its last ack and the + fields that need a restart. A `rejected` ack means that instance refused + the document — validation errors are in `last_error`, and the instance is + still running its previous config. +- **Retry**: the hive pings behind instances every `HIVE_NOTIFY_RETRY_SEC` + and on every announce reply, so a machine that was off catches up by + itself. There is nothing to do manually. +- **Rollback** is a forward publish of an old document. History is never + rewritten, including for a document that turned out to be bad. + +### Trust model + +- The **swarm key** can read the config, therefore it can read whatever + secrets the document holds. That is the price of a fleet-wide key rotation + without per-instance setup; keep port 8087 on the LAN and treat the key as + a secret on the same level as the config itself. +- **Publishing** needs the admin key (`HIVE_ADMIN_KEY_FILE`, default + `.hive-admin-key`, also `HIVE_ADMIN_KEY` env) or a loopback connection + (`HIVE_ALLOW_LOCAL_ADMIN=0` turns even that off). Once a key is configured + it is required from every address, loopback included. +- `hive.db` contains the config history **with secrets**. Back it up like a + credential store (0600) and do not copy it around. + ## Configuration (env vars) @@ -50,6 +101,12 @@ | `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 | +| `HIVE_ADMIN_KEY_FILE` | `.hive-admin-key` | admin key file for publishing/history/rollback/status | +| `HIVE_ADMIN_KEY` | `""` | admin key from the environment (takes precedence) | +| `HIVE_ALLOW_LOCAL_ADMIN` | `1` | allow admin calls from loopback without a key (only when no key is configured) | +| `HIVE_NOTIFY_TIMEOUT_SEC` | `3` | per-instance timeout for the config-change ping | +| `HIVE_NOTIFY_RETRY_SEC` | `300` | how often behind instances are re-pinged | +| `HIVE_NOTIFY_SCHEME` | `http` | scheme used when pinging instances | ## Navi side diff --git a/hive/app.py b/hive/app.py index 3ec6a61..add8547 100644 --- a/hive/app.py +++ b/hive/app.py @@ -1,39 +1,62 @@ -"""hive — the swarm address book. +"""hive — the swarm address book, and the fleet's config store. 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). +agents ask this book who is currently alive and where. It also stores the +central config: one versioned document whose layers are applied over each +instance's own config. 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. +Two credentials: + - the shared swarm key (PSK) — announce/peers/fetch/ack. It is also the + ability to *read* config secrets, which is why the hive belongs on the + LAN, not on a public port; + - the admin key (``HIVE_ADMIN_KEY_FILE``) or a loopback connection — + publishing, history, rollback and status. Deliberately not the swarm key, + so a compromised instance cannot rewrite what the fleet runs. """ from __future__ import annotations +import asyncio import hmac +import logging import os +import time +from contextlib import asynccontextmanager from pathlib import Path +import httpx from fastapi import FastAPI, Header, HTTPException, Request from pydantic import BaseModel, Field +from hive.config_store import ( + ConfigError, + content_hash, + effective_hash, + effective_layers, + validate_layers, +) from hive.db import HiveDB +log = logging.getLogger("hive") + _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") +_ADMIN_KEY_FILE = Path(os.environ.get("HIVE_ADMIN_KEY_FILE", ".hive-admin-key")) _DB_PATH = os.environ.get("HIVE_DB", "hive.db") _TTL_SEC = int(os.environ.get("HIVE_TTL_SEC", "300")) +_NOTIFY_TIMEOUT = float(os.environ.get("HIVE_NOTIFY_TIMEOUT_SEC", "3")) +_NOTIFY_RETRY_SEC = float(os.environ.get("HIVE_NOTIFY_RETRY_SEC", "300")) +_NOTIFY_SCHEME = os.environ.get("HIVE_NOTIFY_SCHEME", "http") -app = FastAPI(title="navi hive", description="Swarm address book for Navi instances.") -_db = HiveDB(_DB_PATH) + +# ── credentials ───────────────────────────────────────────────────────────── def _load_keys() -> list[str]: @@ -58,6 +81,178 @@ raise HTTPException(status_code=403, detail="invalid swarm key") +def _admin_key() -> str: + # Read on every call, not at import: the key can be installed while the + # hive runs, and reading it here keeps tests honest. + value = os.environ.get("HIVE_ADMIN_KEY", "") + if value.strip(): + return value.strip() + try: + return _ADMIN_KEY_FILE.read_text(encoding="utf-8").strip() + except OSError: + return "" + + +def _check_admin(request: Request, x_admin_key: str | None) -> str: + """Return who is acting, for the version's ``created_by``.""" + expected = _admin_key() + if x_admin_key: + if expected and hmac.compare_digest(x_admin_key, expected): + return "admin-key" + raise HTTPException(status_code=403, detail="invalid admin key") + peer = request.client.host if request.client else "" + if ( + not expected + and os.environ.get("HIVE_ALLOW_LOCAL_ADMIN", "1") != "0" + and peer in {"127.0.0.1", "::1", "localhost"} + ): + return f"local:{peer}" + raise HTTPException( + status_code=403, + detail=( + "publishing requires X-Admin-Key (HIVE_ADMIN_KEY_FILE) " + "or a loopback connection" + ), + ) + + +# ── app ───────────────────────────────────────────────────────────────────── + +_db = HiveDB(_DB_PATH) + + +def _reported_config(meta: dict | None) -> dict | None: + config = (meta or {}).get("config") + return config if isinstance(config, dict) else None + + +def _is_behind(reported: dict | None, effective: str) -> bool: + """Is this instance running something other than the current layers? + + ``False`` when the instance does not report config at all (an older + build) — that is "unknown", and re-notifying it forever helps nobody. + """ + if reported is None: + return False + if reported.get("status") == "disabled": + return False + return reported.get("hash") != effective + + +def _identity_of(peer: dict) -> tuple[str, str, list[str]]: + meta = peer.get("meta") or {} + tags = meta.get("tags") or [] + return peer.get("name") or "", peer.get("instance_id") or "", list(tags) + + +async def _notify_one(peer: dict, version: int, doc_hash: str) -> bool: + host, port = peer.get("host"), peer.get("port") + if not host or not port: + return False + keys = _load_keys() + if not keys: + return False + url = f"{_NOTIFY_SCHEME}://{host}:{port}/config/notify" + try: + async with httpx.AsyncClient(timeout=_NOTIFY_TIMEOUT) as client: + resp = await client.post( + url, + json={"version": version, "hash": doc_hash}, + headers={"X-Swarm-Key": keys[0]}, + ) + return resp.status_code < 300 + except Exception as exc: + log.info("config.notify_failed instance=%s error=%s", peer.get("name"), exc) + return False + + +async def _notify_all(version: int, doc_hash: str) -> None: + """Push a 'config changed' ping to every registered instance. + + Best effort by design: the announce reply carries the same hint, so an + instance that is offline (or behind a proxy the hive cannot reach) learns + about the change on its next heartbeat. The hive never needs this to + succeed for the config to be correct. + """ + peers = _db.list_peers(_TTL_SEC) + results = await asyncio.gather( + *(_notify_one(peer, version, doc_hash) for peer in peers), + return_exceptions=True, + ) + reached = sum(1 for r in results if r is True) + log.info( + "config.notified version=%s reached=%s total=%s", version, reached, len(peers) + ) + + +async def _notify_behind() -> None: + """Periodic pass: re-ping instances that still report an older hash.""" + latest = _db.latest_config() + if latest is None: + return + acks = _db.get_acks(latest["id"]) + peers = _db.list_peers(_TTL_SEC) + behind = [] + for peer in peers: + name, instance_id, tags = _identity_of(peer) + eff = effective_hash( + latest["layers"], name=name, instance_id=instance_id, tags=tags + ) + # A fresh ack for the current effective hash is proof enough: the + # announce payload may simply not have caught up yet. + ack = acks.get(instance_id) + if ack and ack.get("hash") == eff: + continue + if _is_behind(_reported_config(peer.get("meta")), eff): + behind.append(peer) + if not behind: + return + log.info( + "config.retry_notify version=%s count=%s names=%s", + latest["id"], + len(behind), + [p.get("name") for p in behind], + ) + await asyncio.gather( + *(_notify_one(peer, latest["id"], latest["hash"]) for peer in behind), + return_exceptions=True, + ) + + +async def _retry_loop() -> None: + while True: + await asyncio.sleep(_NOTIFY_RETRY_SEC) + try: + await _notify_behind() + except asyncio.CancelledError: + raise + except Exception: # a failed sweep must not kill the loop + log.exception("config.retry_loop_error") + + +@asynccontextmanager +async def lifespan(_: FastAPI): + task = asyncio.create_task(_retry_loop()) + try: + yield + finally: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + +app = FastAPI( + title="navi hive", + description="Swarm address book and central config store for Navi instances.", + lifespan=lifespan, +) + + +# ── address book ──────────────────────────────────────────────────────────── + + class AnnouncePayload(BaseModel): name: str = Field(min_length=1, max_length=64) instance_id: str = Field(min_length=8, max_length=64) @@ -68,7 +263,13 @@ @app.get("/health") async def health() -> dict: - return {"status": "ok", "hive": True, "machines": len(_db.list_peers(_TTL_SEC))} + latest = _db.latest_config() + return { + "status": "ok", + "hive": True, + "machines": len(_db.list_peers(_TTL_SEC)), + "config_version": latest["id"] if latest else 0, + } @app.post("/announce") @@ -90,10 +291,270 @@ version=payload.version, meta=payload.meta, ) - return {"ok": True, "host": host} + + # The heartbeat is the fallback delivery channel: tell an instance that it + # is behind, so a missed notification costs one announce interval, not a + # manual push. + hint = None + latest = _db.latest_config() + if latest is not None: + tags = list(payload.meta.get("tags") or []) + eff = effective_hash( + latest["layers"], + name=payload.name, + instance_id=payload.instance_id, + tags=tags, + ) + if _is_behind(_reported_config(payload.meta), eff): + hint = {"fetch": True, "version": latest["id"], "hash": eff} + return {"ok": True, "host": host, "config": hint} @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 + return {"peers": _db.list_peers(_TTL_SEC), "ttl_sec": _TTL_SEC} + + +# ── config: read side (swarm key) ─────────────────────────────────────────── + + +def _tags_param(value: str | None) -> list[str]: + return [t.strip() for t in (value or "").split(",") if t.strip()] + + +@app.get("/config") +async def get_config( + name: str = "", + instance_id: str = "", + tags: str = "", + x_swarm_key: str | None = Header(default=None), +) -> dict: + """The layers that apply to the asking instance. + + ``404`` when nothing was ever published, so an instance can tell "the + hive has no config channel for me" from "the hive says: no overlay". + An empty ``layers`` list is the latter and means "use your local config". + """ + _check_key(x_swarm_key) + latest = _db.latest_config() + if latest is None: + raise HTTPException(status_code=404, detail="no config published yet") + layers = effective_layers( + latest["layers"], + name=name, + instance_id=instance_id, + tags=_tags_param(tags), + ) + return { + "version": latest["id"], + "hash": content_hash(layers), + "layers": layers, + "note": latest["note"], + "updated_at": latest["created_at"], + } + + +class AckPayload(BaseModel): + instance_id: str = Field(min_length=8, max_length=64) + hash: str = Field(default="", max_length=128) + version: int = 0 + status: str = Field(default="applied", max_length=32) + restart_required: list[str] = Field(default_factory=list) + error: str | None = None + + +@app.post("/config/ack") +async def ack( + payload: AckPayload, + x_swarm_key: str | None = Header(default=None), +) -> dict: + _check_key(x_swarm_key) + latest = _db.latest_config() + version = payload.version or (latest["id"] if latest else 0) + if not version: + return {"ok": True, "ignored": "no config published"} + _db.record_ack( + instance_id=payload.instance_id, + version=version, + hash=payload.hash, + status=payload.status, + restart_required=payload.restart_required, + last_error=payload.error, + ) + return {"ok": True, "version": version} + + +# ── config: write and administration (admin key / loopback) ───────────────── + + +class PublishPayload(BaseModel): + layers: list[dict] + note: str = Field(default="", max_length=500) + force: bool = False + + +@app.put("/config") +async def publish( + payload: PublishPayload, + request: Request, + x_admin_key: str | None = Header(default=None), +) -> dict: + """Publish a new document and ping the fleet. + + An identical document is a no-op unless ``force`` is set — republishing + the same config would otherwise churn the whole fleet for nothing. + """ + actor = _check_admin(request, x_admin_key) + try: + layers = validate_layers(payload.layers) + except ConfigError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + doc_hash = content_hash(layers) + latest = _db.latest_config() + if latest is not None and latest["hash"] == doc_hash and not payload.force: + return { + "status": "unchanged", + "version": latest["id"], + "hash": doc_hash, + "layers": len(layers), + } + + if latest is not None and latest["hash"] == doc_hash: + asyncio.create_task(_notify_all(latest["id"], doc_hash)) + return { + "status": "renotified", + "version": latest["id"], + "hash": doc_hash, + "layers": len(layers), + } + + row = _db.add_config( + layers=layers, doc_hash=doc_hash, note=payload.note, created_by=actor + ) + asyncio.create_task(_notify_all(row["id"], doc_hash)) + return { + "status": "published", + "version": row["id"], + "hash": doc_hash, + "layers": len(layers), + "notify": "scheduled", + } + + +@app.get("/config/info") +async def config_info( + request: Request, x_admin_key: str | None = Header(default=None) +) -> dict: + _check_admin(request, x_admin_key) + latest = _db.latest_config() or {"id": 0, "hash": "", "layers": [], "note": ""} + return { + "version": latest["id"], + "hash": latest["hash"], + "note": latest.get("note") or "", + "layers": latest["layers"], + } + + +@app.get("/config/history") +async def config_history( + request: Request, + limit: int = 50, + x_admin_key: str | None = Header(default=None), +) -> dict: + _check_admin(request, x_admin_key) + return {"versions": _db.config_history(limit)} + + +@app.get("/config/version/{version_id}") +async def config_version( + version_id: int, + request: Request, + x_admin_key: str | None = Header(default=None), +) -> dict: + _check_admin(request, x_admin_key) + row = _db.get_config(version_id) + if row is None: + raise HTTPException(status_code=404, detail="no such version") + return row + + +@app.post("/config/rollback/{version_id}") +async def config_rollback( + version_id: int, + request: Request, + x_admin_key: str | None = Header(default=None), +) -> dict: + """Roll back by publishing an old document as a new version. + + History is never rewritten: the fleet's next fetch is a normal publish, + and the bad version stays visible in the history with its note intact. + """ + actor = _check_admin(request, x_admin_key) + row = _db.get_config(version_id) + if row is None: + raise HTTPException(status_code=404, detail="no such version") + new = _db.add_config( + layers=row["layers"], + doc_hash=row["hash"], + note=f"rollback to #{version_id}", + created_by=actor, + ) + asyncio.create_task(_notify_all(new["id"], row["hash"])) + return {"status": "rolled_back", "version": new["id"], "from": version_id} + + +@app.get("/config/status") +async def config_status( + request: Request, x_admin_key: str | None = Header(default=None) +) -> dict: + """Who runs the current config, who is behind, who rejected it. + + This is the tracking the owner asked for: after a publish, look here to + see which machines took it and which still need a nudge (the hive pings + those on its own every ``HIVE_NOTIFY_RETRY_SEC``). + """ + _check_admin(request, x_admin_key) + latest = _db.latest_config() + acks = _db.get_acks(latest["id"]) if latest else {} + instances = [] + behind = [] + for peer in _db.list_peers(_TTL_SEC): + name, instance_id, tags = _identity_of(peer) + reported = _reported_config(peer.get("meta")) + eff = ( + effective_hash( + latest["layers"], name=name, instance_id=instance_id, tags=tags + ) + if latest + else "" + ) + is_behind = bool(latest) and _is_behind(reported, eff) + ack = acks.get(instance_id) + if ack and ack.get("hash") == eff: + # The ack is fresher than the announce payload: it applied the + # current effective layers, whatever the last heartbeat said. + is_behind = False + if is_behind: + behind.append(name) + instances.append( + { + "name": name, + "instance_id": instance_id, + "tags": tags, + "online": peer["online"], + "last_seen": peer["last_seen"], + "reported": reported, + "effective_hash": eff, + "behind": is_behind, + "ack": acks.get(instance_id), + } + ) + return { + "version": latest["id"] if latest else 0, + "hash": latest["hash"] if latest else "", + "generated_at": time.time(), + "behind": behind, + "instances": instances, + } diff --git a/hive/config_store.py b/hive/config_store.py new file mode 100644 index 0000000..8d27dd4 --- /dev/null +++ b/hive/config_store.py @@ -0,0 +1,141 @@ +"""Config documents for the fleet: shape, layer selection, hashing. + +A document is a list of layers, each with a name, a ``config`` merge-patch and +an optional ``match`` selector:: + + {"layers": [ + {"name": "global", "config": {"settings": {"gnauth_client_id": "..."}}}, + {"name": "lab", "match": {"tags": ["lab"]}, + "config": {"settings": {"llm_complete_timeout": 120}}}, + {"name": "yuki", "match": {"names": ["yuki"]}, + "config": {"profiles": {"assistant": {"temperature": 0.2}}}} + ]} + +Layers apply in document order over the instance's local config; a layer with +no ``match`` (or an empty one) applies to everyone. ``enabled: false`` keeps a +layer in the history but out of the fleet. + +The hash is computed over the **layers an instance actually receives**, not +over the whole document, and with exactly the same canonicalisation the +instance uses (``navi/overlay/util.py``). Publishing a layer that targets one +machine therefore leaves every other machine's hash untouched — otherwise each +publish would make the whole fleet re-ack identical content. +""" + +from __future__ import annotations + +import hashlib +import json + +_MATCH_KEYS = {"names", "instance_ids", "tags"} + + +class ConfigError(ValueError): + """A published document is not usable. Reported as HTTP 400.""" + + +def canonical_json(value: object) -> str: + # Keep in lockstep with navi/overlay/util.py — the two hashes are compared. + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + + +def content_hash(value: object) -> str: + digest = hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest() + return f"sha256:{digest}" + + +def _string_list(value: object, where: str) -> list[str]: + if value is None: + return [] + if not isinstance(value, list) or not all(isinstance(v, str) and v for v in value): + raise ConfigError(f"{where} must be a list of non-empty strings") + return list(value) + + +def validate_layers(raw: object) -> list[dict]: + """Normalise a published document; raise ``ConfigError`` on bad shape.""" + if not isinstance(raw, list): + raise ConfigError("'layers' must be a list") + # An empty list is a real publish: it clears the fleet overlay and every + # instance falls back to its local config. + + seen: set[str] = set() + layers: list[dict] = [] + for index, layer in enumerate(raw): + if not isinstance(layer, dict): + raise ConfigError(f"layer #{index} is not an object") + name = layer.get("name") + if not isinstance(name, str) or not name.strip(): + raise ConfigError(f"layer #{index} needs a non-empty 'name'") + if name in seen: + raise ConfigError(f"duplicate layer name {name!r}") + seen.add(name) + + config = layer.get("config") + if not isinstance(config, dict): + raise ConfigError(f"layer {name!r}: 'config' must be an object") + + match_raw = layer.get("match") or {} + if not isinstance(match_raw, dict): + raise ConfigError(f"layer {name!r}: 'match' must be an object") + unknown = set(match_raw) - _MATCH_KEYS + if unknown: + raise ConfigError( + f"layer {name!r}: unknown match keys {sorted(unknown)} " + f"(known: {sorted(_MATCH_KEYS)})" + ) + match = { + key: _string_list(match_raw.get(key), f"layer {name!r}: match.{key}") + for key in _MATCH_KEYS + if match_raw.get(key) is not None + } + if not any(match.values()): + match = {} + + enabled = layer.get("enabled", True) + if not isinstance(enabled, bool): + raise ConfigError(f"layer {name!r}: 'enabled' must be true or false") + + layers.append( + {"name": name, "config": config, "match": match, "enabled": enabled} + ) + return layers + + +def _matches(layer: dict, name: str, instance_id: str, tags: list[str]) -> bool: + match = layer.get("match") or {} + if not match: # a global layer + return True + if name and name in match.get("names", []): + return True + if instance_id and instance_id in match.get("instance_ids", []): + return True + return bool(set(tags) & set(match.get("tags", []))) + + +def effective_layers( + layers: list[dict], + *, + name: str = "", + instance_id: str = "", + tags: list[str] | None = None, +) -> list[dict]: + """The layers this instance receives, in document order, sans metadata.""" + tags = tags or [] + return [ + {"name": layer["name"], "config": layer["config"]} + for layer in layers + if layer.get("enabled", True) and _matches(layer, name, instance_id, tags) + ] + + +def effective_hash( + layers: list[dict], + *, + name: str = "", + instance_id: str = "", + tags: list[str] | None = None, +) -> str: + return content_hash( + effective_layers(layers, name=name, instance_id=instance_id, tags=tags) + ) diff --git a/hive/db.py b/hive/db.py index 31096b0..58968b3 100644 --- a/hive/db.py +++ b/hive/db.py @@ -23,6 +23,31 @@ first_seen REAL NOT NULL, last_seen REAL NOT NULL ); + +-- Append-only config history. A rollback is a new row, never a rewrite, so +-- "what was the fleet told, and when" is always answerable — including for a +-- document that turned out to be bad. +CREATE TABLE IF NOT EXISTS config_versions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + created_at REAL NOT NULL, + created_by TEXT, + note TEXT, + layers TEXT NOT NULL, + hash TEXT NOT NULL +); + +-- The last reply from each instance, per version. This is the tracking the +-- owner asked for: who has it, who staged it, who rejected it and why. +CREATE TABLE IF NOT EXISTS config_acks ( + instance_id TEXT NOT NULL, + version INTEGER NOT NULL, + hash TEXT NOT NULL, + status TEXT NOT NULL, + restart_required TEXT, + last_error TEXT, + acked_at REAL NOT NULL, + PRIMARY KEY (instance_id, version) +); """ @@ -100,4 +125,132 @@ "meta": meta, } ) - return peers \ No newline at end of file + return peers + + # ── config documents ──────────────────────────────────────────────────── + + def add_config( + self, + *, + layers: list[dict], + doc_hash: str, + note: str = "", + created_by: str = "", + ) -> dict: + now = time.time() + cur = self._conn.execute( + """ + INSERT INTO config_versions (created_at, created_by, note, layers, hash) + VALUES (?, ?, ?, ?, ?) + """, + (now, created_by, note, json.dumps(layers), doc_hash), + ) + self._conn.commit() + return self.get_config(int(cur.lastrowid)) # type: ignore[arg-type] + + def _version_row(self, row: sqlite3.Row) -> dict: + try: + layers = json.loads(row["layers"] or "[]") + except json.JSONDecodeError: + layers = [] + return { + "id": row["id"], + "created_at": row["created_at"], + "created_by": row["created_by"], + "note": row["note"], + "layers": layers, + "hash": row["hash"], + } + + def get_config(self, version_id: int) -> dict | None: + row = self._conn.execute( + "SELECT * FROM config_versions WHERE id = ?", (version_id,) + ).fetchone() + return self._version_row(row) if row else None + + def latest_config(self) -> dict | None: + row = self._conn.execute( + "SELECT * FROM config_versions ORDER BY id DESC LIMIT 1" + ).fetchone() + return self._version_row(row) if row else None + + def config_history(self, limit: int = 50) -> list[dict]: + rows = self._conn.execute( + """ + SELECT id, created_at, created_by, note, hash, layers + FROM config_versions ORDER BY id DESC LIMIT ? + """, + (max(1, min(limit, 500)),), + ).fetchall() + out = [] + for row in rows: + try: + count = len(json.loads(row["layers"] or "[]")) + except json.JSONDecodeError: + count = 0 + out.append( + { + "id": row["id"], + "created_at": row["created_at"], + "created_by": row["created_by"], + "note": row["note"], + "hash": row["hash"], + "layers": count, + } + ) + return out + + # ── acks ──────────────────────────────────────────────────────────────── + + def record_ack( + self, + *, + instance_id: str, + version: int, + hash: str, + status: str, + restart_required: list[str] | None = None, + last_error: str | None = None, + ) -> None: + self._conn.execute( + """ + INSERT INTO config_acks + (instance_id, version, hash, status, restart_required, last_error, acked_at) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(instance_id, version) DO UPDATE SET + hash = excluded.hash, + status = excluded.status, + restart_required = excluded.restart_required, + last_error = excluded.last_error, + acked_at = excluded.acked_at + """, + ( + instance_id, + version, + hash, + status, + json.dumps(restart_required or []), + last_error, + time.time(), + ), + ) + self._conn.commit() + + def get_acks(self, version: int) -> dict[str, dict]: + rows = self._conn.execute( + "SELECT * FROM config_acks WHERE version = ?", (version,) + ).fetchall() + out: dict[str, dict] = {} + for row in rows: + try: + restart = json.loads(row["restart_required"] or "[]") + except json.JSONDecodeError: + restart = [] + out[row["instance_id"]] = { + "status": row["status"], + "hash": row["hash"], + "restart_required": restart, + "last_error": row["last_error"], + "acked_at": row["acked_at"], + } + return out \ No newline at end of file diff --git a/tests/unit/test_hive.py b/tests/unit/test_hive.py index fa0e778..3e4cd00 100644 --- a/tests/unit/test_hive.py +++ b/tests/unit/test_hive.py @@ -16,6 +16,9 @@ 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") + # Deterministic admin key: no ambient .hive-admin-key from the checkout. + monkeypatch.setenv("HIVE_ADMIN_KEY", "admin-secret") + monkeypatch.setenv("HIVE_ADMIN_KEY_FILE", str(tmp_path / ".hive-admin-key")) (tmp_path / ".swarm-key").write_text("secret-key-123") import hive.app as hive_app @@ -125,4 +128,306 @@ headers={"X-Swarm-Key": "whatever"}, ) assert resp.status_code == 503 - module._db.close() \ No newline at end of file + module._db.close() + +# ── config documents ──────────────────────────────────────────────────────── + + +def _admin(key: str = "admin-secret") -> dict: + return {"X-Admin-Key": key} + + +def _publish(client, layers, note="", force=False, key="admin-secret"): + return client.put( + "/config", + json={"layers": layers, "note": note, "force": force}, + headers=_admin(key), + ) + + +def _global(**settings): + return {"name": "global", "config": {"settings": settings}} + + +def test_publish_requires_admin_key(hive): + with TestClient(hive.app) as client: + resp = client.put("/config", json={"layers": [_global(llm_complete_timeout=30)]}) + assert resp.status_code == 403 + bad = client.put( + "/config", + json={"layers": [_global(llm_complete_timeout=30)]}, + headers=_admin("wrong"), + ) + assert bad.status_code == 403 + + +def test_fetch_returns_404_before_anything_is_published(hive): + with TestClient(hive.app) as client: + assert client.get("/config", headers=_headers()).status_code == 404 + + +def test_publish_then_fetch_global_layer(hive): + with TestClient(hive.app) as client: + published = _publish(client, [_global(llm_complete_timeout=30)]).json() + assert published["status"] == "published" + assert published["version"] == 1 + + body = client.get("/config", headers=_headers()).json() + assert body["version"] == 1 + assert body["layers"] == [ + {"name": "global", "config": {"settings": {"llm_complete_timeout": 30}}} + ] + + +def test_layers_target_by_tag_and_name(hive): + layers = [ + {"name": "global", "config": {"settings": {"llm_complete_timeout": 30}}}, + {"name": "lab", "match": {"tags": ["lab"]}, "config": {"settings": {"llm_complete_retries": 1}}}, + {"name": "yuki", "match": {"names": ["yuki"]}, "config": {"settings": {"navi_push_cooldown_sec": 5}}}, + ] + with TestClient(hive.app) as client: + _publish(client, layers) + + plain = client.get("/config", params={"name": "other"}, headers=_headers()).json() + assert [layer["name"] for layer in plain["layers"]] == ["global"] + + lab = client.get( + "/config", params={"name": "other", "tags": "beta,lab"}, headers=_headers() + ).json() + assert [layer["name"] for layer in lab["layers"]] == ["global", "lab"] + + named = client.get("/config", params={"name": "yuki"}, headers=_headers()).json() + assert [layer["name"] for layer in named["layers"]] == ["global", "yuki"] + + # The hash is over the layers that instance receives, not the document. + assert plain["hash"] != named["hash"] + + +def test_identical_publish_is_a_no_op(hive): + with TestClient(hive.app) as client: + _publish(client, [_global(llm_complete_timeout=30)]) + again = _publish(client, [_global(llm_complete_timeout=30)]).json() + assert again["status"] == "unchanged" + assert again["version"] == 1 + + forced = _publish(client, [_global(llm_complete_timeout=30)], force=True).json() + assert forced["status"] == "renotified" + assert forced["version"] == 1 + assert len(client.get("/config/history", headers=_admin()).json()["versions"]) == 1 + + +def test_empty_document_clears_the_overlay(hive): + with TestClient(hive.app) as client: + _publish(client, [_global(llm_complete_timeout=30)]) + cleared = _publish(client, [], note="back to local").json() + assert cleared["status"] == "published" + assert cleared["version"] == 2 + + body = client.get("/config", headers=_headers()).json() + assert body["layers"] == [] + + +def test_invalid_document_is_rejected_with_a_reason(hive): + with TestClient(hive.app) as client: + resp = _publish(client, [{"name": "x"}]) + assert resp.status_code == 400 + assert "config" in resp.json()["detail"] + + resp = _publish(client, [{"name": "x", "config": {}, "match": {"nope": ["a"]}}]) + assert resp.status_code == 400 + + +def test_announce_hint_tells_a_behind_instance_to_fetch(hive): + with TestClient(hive.app) as client: + _publish(client, [_global(llm_complete_timeout=30)]) + + # An older build that reports no config: unknown, not behind. + old = _announce(client).json() + assert old["config"] is None + + # A current build with an empty hash is behind. + behind = client.post( + "/announce", + json={ + "name": "quiet-otter", + "instance_id": "11111111-2222-3333-4444-555555555555", + "port": 8099, + "meta": {"config": {"hash": "", "status": "applied"}}, + }, + headers=_headers(), + ).json() + assert behind["config"]["fetch"] is True + assert behind["config"]["version"] == 1 + + # One that reports the exact effective hash is up to date. + effective = client.get("/config", headers=_headers()).json()["hash"] + current = client.post( + "/announce", + json={ + "name": "quiet-otter", + "instance_id": "11111111-2222-3333-4444-555555555555", + "port": 8099, + "meta": {"config": {"hash": effective, "status": "applied"}}, + }, + headers=_headers(), + ).json() + assert current["config"] is None + + +def test_ack_is_recorded_and_shown_in_status(hive): + with TestClient(hive.app) as client: + _publish(client, [_global(llm_complete_timeout=30)]) + _announce(client) + resp = client.post( + "/config/ack", + json={ + "instance_id": "11111111-2222-3333-4444-555555555555", + "version": 1, + "hash": "sha256:x", + "status": "staged", + "restart_required": ["openai_api_key"], + }, + headers=_headers(), + ) + assert resp.status_code == 200 + + status = client.get("/config/status", headers=_admin()).json() + row = status["instances"][0] + assert row["ack"]["status"] == "staged" + assert row["ack"]["restart_required"] == ["openai_api_key"] + + +def test_status_marks_behind_instances(hive): + with TestClient(hive.app) as client: + _publish(client, [_global(llm_complete_timeout=30)]) + effective = client.get("/config", headers=_headers()).json()["hash"] + client.post( + "/announce", + json={ + "name": "quiet-otter", + "instance_id": "11111111-2222-3333-4444-555555555555", + "port": 8099, + "meta": {"config": {"hash": "sha256:old", "status": "applied"}}, + }, + headers=_headers(), + ) + status = client.get("/config/status", headers=_admin()).json() + assert status["behind"] == ["quiet-otter"] + assert status["instances"][0]["effective_hash"] == effective + + client.post( + "/announce", + json={ + "name": "quiet-otter", + "instance_id": "11111111-2222-3333-4444-555555555555", + "port": 8099, + "meta": {"config": {"hash": effective, "status": "applied"}}, + }, + headers=_headers(), + ) + assert client.get("/config/status", headers=_admin()).json()["behind"] == [] + + +def test_history_version_and_rollback(hive): + with TestClient(hive.app) as client: + _publish(client, [_global(llm_complete_timeout=30)], note="first") + _publish(client, [_global(llm_complete_timeout=45)], note="second") + + history = client.get("/config/history", headers=_admin()).json()["versions"] + assert [v["id"] for v in history] == [2, 1] + assert history[1]["note"] == "first" + + version = client.get("/config/version/1", headers=_admin()).json() + assert version["layers"][0]["config"]["settings"]["llm_complete_timeout"] == 30 + + rolled = client.post("/config/rollback/1", headers=_admin()).json() + assert rolled["status"] == "rolled_back" + assert rolled["version"] == 3 + assert client.get("/config", headers=_headers()).json()["layers"][0]["config"] == { + "settings": {"llm_complete_timeout": 30} + } + # History keeps the bad version rather than rewriting it. + assert len(client.get("/config/history", headers=_admin()).json()["versions"]) == 3 + + +async def test_notify_all_pings_every_registered_instance(hive, monkeypatch): + with TestClient(hive.app) as client: + _announce(client, name="one", instance_id="aaaaaaaa-0000-0000-0000-000000000001") + _announce(client, name="two", instance_id="aaaaaaaa-0000-0000-0000-000000000002") + calls = [] + + async def fake(peer, version, doc_hash): + calls.append((peer["name"], version, doc_hash)) + return True + + monkeypatch.setattr(hive, "_notify_one", fake) + await hive._notify_all(4, "sha256:x") + assert sorted(name for name, _, _ in calls) == ["one", "two"] + + +async def test_retry_sweep_targets_only_the_instance_that_is_behind(hive, monkeypatch): + with TestClient(hive.app) as client: + _announce(client, name="one", instance_id="aaaaaaaa-0000-0000-0000-000000000001") + _announce(client, name="two", instance_id="aaaaaaaa-0000-0000-0000-000000000002") + _publish(client, [_global(llm_complete_timeout=30)]) + effective = client.get("/config", headers=_headers()).json()["hash"] + + client.post( + "/announce", + json={ + "name": "one", + "instance_id": "aaaaaaaa-0000-0000-0000-000000000001", + "port": 8099, + "meta": {"config": {"hash": effective, "status": "applied"}}, + }, + headers=_headers(), + ) + client.post( + "/announce", + json={ + "name": "two", + "instance_id": "aaaaaaaa-0000-0000-0000-000000000002", + "port": 8099, + "meta": {"config": {"hash": "sha256:old", "status": "applied"}}, + }, + headers=_headers(), + ) + + calls = [] + + async def fake(peer, version, doc_hash): + calls.append(peer["name"]) + return True + + monkeypatch.setattr(hive, "_notify_one", fake) + await hive._notify_behind() + assert calls == ["two"] + + +def test_a_fresh_ack_beats_a_stale_announce(hive): + """The announce may lag an interval behind; an ack for the current hash is proof.""" + iid = "11111111-2222-3333-4444-555555555555" + with TestClient(hive.app) as client: + _publish(client, [_global(llm_complete_timeout=30)]) + effective = client.get("/config", headers=_headers()).json()["hash"] + + client.post( + "/announce", + json={ + "name": "quiet-otter", + "instance_id": iid, + "port": 8099, + "meta": {"config": {"hash": "sha256:old", "status": "applied"}}, + }, + headers=_headers(), + ) + assert client.get("/config/status", headers=_admin()).json()["behind"] == [ + "quiet-otter" + ] + + client.post( + "/config/ack", + json={"instance_id": iid, "version": 1, "hash": effective, "status": "applied"}, + headers=_headers(), + ) + assert client.get("/config/status", headers=_admin()).json()["behind"] == [] diff --git a/tests/unit/test_hive_config_store.py b/tests/unit/test_hive_config_store.py new file mode 100644 index 0000000..a45805c --- /dev/null +++ b/tests/unit/test_hive_config_store.py @@ -0,0 +1,113 @@ +"""hive.config_store — document shape, layer targeting, and the hash contract.""" + +import pytest + +from hive.config_store import ( + ConfigError, + content_hash, + effective_hash, + effective_layers, + validate_layers, +) + + +def test_validate_accepts_a_minimal_layer(): + layers = validate_layers([{"name": "g", "config": {"settings": {"a": 1}}}]) + assert layers == [{"name": "g", "config": {"settings": {"a": 1}}, "match": {}, "enabled": True}] + + +def test_validate_rejects_duplicate_names(): + with pytest.raises(ConfigError, match="duplicate"): + validate_layers( + [ + {"name": "g", "config": {}}, + {"name": "g", "config": {}}, + ] + ) + + +def test_validate_rejects_non_object_config(): + with pytest.raises(ConfigError): + validate_layers([{"name": "g", "config": []}]) + + +def test_validate_rejects_unknown_match_keys(): + with pytest.raises(ConfigError, match="unknown match keys"): + validate_layers([{"name": "g", "config": {}, "match": {"hostnames": ["x"]}}]) + + +def test_validate_rejects_non_bool_enabled(): + with pytest.raises(ConfigError): + validate_layers([{"name": "g", "config": {}, "enabled": "yes"}]) + + +def test_empty_match_lists_mean_global(): + layers = validate_layers( + [{"name": "g", "config": {}, "match": {"tags": [], "names": []}}] + ) + assert layers[0]["match"] == {} + assert effective_layers(layers) == [{"name": "g", "config": {}}] + + +def test_effective_layers_keep_document_order(): + layers = validate_layers( + [ + {"name": "yuki", "match": {"names": ["yuki"]}, "config": {"a": 2}}, + {"name": "global", "config": {"a": 1}}, + ] + ) + got = effective_layers(layers, name="yuki") + assert [layer["name"] for layer in got] == ["yuki", "global"] + + +def test_disabled_layer_is_skipped_but_kept_in_history(): + layers = validate_layers( + [ + {"name": "g", "config": {"a": 1}}, + {"name": "off", "config": {"b": 2}, "enabled": False}, + ] + ) + assert effective_layers(layers) == [{"name": "g", "config": {"a": 1}}] + assert len(layers) == 2 + + +def test_instance_id_matching(): + layers = validate_layers( + [{"name": "pinned", "match": {"instance_ids": ["abc"]}, "config": {"a": 1}}] + ) + assert effective_layers(layers, instance_id="abc") == [{"name": "pinned", "config": {"a": 1}}] + assert effective_layers(layers, instance_id="other") == [] + + +def test_hash_is_over_the_layers_the_instance_receives(): + layers = validate_layers( + [ + {"name": "global", "config": {"a": 1}}, + {"name": "lab", "match": {"tags": ["lab"]}, "config": {"b": 2}}, + ] + ) + plain = effective_layers(layers) + lab = effective_layers(layers, tags=["lab"]) + assert effective_hash(layers) == content_hash(plain) + assert effective_hash(layers, tags=["lab"]) == content_hash(lab) + assert effective_hash(layers) != effective_hash(layers, tags=["lab"]) + + +def test_hash_ignores_match_and_enabled_metadata(): + # Targeting metadata is not part of what an instance receives: a machine + # that gets the same layer content hashes the same whether the layer was + # global or addressed to it. Otherwise every re-targeting publish would + # make unaffected machines re-ack identical content. + a = validate_layers([{"name": "g", "config": {"a": 1}}]) + b = validate_layers( + [{"name": "g", "config": {"a": 1}, "match": {"tags": ["x"]}}] + ) + assert effective_hash(a) == effective_hash(b, tags=["x"]) + + +def test_hash_function_matches_the_instance_side(): + """The hive and the instance must agree byte for byte on 'same config'.""" + from navi.overlay.util import content_hash as navi_hash + + value = [{"name": "g", "config": {"settings": {"x": [1, 2], "y": {"z": "ю"}}}}] + assert content_hash(value) == navi_hash(value)