diff --git a/.gitignore b/.gitignore index c8fedf8..8d35215 100644 --- a/.gitignore +++ b/.gitignore @@ -34,3 +34,9 @@ instance.json .swarm-key .swarm-key.previous +# central config: the cache holds the hive document, secrets included +.hive-config.cache.json +.hive-admin-key +hive.db +hive.db-wal +hive.db-shm diff --git a/docs/config.md b/docs/config.md index 223d9e8..50aa479 100644 --- a/docs/config.md +++ b/docs/config.md @@ -193,6 +193,79 @@ | `INSTANCE_FILE` | str | `instance.json` | This navi's identity for the swarm: generated name (single female name from Japanese/American/Spanish pools) + instance id. Rename = edit the file, restart. | | `PEER_ASK_PROFILE` | str | `server_admin` | Which profile answers incoming `/peer/ask` questions (one-shot agent run). | | `PEER_ASK_TIMEOUT_SEC` | int | `120` | Hard ceiling for a single peer ask — the LLM turn inside `/peer/ask`. | +| `HIVE_CONFIG_ENABLED` | bool | `true` | Use the hive's central config when `HIVE_URL` is set. `NAVI_OVERLAY=off` in the environment disables it for one process (tests do this). | +| `HIVE_CONFIG_CACHE_FILE` | str | `.hive-config.cache.json` | Last validated overlay + the boot-loop guard state. Gitignored, 0600; it holds secrets. | +| `HIVE_CONFIG_TIMEOUT_SEC` | float | `2.0` | Timeout for the config fetch at boot and on refresh. | + +### Central config (hive overlay) + +The hive can store one config document for the whole fleet. Every instance +that knows about the hive fetches it and applies it **over** its own config: +fields present in the document win, fields it does not mention come from the +local files, and an instance the document does not target runs purely on its +local config. Secrets may live in the document — rotating a key across the +fleet without touching a single machine is the main purpose. + +**Shape.** A document is a list of layers applied in order. Each layer has a +name, a `config` merge-patch (RFC 7386: objects merge, arrays and scalars +replace, `null` drops the key) and an optional `match` targeting +`names` / `instance_ids` / `tags` (labels from `instance.json`). No `match` +means every instance. + +```json +{"layers": [ + {"name": "global", "config": {"settings": {"llm_complete_timeout": 90}}}, + {"name": "lab", "match": {"tags": ["lab"]}, + "config": {"llm": {"targets": {"home": {"host": "http://192.168.1.129:11434"}}}}}, + {"name": "yuki", "match": {"names": ["yuki"]}, + "config": {"profiles": {"assistant": {"temperature": 0.3}}}} +]} +``` + +Sections: `settings` (any `Settings` field except the deny-list below), +`profiles` (any profile field, including `tools.agent.native`), `llm` (merged +into the local `llm.json`) and `mcp_servers` (add or override a server +definition — not remove one; `null` is consumed by the merge). + +**Never overlaid:** `HIVE_URL`, `SWARM_KEY_FILE`, `INSTANCE_FILE`, +`HIVE_CONFIG_*`, `DATABASE_URL`, `NAVI_HOST`, `NAVI_PORT`, +`NAVI_UI_MCP_HOST/PORT`, `NAVI_AUTH_ENABLED`. These are the bootstrap fields +(needed before the overlay exists) and the values that can make a machine +unreachable or point it at the wrong database — a valid but wrong value is the +one failure validation cannot catch. + +**Validation is all-or-nothing.** Unknown sections, settings, profile fields +and MCP fields are dropped with a warning (forward compatibility: a newer hive +may know more than this instance). A *known* field with an invalid value +rejects the whole document — half a document is a state nobody wrote — and the +instance keeps running the previous one. A rejected document never overwrites +the cache. + +**Delivery.** On publish the hive pings every registered instance +(`POST /config/notify`, swarm key); each instance fetches and validates on its +own. The announce reply carries the same hint, so a machine that was offline +learns about the change on its next heartbeat, and the hive re-pings +still-behind instances periodically. `GET /config/status` on the hive lists +who has it, who staged it, who rejected it and why. + +**Boot safety.** The last validated overlay is cached and used when the hive is +unreachable or the fetched document is invalid. A config that differed from the +last confirmed one is written as *pending* and confirmed only once the process +reaches the end of startup; if it prevents startup twice, it is blocked and the +last good one is used (a new publish clears the block). + +**Hot vs restart.** Secrets (`GNAUTH_*`, synapse source), timeouts, tool +policies and profiles are applied to the running process. Anything built at +startup (backends, the MCP pool, CORS, the push service) is reported as +`restart_required` in `GET /health` → `config` and in the hive's ack view; +it takes effect on the next start. + +**Trust.** The hive is a LAN service. The swarm key is the fleet's read +credential and therefore the ability to read config secrets; publishing, +history and rollback require the admin key (`HIVE_ADMIN_KEY_FILE`) or a +loopback connection. The hive's `hive.db` contains secrets — back it up +accordingly and do not expose port 8087 outside the LAN. + 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. diff --git a/navi/api/routes/config_overlay.py b/navi/api/routes/config_overlay.py new file mode 100644 index 0000000..d337a18 --- /dev/null +++ b/navi/api/routes/config_overlay.py @@ -0,0 +1,77 @@ +"""Central config endpoints — the hive tells this instance that config changed. + +The hive does not push the document itself. It posts ``{"version", "hash"}`` +to every registered instance; each instance fetches, validates and applies the +document on its own, which keeps the instance in control of what it runs and +lets a machine that was offline catch up on its next start (or on its next +announce reply). + +``POST /config/notify`` accepts a request from anyone holding the swarm key. +That key is the fleet's read credential and it is enough here: a notification +causes a *fetch* against the hive, so a forged one can at most make this +instance re-read the config it would have read anyway. +""" + +from __future__ import annotations + +import asyncio + +import structlog +from fastapi import APIRouter, Header, HTTPException +from pydantic import BaseModel + +from navi.overlay import refresh, status_dict +from navi.overlay.state import get_state + +router = APIRouter(prefix="/config", tags=["config"]) + +log = structlog.get_logger() + + +def _require_swarm_key(x_swarm_key: str | None) -> None: + from navi.swarm import verify_swarm_key + + if not x_swarm_key: + raise HTTPException(status_code=401, detail="X-Swarm-Key header required") + if not verify_swarm_key(x_swarm_key): + raise HTTPException(status_code=403, detail="invalid swarm key") + + +class NotifyPayload(BaseModel): + version: int | None = None + hash: str | None = None + + +@router.post("/notify", status_code=202) +async def config_notify( + payload: NotifyPayload | None = None, + x_swarm_key: str | None = Header(None), +) -> dict: + """Schedule a refresh; answer at once so the hive's fan-out stays fast.""" + _require_swarm_key(x_swarm_key) + state = get_state() + if not state.enabled or state.base_settings is None: + return {"status": state.status, "detail": "central config disabled on this instance"} + + notified = payload or NotifyPayload() + log.info( + "hive.config.notified", + version=notified.version, + hash=notified.hash, + current_hash=state.hash, + ) + # Blocking HTTP inside the refresh — run it off the event loop. + asyncio.create_task(asyncio.to_thread(refresh, "notify")) + return { + "status": "scheduled", + "version": notified.version, + "hash": notified.hash, + "current": status_dict()["status"], + } + + +@router.get("/status") +async def config_status(x_swarm_key: str | None = Header(None)) -> dict: + """What this instance currently runs — the same block /health shows.""" + _require_swarm_key(x_swarm_key) + return status_dict() diff --git a/navi/api/routes/health.py b/navi/api/routes/health.py index 2c1f3b9..a243eba 100644 --- a/navi/api/routes/health.py +++ b/navi/api/routes/health.py @@ -43,6 +43,11 @@ # Legacy payloads (pre-contract) — kept for existing tooling. "embed": embed_status, "hive": hive, + # Informational only: what central config this process runs, and + # whether applying the last one still needs a restart. An instance + # that fell back to local config is not "down" — the overlay is a + # convenience channel, not a dependency. + "config": _config_status(), } export_status = _export_status() if export_status is not None: @@ -52,6 +57,16 @@ return body +def _config_status() -> dict: + """Compact overlay state; never raises into the health response.""" + try: + from navi.overlay import get_state, status_summary + + return {**status_summary(), "source": get_state().source} + except Exception: + return {"status": "unknown"} + + def _export_status() -> dict | None: """Freshness of the history snapshot, when the export is configured.""" if not settings.navi_export_configured: diff --git a/navi/auth/client.py b/navi/auth/client.py index 8019625..38f5f13 100644 --- a/navi/auth/client.py +++ b/navi/auth/client.py @@ -70,3 +70,14 @@ if _default_gauth_client is None: _default_gauth_client = _make_client(settings.gnauth_redirect_uri) return _default_gauth_client + + +def reset_default_client() -> None: + """Drop the cached client so the next call picks up new credentials. + + Called after a hive config refresh rotated ``GNAUTH_*``. The shared state + and PKCE stores live in this module and are deliberately kept — an + in-flight login must survive a key rotation. + """ + global _default_gauth_client + _default_gauth_client = None diff --git a/navi/config.py b/navi/config.py index 162738a..97b319a 100644 --- a/navi/config.py +++ b/navi/config.py @@ -1,310 +1,20 @@ -from pathlib import Path +"""Process-wide settings. -from pydantic import model_validator -from pydantic_settings import BaseSettings, SettingsConfigDict +The model lives in :mod:`navi.settings_model` (no side effects — the hive +imports it to know which field names exist). This module builds the singleton: +local environment and ``.env`` first, then the hive's central overlay on top. +The overlay is fetched here, at import time, because every other module +imports ``settings`` and a configuration that arrives after them would only +ever be half-applied. The fetch is bounded (``HIVE_CONFIG_TIMEOUT_SEC``), +never fatal, and falls back to the last validated overlay cached on disk. +Without ``HIVE_URL`` — or with ``NAVI_OVERLAY=off`` — this is exactly the +old behaviour: a pure local config. +""" -class Settings(BaseSettings): - model_config = SettingsConfigDict( - env_file=".env", env_file_encoding="utf-8", extra="ignore", frozen=True - ) +from navi.overlay import bootstrap_settings +from navi.settings_model import Settings - ollama_host: str = "http://localhost:11434" - ollama_api_key: str = "" - ollama_default_model: str = "gemma4:31b-cloud" - ollama_num_ctx: int = 65536 - ollama_think: bool = True - ollama_request_timeout: int = 30 +__all__ = ["Settings", "settings"] - # Embedding model for memory vector search (Ollama API) - # When embedding_ollama_host is empty, falls back to ollama_host. - embedding_ollama_host: str = "" - embedding_ollama_api_key: str = "" - embedding_model: str = "nomic-embed-text:latest" - # Route in llm.json used for embeddings (same failover as chat). Ignored - # when embedding_ollama_host pins a dedicated host. - embedding_route: str = "embedding" - embedding_dimensions: int = 768 - # Embedding models reject input past their context window with a 400, which - # costs the whole embedding — recall then falls back to ILIKE. Clip instead: - # a reaction-session prompt (a full event envelope inlined) failed 41 times. - # 0 disables the cap. - embedding_max_chars: int = 6000 - - openai_api_key: str = "" - openai_model: str = "gpt-4" - openai_base_url: str | None = None - anthropic_api_key: str = "" - - # Web search is not configured here: it is the navi-web MCP server's own - # settings (NAVI_WEB_*), mapped from .env by mcp_servers.d/navi-web.json. - - # Filesystem tool: comma-separated allowed root paths - fs_allowed_paths: str = "*" - - # Terminal tool: "*" = allow all commands (recommended for local use) - # or comma-separated list of allowed executables, e.g. "ls,cat,git" - terminal_allowed_commands: str = "*" - # Terminal allowlist for non-admin users (multi-user mode). - # Admin bypasses this restriction. - # Interpreters stay (python3/node) so scripting work is possible; package managers and - # compilers do not — the user profiles forbid installing packages or changing - # machine-wide configuration, and a toolchain is the easiest way to run foreign code. - terminal_user_allowed_commands: str = ( - "ls,cat,git,grep,egrep,fgrep,head,tail,wc,find,locate," - "ps,df,du,free,uptime,whoami,id,mkdir,touch,cp,mv,rm,chmod," - "echo,printf,sort,uniq,awk,sed,tr,cut,date,uname,hostname,which,whereis," - "python,python3,node,env,export,source" - ) - - # SSH tool: path to JSON file with named connections - ssh_hosts_file: str = "ssh_hosts.json" - - # LLM routing config: targets (hosts) and routes (ordered model@target - # chains). Relative paths resolve against CWD, then the repo root. Absent - # file = single target from OLLAMA_HOST, as in a bare dev checkout. - llm_config_file: str = "llm.json" - - # Database — PostgreSQL is required (SQLite support removed). - database_url: str = "" - - log_level: str = "INFO" - - # Directory for user-defined tools (auto-discovered at startup) - tools_dir: str = "tools" - - # Directory for user-defined context providers (auto-discovered at startup) - context_providers_dir: str = "context_providers" - - # Session file uploads - session_files_dir: str = "session_files" - session_files_max_size_mb: int = 200 - session_messages_window: int = 1000 # max hot messages per session; older -> archive - ws_replay_buffer_size: int = 500 # max events retained for WS reconnect replay - share_file_max_size_mb: int = 1024 - - # Background tasks (see docs/tasks.md). Tools in `backgroundable_tools` - # accept background=true: the call is detached into a TaskManager job and - # the tool call returns a task_id immediately. Tasks are in-memory — a - # restart loses them (same as terminal processes). - tasks_max_per_session: int = 5 # concurrent running tasks per session - tasks_max_global: int = 20 # concurrent running tasks server-wide - tasks_max_spawn: int = 2 # concurrent background spawn_agent per session - tasks_ttl_sec: int = 3600 # finished-task retention before reap - tasks_event_buffer_size: int = 50 # per-task event ring (spawn tasks: 200) - tasks_rate_limit: int = 10 # max task spawns per 5 min per session - task_notes_max_pending: int = 20 # pending completion notes per session - task_notes_per_turn: int = 5 # max notes coalesced into one turn injection - message_queue_max: int = 5 # user messages queued while a run is active - backgroundable_tools: str = "terminal,ssh_exec,peer,spawn_agent,code_exec" - parallel_tool_calls: bool = False # execute a multi-tool-call batch concurrently - - # Public base URL used by share_file tool to build download links. - # Change if the server is behind a reverse proxy or runs on a different port. - public_url: str = "http://localhost:8099" - - # --- Export: bulk history snapshot for an external consumer (n-gram tables) - # Every navi_export_interval_hours one gzip-JSONL file of message text is built - # and age-encrypted to navi_export_age_recipient. The private half of that key - # never reaches this server, so Navi cannot decrypt its own export. - # navi_export_token is the bearer token the consumer presents; only its SHA-256 - # digest is compared. Both fields must be set for the builder to run and - # the endpoints to answer at all. - navi_export_age_recipient: str = "" - navi_export_token: str = "" - navi_export_interval_hours: float = 4.0 - # Which roles make the corpus. Tool results and system prompts are opt-in: - # tool output is scraped web text and command logs (52 MB of prod's 74 MB), - # and system messages are near-duplicate boilerplate — both would drown the - # actual dialogue in an n-gram table. - navi_export_roles: str = "user,assistant" - # Rolling window, whole sessions, newest first, until this much raw text. - # A cap that never binds today (the whole corpus is ~13 MB); it exists so - # the snapshot stays bounded as history grows. - navi_export_max_bytes: int = 150 * 1024 * 1024 - navi_export_dir: str = "export_files" - - # Gmail IMAP/SMTP (App Password auth) - gmail_address: str = "" - gmail_app_password: str = "" - - # gnexus-auth OAuth integration - gnauth_base_url: str = "http://gnexus-auth.local" - gnauth_client_id: str = "" - gnauth_client_secret: str = "" - gnauth_redirect_uri: str = "http://localhost:8099/auth/callback" - gnauth_admin_role_slug: str = "navi_admin" - gnauth_user_role_slug: str = "navi_user" - gnauth_profile_path: str = "/account/profile" # appended to gnauth_base_url for profile links - - # HMAC secret for gnexus-auth webhooks (from the gnexus-auth admin panel). - # When empty and navi_auth_enabled=true, the webhook endpoint refuses to - # process unsigned events (503). When auth is disabled, unsigned webhooks - # are accepted with a warning. - gnauth_webhook_secret: str = "" - - # Master auth switch. When false, Navi skips OAuth/API-token checks and - # treats every request as the anonymous admin user. Use only for trusted - # single-user/local deployments. - navi_auth_enabled: bool = True - - # Default profile used when a client creates a session without specifying - # a profile_id. Empty string means "no default"; the client must supply one. - navi_default_profile_id: str = "" - - # Web UI switch. When false, the static mounts ("/", /assets, /images, - # /content-viewers, /content), the debug/admin panels and the navi_ui MCP - # server are not registered — pure API/WS server for terminal clients. - # The API itself is unaffected; re-enable later by editing .env only. - navi_webclient_enabled: bool = True - - # Bind address for the navi-server launcher (navi/server.py + systemd). - # 127.0.0.1 keeps the server local-only; terminals on other machines - # connect through SSH tunneling or a reverse proxy. - # 8099/8098 instead of the common 8000/8001: navi runs on shared servers - # where those ports are usually taken. - navi_host: str = "127.0.0.1" - navi_port: int = 8099 - - # Internal navi_ui MCP server — lets the agent push structured UI components - # (card_grid, form) to the webclient via the render_component tool. - navi_ui_mcp_enabled: bool = True - 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 - - # Peer-to-peer swarm channel (stage 2). Incoming /peer/ask requests are - # answered by a one-shot agent run under this profile — pick a profile - # with broad but non-destructive tools (default: server_admin). - peer_ask_profile: str = "server_admin" - # Hard ceiling for a single peer ask (the LLM run inside /peer/ask). - peer_ask_timeout_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" - navi_auth_cookie_secure: bool = False - navi_auth_cookie_samesite: str = "lax" - navi_auth_cookie_max_age_days: int = 30 - - # Comma-separated list of origins allowed by CORS when navi_auth_enabled=true - # (cookie-credentialed requests need an explicit origin list — "*" is not - # accepted by browsers together with credentials). Required in auth mode: - # the app refuses to start with an empty list. Ignored when auth is disabled. - navi_allowed_origins: str = "" - - # Web push (PWA). Empty VAPID keys = push fully disabled: no endpoints - # registered, no notifications sent, the client panel shows the toggle - # as unavailable. - navi_push_vapid_public_key: str = "" - navi_push_vapid_private_key: str = "" - navi_push_vapid_subject: str = "mailto:admin@navi.local" - # Minimum seconds between two pushes for the same session (recall + agent - # runs can overlap on one session; also dedups rapid-fire turns). - navi_push_cooldown_sec: int = 30 - - # Synapse ingest — navi as an event SOURCE (outbound reaction notifications). - # Empty api key = navi cannot emit events to Synapse: the settings UI - # disables the synapse push-target options and the notify tool silently - # skips that leg of delivery. - synapse_source_url: str = "" - synapse_source_api_key: str = "" - synapse_source_name: str = "navi" - - # LLM call timeouts - # complete() is non-streaming (compression) — blocked until full response - llm_complete_timeout: int = 120 - # Non-streaming planning calls (the `plan` tool / subagent pipeline). Wider - # than llm_complete_timeout: cloud models prefill big planning prompts for - # up to a couple of minutes, and a timeout here silently kills the plan. - planning_llm_timeout_sec: int = 240 - # stream_complete(): how long to wait for the FIRST token (prefill phase) - # Large contexts can take 60-90s to prefill; 90s matches user expectation - llm_stream_first_chunk_timeout: int = 90 - # stream_complete(): max gap between any two subsequent tokens - llm_stream_chunk_timeout: int = 60 - - # Sub-agent guards. The thinking bound catches a model that degenerates - # into an endless internal loop; 12k chars was reachable in a healthy - # iteration of a cloud reasoning model, so healthy runs were killed - # mid-thought and reported as stalled. - subagent_thinking_stall_seconds: float = 60.0 - subagent_thinking_stall_chars: int = 48_000 - # Whole sub-agent run. 300s routinely cut a long ops task in half; the - # parent can still stop the run, and a background spawn reports separately. - subagent_timeout_seconds: float = 600.0 - # complete()/embed(): attempts per (server, model) on a connection error - # (timeout, refused, DNS) before the fallback gives up. A cloud backend times - # out on its own schedule, and one ReadTimeout from the only configured - # server used to abort the call outright — on prod 2026-10-08 that ended 27 - # calls in a single day, 23 of them memory summarisation. Only connection - # errors are retried; a 4xx/5xx or a missing model is a real answer, not a - # hiccup. - llm_complete_retries: int = 2 - llm_retry_backoff_sec: float = 2.0 - - # Context compression - context_compression_enabled: bool = True - context_compression_threshold: float = 0.90 # trigger at 90% of ollama_num_ctx - context_compression_target: float = 0.65 # hysteresis target — after compression the kept region should fit ~65% of the window (trigger 90% → target 65% leaves real headroom instead of skimming just below the trigger and re-firing a few messages later) - context_keep_recent: int = 4 # conversational turns to keep verbatim - context_summary_temperature: float = 0.3 - context_summary_max_tokens: int = 6000 # max output tokens for the summary LLM call - output_reserve_tokens: int = 2048 # headroom reserved for model response in context checks - # Per-message token budget for the LLM context view. A single tool/assistant - # message whose estimated size exceeds this is head/tail-truncated in the - # built context (never in stored history) so one huge tool result cannot - # alone blow the window. 0 = auto (ollama_num_ctx // 6). - context_message_token_budget: int = 0 - - # Global personality prompt prepended to every agent's system prompt. - # Multi-line values don't survive .env parsing reliably, so prefer - # navi_persona_file (path to a plain .txt file) over inline navi_persona. - navi_persona: str = "" - navi_persona_file: str = "" - - @model_validator(mode="before") - def _load_persona_from_file(cls, values: dict) -> dict: - if not values.get("navi_persona") and values.get("navi_persona_file"): - try: - values["navi_persona"] = Path(values["navi_persona_file"]).read_text( - encoding="utf-8" - ).strip() - except Exception: - pass - return values - - @property - def fs_allowed_paths_list(self) -> list[str]: - return [p.strip() for p in self.fs_allowed_paths.split(",") if p.strip()] - - @property - def navi_export_roles_list(self) -> list[str]: - return [r.strip() for r in self.navi_export_roles.split(",") if r.strip()] - - @property - def navi_export_configured(self) -> bool: - """Both a recipient key and a token are required — no half-configured export.""" - return bool(self.navi_export_age_recipient.strip() and self.navi_export_token.strip()) - - @property - def terminal_allowed_commands_list(self) -> list[str]: - return [c.strip() for c in self.terminal_allowed_commands.split(",") if c.strip()] - - @property - def terminal_user_allowed_commands_list(self) -> list[str]: - return [c.strip() for c in self.terminal_user_allowed_commands.split(",") if c.strip()] - - @property - def navi_allowed_origins_list(self) -> list[str]: - return [o.strip() for o in self.navi_allowed_origins.split(",") if o.strip()] - - -settings = Settings() +settings = bootstrap_settings(Settings) diff --git a/navi/core/container.py b/navi/core/container.py index 30d66a3..11beb71 100644 --- a/navi/core/container.py +++ b/navi/core/container.py @@ -108,7 +108,7 @@ from navi.core.pg_session_store import PgSessionStore from navi.db import Database from navi.memory import MemoryStore - from navi.mcp import McpManager, load_mcp_servers + from navi.mcp import McpManager from navi.mcp.tools import McpTool from navi.skills import ProtocolRunStore, SkillStore from navi.store import KvStore @@ -129,9 +129,14 @@ protocol_store = ProtocolRunStore(pool) scheduler = RecallScheduler(pool) - # Create MCP manager early so build_default_registries can wire it in + # Create MCP manager early so build_default_registries can wire it in. + # The hive overlay is applied to the server definitions here, so a + # centrally added or retargeted server arrives with the next start. + from navi.mcp.config import load_mcp_servers + from navi.overlay import overlay_mcp_configs + mcp_manager = McpManager() - await mcp_manager.load_all() + await mcp_manager.load_all(overlay_mcp_configs(load_mcp_servers())) # BYOK: per-user MCP credentials resolve through the process-wide resolver. from navi.mcp.keystore import get_key_resolver diff --git a/navi/core/registry.py b/navi/core/registry.py index 21e6a09..0ee55c5 100644 --- a/navi/core/registry.py +++ b/navi/core/registry.py @@ -5,7 +5,13 @@ from navi.config import settings from navi.exceptions import ProfileNotFound, ToolNotFound from navi.llm.base import LLMBackend -from navi.llm.config import config_from_host, load_llm_config +from navi.llm.config import ( + config_from_host, + load_llm_raw, + parse_llm_config, + resolve_config_path, +) +from navi.overlay import apply_profiles_list, overlay_llm_raw from navi.llm.fallback import FallbackOllamaBackend from navi.profiles import ALL_PROFILES from navi.profiles.base import AgentProfile @@ -175,7 +181,20 @@ # Ollama backend (primary). Routes come from llm.json — targets (hosts) and # ordered model@target chains. Without the file the backend falls back to a # single target from OLLAMA_HOST, which keeps a bare dev checkout working. - llm_config = load_llm_config(settings.llm_config_file) + # The hive overlay is merged into the raw document, so a centrally changed + # host or route does not have to be written to every machine's file. + local_raw = load_llm_raw(settings.llm_config_file) + raw_llm = overlay_llm_raw(local_raw) + llm_config = None + if raw_llm is not None: + llm_config = parse_llm_config( + raw_llm, + source=( + str(resolve_config_path(settings.llm_config_file)) + if local_raw is not None + else "hive-overlay" + ), + ) if llm_config is not None: log.info( "llm.config_loaded", @@ -231,7 +250,7 @@ ) profiles = ProfileRegistry() - for p in ALL_PROFILES: + for p in apply_profiles_list(ALL_PROFILES): profiles.register(p) cp_registry = ContextProviderRegistry() diff --git a/navi/identity.py b/navi/identity.py index cb72ee5..6dcb455 100644 --- a/navi/identity.py +++ b/navi/identity.py @@ -60,6 +60,9 @@ name: str instance_id: str created_at: str + # Optional labels from instance.json, used by the hive's config layers to + # target a group of machines ("lab", "prod") without naming each one. + tags: tuple[str, ...] = () def generate_name() -> str: @@ -80,6 +83,7 @@ name=str(data["name"]), instance_id=str(data["instance_id"]), created_at=str(data.get("created_at", "")), + tags=tuple(str(t) for t in (data.get("tags") or []) if str(t).strip()), ) diff --git a/navi/llm/config.py b/navi/llm/config.py index 837f731..522c261 100644 --- a/navi/llm/config.py +++ b/navi/llm/config.py @@ -188,13 +188,13 @@ return _PROJECT_ROOT / p -def load_llm_config(path: str | Path) -> LLMConfig | None: - """Load and validate ``llm.json``; ``None`` when the file is absent. +def load_llm_raw(path: str | Path) -> dict | None: + """Read ``llm.json`` as a plain document; ``None`` when there is none. - A missing file is not an error — the backend then falls back to a single - target built from ``OLLAMA_HOST``, which is what a bare dev checkout has. - A *broken* file is an error: silently routing to the wrong host is worse - than refusing to start. + Split out from parsing so a hive overlay can be merged *into the local + document* before it is validated — merging at the object level would + lose the distinction between "the hive said nothing" and "the hive set + the local value". """ if not str(path).strip(): return None # explicitly disabled (LLM_CONFIG_FILE="") @@ -208,7 +208,23 @@ if not isinstance(raw, dict): raise LLMConfigError(f"{resolved}: expected a JSON object") + return raw + +def parse_llm_config( + raw: dict, + *, + source: str = "", + resolve_secrets: bool = True, +) -> LLMConfig: + """Validate a raw ``llm.json`` document. + + ``resolve_secrets=False`` is for the overlay validator: it runs while + ``navi.config`` is still being constructed, and resolving an + ``api_key_env`` would reach back into that half-built module. Names are + resolved again when the backends are actually built. + """ + resolved = source or "" targets_raw = raw.get("targets") or {} if not isinstance(targets_raw, dict) or not targets_raw: raise LLMConfigError(f"{resolved}: 'targets' must be a non-empty object") @@ -218,7 +234,7 @@ if not isinstance(spec, dict) or not spec.get("host"): raise LLMConfigError(f"{resolved}: target '{name}' needs a 'host'") api_key = spec.get("api_key", "") - if not api_key and spec.get("api_key_env"): + if resolve_secrets and not api_key and spec.get("api_key_env"): api_key = _resolve_secret(spec["api_key_env"]) if not api_key: log.warning( @@ -270,6 +286,20 @@ ) +def load_llm_config(path: str | Path) -> LLMConfig | None: + """Load and validate ``llm.json``; ``None`` when the file is absent. + + A missing file is not an error — the backend then falls back to a single + target built from ``OLLAMA_HOST``, which is what a bare dev checkout has. + A *broken* file is an error: silently routing to the wrong host is worse + than refusing to start. + """ + raw = load_llm_raw(path) + if raw is None: + return None + return parse_llm_config(raw, source=str(resolve_config_path(path))) + + def config_from_host(host: str, api_key: str = "") -> LLMConfig: """Legacy single-target config for a deployment without ``llm.json``. diff --git a/navi/main.py b/navi/main.py index 68574bb..b9f0559 100644 --- a/navi/main.py +++ b/navi/main.py @@ -13,7 +13,7 @@ from fastapi.staticfiles import StaticFiles from navi.api.deps import require_admin -from navi.api.routes import agents, api_tokens, auth, export, health, messages, mcp_keys, peer, push, sessions, skills, webhooks, synapse +from navi.api.routes import agents, api_tokens, auth, config_overlay, export, health, messages, mcp_keys, peer, push, sessions, skills, webhooks, synapse from navi.api.routes.admin import router as admin_router from navi.api.websocket import router as ws_router from navi.config import settings @@ -32,6 +32,43 @@ log = structlog.get_logger() +async def _apply_profile_overrides(container) -> None: + """Re-apply the ``is_admin_only`` flags stored in the DB. + + Profiles are rebuilt from their local files whenever the hive overlay is + applied, which would otherwise silently undo an operator's availability + change on every config refresh. + """ + from navi.profiles._overrides import ensure_table, load_overrides + + try: + pool = await container.database.pool() + await ensure_table(pool) + overrides = await load_overrides(pool) + for pid, is_admin_only in overrides.items(): + try: + container.profile_registry.get(pid).is_admin_only = is_admin_only + except Exception: + pass + if overrides: + log.info("startup.profile_overrides_applied", count=len(overrides)) + except Exception: + log.warning("startup.profile_overrides_failed", exc_info=True) + + +_override_context: dict = {"loop": None, "container": None} + + +def _reapply_profile_overrides_hook() -> None: + """Overlay post-apply hook — hops from the refresh thread to the loop.""" + context = _override_context + if context["loop"] is None or context["container"] is None: + return + asyncio.run_coroutine_threadsafe( + _apply_profile_overrides(context["container"]), context["loop"] + ) + + def _is_transient_db_error(e: Exception) -> bool: """Connection-level failures a starting Postgres clears on its own.""" import asyncpg @@ -142,7 +179,6 @@ from navi.content_store import ensure_tables from navi.session_files import cleanup_loop from navi.auth import _ensure_auth_tables - from navi.profiles._overrides import ensure_table, load_overrides from navi.api.routes.health import _check_embed from navi.core.scheduler import recall_scheduler_loop from navi.core.pg_session_store import pending_sweep_loop @@ -168,20 +204,7 @@ log.error("startup.ensure_tables_failed", error=str(e)) # Apply persisted profile overrides - try: - pool = await container.database.pool() - await ensure_table(pool) - overrides = await load_overrides(pool) - if overrides: - for pid, is_admin_only in overrides.items(): - try: - profile = container.profile_registry.get(pid) - profile.is_admin_only = is_admin_only - except Exception: - pass - log.info("startup.profile_overrides_applied", count=len(overrides)) - except Exception: - log.warning("startup.profile_overrides_failed", exc_info=True) + await _apply_profile_overrides(container) # Check embedding backend health embed_status = await _check_embed() @@ -224,6 +247,19 @@ set_announcer(announcer) announce_task = asyncio.create_task(announcer.run()) + # A hive notify arrives on a worker thread (the refresh uses blocking + # HTTP); DB work has to come back to this loop. The context is a module + # global so the hook itself is registered once per process, not per start. + from navi.overlay import confirm_startup, register_apply_hook + + _override_context["loop"] = asyncio.get_running_loop() + _override_context["container"] = container + register_apply_hook(_reapply_profile_overrides_hook) + + # The process is up with whatever the overlay chose — that config works, + # so the boot-loop guard can mark it confirmed. + confirm_startup(settings) + yield # Shutdown @@ -295,6 +331,7 @@ app.include_router(health.router) app.include_router(peer.router) +app.include_router(config_overlay.router) app.include_router(auth.router) app.include_router(api_tokens.router) app.include_router(agents.router) diff --git a/navi/overlay/__init__.py b/navi/overlay/__init__.py new file mode 100644 index 0000000..7e8a1e7 --- /dev/null +++ b/navi/overlay/__init__.py @@ -0,0 +1,54 @@ +"""Central configuration overlay: the hive's config on top of the local one. + +The hive (``hive/app.py``) stores one versioned document. An instance asks +for the layers that target it, merges them in order, and applies the result +over its own config — only the fields the hive names, everything else from +the local files. No document, or a document with no matching layer, means the +local config stands unchanged. + +Reading order in ``navi.config``:: + + navi.config.settings = bootstrap_settings(Settings) + +which captures the local base, fetches (with the last-good cache as a +fallback), validates, and returns the overlaid settings object. The other +sections are picked up where they are built: ``apply_profiles_list`` in the +profile registry, ``overlay_llm_raw`` in the LLM registry, +``overlay_mcp_configs`` in the container. + +A live push (``POST /config/notify`` from the hive) calls ``refresh``: hot +settings are rewritten in the live singleton, profiles are swapped in the +registry, and anything that needs a restart is reported as such instead of +being pretended into effect. +""" + +from __future__ import annotations + +from navi.overlay.apply import ( + apply_profiles_list, + apply_profiles_registry, + apply_settings_live, + bootstrap_settings, + overlay_llm_raw, + overlay_mcp_configs, + refresh, + register_apply_hook, +) +from navi.overlay.fetch import confirm_startup +from navi.overlay.state import get_state, reset_state, status_dict, status_summary + +__all__ = [ + "apply_profiles_list", + "apply_profiles_registry", + "apply_settings_live", + "bootstrap_settings", + "confirm_startup", + "get_state", + "overlay_llm_raw", + "overlay_mcp_configs", + "refresh", + "register_apply_hook", + "reset_state", + "status_dict", + "status_summary", +] diff --git a/navi/overlay/apply.py b/navi/overlay/apply.py new file mode 100644 index 0000000..65792ac --- /dev/null +++ b/navi/overlay/apply.py @@ -0,0 +1,407 @@ +"""Put a validated overlay into effect — at startup and on a live refresh. + +Startup is simple: settings are constructed once, from base values plus the +overlay, and the other sections are picked up by the registry/container while +they build. + +A live refresh is where the distinction between hot and staged matters: + + - **profiles** are rebuilt from the local files and swapped in the registry. + The agent reads a profile per run, so the next LLM call sees it — the same + guarantee ``switch_profile`` and the admin profile editor already rely on; + - **hot settings** (secrets, timeouts, guards — see ``schema.py``) are + written into the live singleton. Every module holds the same object and + reads attributes at call time, so rotating a key takes effect without a + restart; + - **everything else** is reported as ``restart_required``. Claiming an + effect that only happens on restart would be worse than asking for one. +""" + +from __future__ import annotations + +import os +import time +from typing import Callable + +import structlog + +from navi.overlay.fetch import Fetched, fetch_layers, store_applied +from navi.overlay.merge import merge_patch +from navi.overlay.schema import HOT_SETTINGS +from navi.overlay.state import get_state, lock, status_summary +from navi.overlay.validate import validate_layers + +log = structlog.get_logger() + +_apply_hooks: list[Callable[[], None]] = [] + + +def register_apply_hook(fn: Callable[[], None]) -> None: + """Run *fn* after hot settings changed (caches built from old values).""" + if fn not in _apply_hooks: + _apply_hooks.append(fn) + + +def _run_apply_hooks() -> None: + # Built-in: the default OAuth client caches client_id/secret at creation. + try: + from navi.auth.client import reset_default_client + + reset_default_client() + except Exception: + log.warning("hive.config.oauth_client_reset_failed", exc_info=True) + for hook in list(_apply_hooks): + try: + hook() + except Exception: + log.warning("hive.config.apply_hook_failed", hook=repr(hook), exc_info=True) + + +# ── settings ──────────────────────────────────────────────────────────────── + + +def build_settings(base, patch: dict): + """Construct the process settings from local base + overlay settings.""" + if not patch: + return base + return type(base)(**{**base.model_dump(), **patch}) + + +def _set_in_place(settings_obj, name: str, value: object) -> None: + """Write one field into the frozen settings singleton. + + ``frozen=True`` forbids ``__setattr__``; ``object.__setattr__`` is the one + supported escape hatch and is used deliberately, in one place, so a hot + refresh is visible to every module that imported the object. + """ + object.__setattr__(settings_obj, name, value) + + +def apply_settings_live(merged_settings: dict) -> tuple[list[str], list[str]]: + """Apply hot fields in place; return (hot_applied, restart_required). + + Values removed from the overlay fall back to the local base, so a field + can be handed back without editing any file. + """ + state = get_state() + base = state.base_settings + previous = dict((state.merged or {}).get("settings") or {}) + if base is None: + return [], [] + + from navi.config import settings as live + + hot: list[str] = [] + restart: list[str] = [] + for name in set(previous) | set(merged_settings): + if not hasattr(base, name): + continue + target = merged_settings.get(name, getattr(base, name)) + if getattr(live, name) == target: + continue + if name in HOT_SETTINGS: + _set_in_place(live, name, target) + hot.append(name) + else: + restart.append(name) + + if hot: + _run_apply_hooks() + log.info("hive.config.hot_settings_applied", fields=sorted(hot)) + return sorted(hot), sorted(restart) + + +# ── profiles ──────────────────────────────────────────────────────────────── + + +def apply_profiles_list(profiles: list): + """Return profiles with the overlay applied (used while building the registry).""" + patch = (get_state().merged or {}).get("profiles") or {} + if not patch: + return profiles + from navi.profiles.base import AgentProfile + + out = [] + for profile in profiles: + if profile.id in patch: + merged = merge_patch(profile.model_dump(), patch[profile.id]) + out.append(AgentProfile.model_validate(merged)) + else: + out.append(profile) + return out + + +def apply_profiles_registry(registry) -> None: + """Rebuild every known profile from its local base and swap it in place. + + Rebuilding from the base (not from the current object) is what makes a + refresh idempotent: a field dropped from the overlay is dropped from the + running profile too, instead of lingering from the previous overlay. + """ + patch = (get_state().merged or {}).get("profiles") or {} + from navi.profiles import ALL_PROFILES + from navi.profiles.base import AgentProfile + + base_by_id = {p.id: p for p in ALL_PROFILES} + changed = False + for current in registry.all(): + base = base_by_id.get(current.id) + if base is None: + continue # created at runtime through the admin API — not ours + merged = merge_patch(base.model_dump(), patch.get(current.id, {})) + registry.update(AgentProfile.model_validate(merged)) + changed = True + if changed: + try: + from navi.api.deps import get_agent + + get_agent().invalidate_prompt_cache() + except Exception: + log.warning("hive.config.prompt_cache_invalidate_failed", exc_info=True) + log.info("hive.config.profiles_applied", profiles=sorted(patch)) + + +# ── llm / mcp ─────────────────────────────────────────────────────────────── + + +def overlay_llm_raw(raw: dict | None) -> dict | None: + patch = (get_state().merged or {}).get("llm") + if not patch: + return raw + return merge_patch(raw or {}, patch) # type: ignore[return-value] + + +def overlay_mcp_configs(configs: dict) -> dict: + patch = (get_state().merged or {}).get("mcp_servers") + if patch is None: + return configs + from navi.mcp.config import McpServerConfig + + dumped = {name: cfg.model_dump() for name, cfg in configs.items()} + merged = merge_patch(dumped, patch) + return { + name: McpServerConfig.model_validate(value) + for name, value in merged.items() + if isinstance(value, dict) + } + + +# ── live refresh ──────────────────────────────────────────────────────────── + + +def _ack(base_settings, fetched: Fetched, status: str, restart_required: list[str], + errors: list[str]) -> None: + if not base_settings.hive_url: + return + from navi.overlay.fetch import _identity_params, _read_key + + key = _read_key(base_settings.swarm_key_file) + identity = _identity_params(base_settings.instance_file).get("instance_id", "") + if not key or not identity: + return + import httpx + + payload = { + "instance_id": identity, + "hash": fetched.hash, + "version": fetched.version, + "status": status, + "restart_required": restart_required, + "error": errors[0] if errors else None, + } + try: + with httpx.Client(timeout=base_settings.hive_config_timeout_sec) as client: + client.post( + base_settings.hive_url.rstrip("/") + "/config/ack", + json=payload, + headers={"X-Swarm-Key": key}, + ) + except Exception as exc: # the announce payload is the redundant path + log.info("hive.config.ack_failed", error=f"{type(exc).__name__}: {exc}") + + +def refresh(trigger: str = "notify") -> dict: + """Fetch, validate and apply the current hive config. Never raises.""" + state = get_state() + base = state.base_settings + if base is None or not state.enabled: + return {"status": "disabled", "trigger": trigger} + if not lock.acquire(blocking=False): + return {"status": "busy", "trigger": trigger} + + try: + try: + fetched = fetch_layers(base) + except Exception as exc: + state.status = "unreachable" + state.last_error = f"{type(exc).__name__}: {exc}" + log.warning("hive.config.fetch_failed", trigger=trigger, error=state.last_error) + return {"status": "unreachable", "error": state.last_error, "trigger": trigger} + + if fetched is None: + # No config channel (older hive) or no key — keep what we run. + state.last_error = "hive has no config endpoint" + return {"status": state.status, "trigger": trigger} + + validation = validate_layers(base, fetched.layers) + state.last_fetch = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + state.warnings = validation.warnings + state.unknown_fields = validation.unknown_fields + state.forbidden_fields = validation.forbidden_fields + + if not validation.ok: + state.status = "rejected" + state.errors = validation.errors + state.last_error = validation.errors[0] if validation.errors else None + log.error( + "hive.config.rejected", + trigger=trigger, + version=fetched.version, + hash=fetched.hash, + errors=validation.errors, + ) + # Tell the hive, so a rejected document is visible in its status + # view instead of looking like a machine that never answered. + _ack(base, fetched, "rejected", [], validation.errors) + return { + "status": "rejected", + "version": fetched.version, + "hash": fetched.hash, + "error": state.last_error, + "trigger": trigger, + } + + # Settings first (they may be read by the other sections). The apply + # compares against the *previous* merged document to know which fields + # were handed back to local values, so it must run before we replace + # ``state.merged``. + hot, restart = apply_settings_live(validation.merged.get("settings") or {}) + state.merged = validation.merged + apply_profiles_registry_from_state() + + state.version = fetched.version + state.hash = fetched.hash + state.source = "hive" + state.layers = [layer["name"] for layer in fetched.layers] + state.errors = [] + state.last_error = None + state.restart_required = restart + state.hot_applied = hot + state.status = "staged" if restart else "applied" + store_applied(base, fetched) + _ack(base, fetched, state.status, restart, []) + + log.info( + "hive.config.applied", + trigger=trigger, + version=fetched.version, + hash=fetched.hash, + status=state.status, + hot=hot, + restart_required=restart, + unknown_fields=validation.unknown_fields, + forbidden_fields=validation.forbidden_fields, + ) + return {"status": state.status, **status_summary(), "trigger": trigger} + finally: + lock.release() + + +def apply_profiles_registry_from_state() -> None: + """Refresh the live profile registry; safe to call before it exists.""" + try: + from navi.api.deps import get_profile_registry + + apply_profiles_registry(get_profile_registry()) + except Exception: + # Startup imports the settings long before the container exists; there + # the registry is built from the overlay directly (apply_profiles_list). + log.debug("hive.config.profile_registry_unavailable") + + +# ── startup ───────────────────────────────────────────────────────────────── + + +def overlay_disabled_by_env() -> bool: + return os.environ.get("NAVI_OVERLAY", "").strip().lower() in {"off", "0", "false", "no"} + + +def bootstrap_settings(settings_cls): + """Build the process settings: local base, then the hive overlay.""" + from navi.overlay.fetch import resolve_startup + + base = settings_cls() + state = get_state() + state.base_settings = base + state.configured = bool(base.hive_url) + + if overlay_disabled_by_env() or not base.hive_config_enabled: + state.enabled = False + state.status = "disabled" + return base + if not base.hive_url: + state.enabled = False + state.status = "disabled" + return base + + try: + result = resolve_startup(base) + except Exception as exc: + state.status = "unreachable" + state.last_error = f"{type(exc).__name__}: {exc}" + log.exception("hive.config.startup_failed") + return base + + state.boot_guard = result.boot_guard + if result.validation is not None: + state.warnings = result.validation.warnings + state.unknown_fields = result.validation.unknown_fields + state.forbidden_fields = result.validation.forbidden_fields + state.errors = result.validation.errors + if result.error: + state.last_error = result.error + + if not result.layers: + # Nothing to apply. Two different situations share this branch: the + # hive answered "no layers for you" (source="hive" — a real answer, + # the local config is final) and the hive was unreachable or the + # document was rejected (source="local" — try again next boot). + state.source = result.source + if result.source == "hive": + state.last_error = None + state.status = "applied" + state.version = result.version + state.hash = result.hash + state.layers = [] + state.last_fetch = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + else: + state.status = "unreachable" if result.error else "not_targeted" + return base + + try: + overlaid = build_settings(base, result.merged.get("settings") or {}) + except Exception as exc: # validation already passed; belt and braces + state.status = "rejected" + state.errors = [f"settings: {exc}"] + state.last_error = state.errors[0] + log.exception("hive.config.settings_build_failed") + return base + + state.merged = result.merged + state.source = result.source + state.last_error = None + state.version = result.version + state.hash = result.hash + state.layers = [layer["name"] for layer in result.layers] + state.status = "applied" + state.last_fetch = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + log.info( + "hive.config.loaded", + source=result.source, + version=result.version, + hash=result.hash, + layers=state.layers, + unknown_fields=state.unknown_fields, + forbidden_fields=state.forbidden_fields, + ) + return overlaid diff --git a/navi/overlay/fetch.py b/navi/overlay/fetch.py new file mode 100644 index 0000000..bedcfa3 --- /dev/null +++ b/navi/overlay/fetch.py @@ -0,0 +1,356 @@ +"""Getting the overlay from the hive, and surviving the cases where it fails. + +Two entry points: + +``resolve_startup`` runs while ``navi.config`` is being imported. It fetches +with a short timeout, falls back to the last validated overlay on disk, and +runs the boot-loop guard: a config that was written as ``pending`` and never +confirmed means the previous start died with it, so it is demoted after a +second failure instead of crash-looping forever. + +``refresh`` runs in a live process, triggered by a hive notification or by an +announce reply. It never raises into the caller: every failure becomes a +status line, because a broken config channel must not break the server that +is supposed to keep running. + +The cache keeps the last **validated** overlay. A newly fetched document that +fails validation never overwrites it — that is what "last known good" means +here. +""" + +from __future__ import annotations + +import json +import os +import time +from dataclasses import dataclass, field +from pathlib import Path + +import structlog + +from navi.overlay.state import get_state +from navi.overlay.util import canonical_json, content_hash +from navi.overlay.validate import ValidationResult, validate_envelope, validate_layers + +log = structlog.get_logger() + +_CACHE_VERSION = 1 + + +@dataclass +class Fetched: + version: int + layers: list[dict] + hash: str + note: str = "" + + +@dataclass +class CachedState: + good: dict | None = None + pending: dict | None = None + blocked_hash: str = "" + + +@dataclass +class StartupResult: + layers: list[dict] = field(default_factory=list) + source: str = "local" + version: int = 0 + hash: str = "" + merged: dict = field(default_factory=dict) + validation: ValidationResult | None = None + error: str | None = None + boot_guard: bool = False + + +# ── local helpers ─────────────────────────────────────────────────────────── + + +def _cache_path(base_settings) -> Path: + return Path(base_settings.hive_config_cache_file).expanduser() + + +def _read_key(path: str | Path) -> str | None: + try: + value = Path(path).read_text(encoding="utf-8").strip() + except OSError: + return None + return value or None + + +def _identity_params(path: str | Path) -> dict[str, str]: + """Name, id and tags from instance.json. + + Not ``ensure_identity``: at import time the file may legitimately not + exist yet (first start), and a config fetch must not create it as a side + effect. Without identity the hive still returns the global layers. + """ + try: + data = json.loads(Path(path).read_text(encoding="utf-8")) + except (OSError, ValueError): + return {} + params: dict[str, str] = {} + for key in ("name", "instance_id"): + value = data.get(key) + if value: + params[key] = str(value) + tags = data.get("tags") + if isinstance(tags, list) and tags: + params["tags"] = ",".join(str(t) for t in tags) + return params + + +def load_cached(path: Path) -> CachedState: + try: + raw = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return CachedState() + if not isinstance(raw, dict): + return CachedState() + + def _slot(name: str) -> dict | None: + value = raw.get(name) + if not isinstance(value, dict) or not isinstance(value.get("layers"), list): + return None + return value + + return CachedState( + good=_slot("good"), + pending=_slot("pending"), + blocked_hash=str(raw.get("blocked_hash") or ""), + ) + + +def save_cached(path: Path, state: CachedState) -> None: + """Atomic replace; a cache that cannot be written only costs a fetch.""" + payload = { + "cache_version": _CACHE_VERSION, + "good": state.good, + "pending": state.pending, + "blocked_hash": state.blocked_hash, + } + tmp = path.with_name(path.name + ".tmp") + try: + tmp.write_text( + canonical_json(payload) + "\n", + encoding="utf-8", + ) + try: + os.chmod(tmp, 0o600) + except OSError: + pass + os.replace(tmp, path) + except OSError: + log.warning("hive.config.cache_write_failed", path=str(path)) + + +# ── HTTP ──────────────────────────────────────────────────────────────────── + + +def fetch_layers(base_settings) -> Fetched | None: + """GET the overlay for this instance. + + ``None`` means the hive has no config endpoint (an older hive) or answered + with something unusable; an empty layer list is a real answer ("publish + nothing"), which reverts an instance to its local config. + + A missing swarm key raises: ``HIVE_URL`` without a key is a misconfigured + channel, not the hive saying "no overlay for you", and the status view + should send the operator to the key file rather than call the machine + untargeted. + """ + if not base_settings.hive_url: + return None + key = _read_key(base_settings.swarm_key_file) + if not key: + log.warning( + "hive.config.key_missing", + key_file=base_settings.swarm_key_file, + detail="HIVE_URL is set but there is no swarm key; central config disabled", + ) + raise RuntimeError(f"swarm key missing: {base_settings.swarm_key_file}") + + import httpx + + url = base_settings.hive_url.rstrip("/") + "/config" + params = _identity_params(base_settings.instance_file) + try: + with httpx.Client(timeout=base_settings.hive_config_timeout_sec) as client: + resp = client.get(url, params=params, headers={"X-Swarm-Key": key}) + except Exception as exc: + raise RuntimeError(f"{type(exc).__name__}: {exc}") from exc + + if resp.status_code == 404: + return None + if resp.status_code != 200: + raise RuntimeError(f"HTTP {resp.status_code}: {resp.text[:200]}") + + try: + data = resp.json() + except ValueError as exc: + raise RuntimeError(f"invalid JSON from hive: {exc}") from exc + if not isinstance(data, dict): + raise RuntimeError("hive answered with a non-object") + + raw_layers = data.get("layers", []) + layers, _ = validate_envelope(raw_layers) + version = data.get("version", 0) + if not isinstance(version, int) or isinstance(version, bool): + version = 0 + # Trust the bytes, not the header: the hash is what acks are compared by. + return Fetched(version=version, layers=layers, hash=content_hash(layers)) + + +# ── startup ───────────────────────────────────────────────────────────────── + + +def resolve_startup(base_settings) -> StartupResult: + """Pick the overlay this process starts with. + + Candidates in order: an unconfirmed ``pending`` from the previous start + (one retry), the freshly fetched overlay, the last known good from disk. + The first one that validates wins; if none does, the local config stands. + """ + path = _cache_path(base_settings) + cached = load_cached(path) + + fetched: Fetched | None = None + fetch_error: str | None = None + try: + fetched = fetch_layers(base_settings) + except Exception as exc: + fetch_error = str(exc) + log.warning("hive.config.fetch_failed", error=fetch_error) + + if fetched is not None and cached.blocked_hash and fetched.hash != cached.blocked_hash: + # A new document was published — the guard was about one bad config, + # not about the channel. + cached.blocked_hash = "" + + boot_guard = False + pending = cached.pending + if pending is not None: + attempts = int(pending.get("attempts") or 0) + if attempts >= 2: + boot_guard = True + cached.blocked_hash = str(pending.get("hash") or "") + cached.pending = None + save_cached(path, cached) + log.error( + "hive.config.boot_guard", + hash=cached.blocked_hash, + detail="the previous config prevented startup twice; keeping the last good one", + ) + pending = None + + candidates: list[tuple[str, Fetched]] = [] + if pending is not None: + candidates.append( + ( + "cache", + Fetched( + version=int(pending.get("version") or 0), + layers=list(pending.get("layers") or []), + hash=str(pending.get("hash") or ""), + ), + ) + ) + if fetched is not None: + candidates.append(("hive", fetched)) + if cached.good is not None: + candidates.append( + ( + "cache", + Fetched( + version=int(cached.good.get("version") or 0), + layers=list(cached.good.get("layers") or []), + hash=str(cached.good.get("hash") or ""), + ), + ) + ) + + seen: set[str] = set() + last_validation: ValidationResult | None = None + for source, item in candidates: + if not item.hash: + item.hash = content_hash(item.layers) + if item.hash in seen or item.hash == cached.blocked_hash: + continue + seen.add(item.hash) + validation = validate_layers(base_settings, item.layers) + if not validation.ok: + last_validation = validation + log.error( + "hive.config.rejected", + source=source, + version=item.version, + hash=item.hash, + errors=validation.errors, + ) + continue + + # A config that differs from the last confirmed one is "pending" + # until the lifespan reaches its end; that is the boot-loop guard. + if cached.good is None or item.hash != cached.good.get("hash"): + previous = cached.pending or {} + attempts = ( + int(previous.get("attempts") or 0) + 1 + if previous.get("hash") == item.hash + else 1 + ) + cached.pending = { + "version": item.version, + "hash": item.hash, + "layers": item.layers, + "attempts": attempts, + } + else: + cached.pending = None + save_cached(path, cached) + + return StartupResult( + layers=item.layers, + source=source, + version=item.version, + hash=item.hash, + merged=validation.merged, + validation=validation, + error=fetch_error, + boot_guard=boot_guard, + ) + + # Nothing usable. Keep whatever the cache says is good out of the way — + # the local config is the base, and the next boot will try again. + return StartupResult( + source="local", + validation=last_validation, + error=fetch_error, + boot_guard=boot_guard, + ) + + +def confirm_startup(base_settings) -> None: + """The process reached the end of startup: the pending config works.""" + path = _cache_path(base_settings) + cached = load_cached(path) + if cached.pending is None: + return + cached.good = cached.pending + cached.pending = None + save_cached(path, cached) + log.info("hive.config.confirmed", version=cached.good.get("version")) + + +def store_applied(base_settings, fetched: Fetched) -> None: + """A live refresh succeeded — this is now the config to boot from.""" + path = _cache_path(base_settings) + cached = load_cached(path) + cached.good = { + "version": fetched.version, + "hash": fetched.hash, + "layers": fetched.layers, + "fetched_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + } + cached.pending = None + save_cached(path, cached) diff --git a/navi/overlay/merge.py b/navi/overlay/merge.py new file mode 100644 index 0000000..726b32d --- /dev/null +++ b/navi/overlay/merge.py @@ -0,0 +1,33 @@ +"""RFC 7386 JSON merge patch — the one merge rule the overlay uses. + +Why merge patch rather than a bespoke merger: + + - an absent field means "no opinion" — the local value stays (the whole + point of the overlay); + - ``null`` means "drop the override", so a field can be handed back to the + local config without deleting anything from the local files; + - objects merge recursively, which is what per-profile and per-MCP-server + overlays need; + - arrays are replaced wholesale — "add one tool" spells out the list, and + nothing silently reorders a toolset. + +The function never mutates its arguments: callers keep the local base intact +and can re-merge at any time (a refresh must not stack overlay on overlay). +""" + +from __future__ import annotations + +import copy + + +def merge_patch(target: object, patch: object) -> object: + """Return *patch* applied over *target*, per RFC 7386.""" + if not isinstance(patch, dict): + return copy.deepcopy(patch) + result: dict = copy.deepcopy(target) if isinstance(target, dict) else {} + for key, value in patch.items(): + if value is None: + result.pop(key, None) + else: + result[key] = merge_patch(result.get(key), value) + return result diff --git a/navi/overlay/schema.py b/navi/overlay/schema.py new file mode 100644 index 0000000..f02e131 --- /dev/null +++ b/navi/overlay/schema.py @@ -0,0 +1,108 @@ +"""Which settings the central overlay may touch, and which take effect live. + +Two lists, and the asymmetry between them is deliberate. + +``DENIED_SETTINGS`` are fields a hive document can never change. They are not +a matter of taste: the bootstrap ones (``HIVE_URL``, the key path, the cache +path) are needed *before* the overlay can be fetched, so an overlay that set +them could never describe the channel it arrives through. The rest are the +ways a config can make every machine unreachable or silently point it at the +wrong database — a valid value that is merely wrong is the one failure class +validation cannot catch. + +``HOT_SETTINGS`` are fields read from the singleton at call time, so writing a +new value into the live object takes effect immediately. Everything else +changes something built at startup (CORS middleware, backends, the MCP pool, +the push service) and is reported as ``restart_required`` instead of being +pretended into effect. +""" + +from __future__ import annotations + +# Never overlaid. Unknown to the hive's validator? No — known, and refused. +DENIED_SETTINGS: frozenset[str] = frozenset( + { + # Bootstrap: needed before the overlay exists. + "hive_url", + "swarm_key_file", + "instance_file", + "hive_config_enabled", + "hive_config_cache_file", + "hive_config_timeout_sec", + # Reachability and data: a wrong-but-valid value is unrecoverable + # without a local edit (or, for the DB, looks like lost history). + "database_url", + "navi_host", + "navi_port", + "navi_ui_mcp_host", + "navi_ui_mcp_port", + "navi_auth_enabled", + } +) + +# Applied to the live settings object on refresh; everything else is staged +# until the next start. Only fields verified to be read at call time belong +# here — a wrong entry would claim an effect that never happens. +HOT_SETTINGS: frozenset[str] = frozenset( + { + # gnexus-auth: read per request / per login. The cached default OAuth + # client is reset by an apply hook (navi/auth/client.py). + "gnauth_base_url", + "gnauth_client_id", + "gnauth_client_secret", + "gnauth_redirect_uri", + "gnauth_admin_role_slug", + "gnauth_user_role_slug", + "gnauth_profile_path", + "gnauth_webhook_secret", + # Synapse outbound: a client is built per emit. + "synapse_source_url", + "synapse_source_api_key", + "synapse_source_name", + # Read at call time. + "public_url", + "navi_default_profile_id", + "navi_push_cooldown_sec", + "peer_ask_timeout_sec", + "session_files_max_size_mb", + "share_file_max_size_mb", + "terminal_allowed_commands", + "terminal_user_allowed_commands", + "fs_allowed_paths", + "backgroundable_tools", + "parallel_tool_calls", + "message_queue_max", + "tasks_max_per_session", + "tasks_max_global", + "tasks_max_spawn", + "tasks_ttl_sec", + "tasks_event_buffer_size", + "tasks_rate_limit", + "task_notes_max_pending", + "task_notes_per_turn", + # LLM timeouts / retries. + "llm_complete_timeout", + "planning_llm_timeout_sec", + "llm_stream_first_chunk_timeout", + "llm_stream_chunk_timeout", + "llm_complete_retries", + "llm_retry_backoff_sec", + # Sub-agent guards. + "subagent_thinking_stall_seconds", + "subagent_thinking_stall_chars", + "subagent_timeout_seconds", + # Context compression. + "context_compression_enabled", + "context_compression_threshold", + "context_compression_target", + "context_keep_recent", + "context_summary_temperature", + "context_summary_max_tokens", + "output_reserve_tokens", + "context_message_token_budget", + } +) + +KNOWN_SECTIONS: frozenset[str] = frozenset( + {"settings", "profiles", "llm", "mcp_servers"} +) diff --git a/navi/overlay/state.py b/navi/overlay/state.py new file mode 100644 index 0000000..d13b3e0 --- /dev/null +++ b/navi/overlay/state.py @@ -0,0 +1,80 @@ +"""Process-wide overlay state — what was fetched, what applied, what failed. + +One object, written under a lock by the fetch/apply path and read by /health, +the swarm announce payload, the notify endpoint and the admin panel. It is a +plain summary: the document itself lives in ``merged`` only for as long as the +process runs, and in the cache file between runs. +""" + +from __future__ import annotations + +import threading +from dataclasses import asdict, dataclass, field + +# One refresh at a time. A hive notification and an announce hint can arrive +# together; stacking two applies would let an older document overwrite a newer +# one, and the second validate against a half-updated base. +lock = threading.Lock() + + +@dataclass +class OverlayState: + configured: bool = False # HIVE_URL + swarm key present + enabled: bool = True # HIVE_CONFIG_ENABLED and not NAVI_OVERLAY=off + source: str = "local" # local | hive | cache + version: int = 0 # document version assigned by the hive + hash: str = "" # effective hash for this instance + layers: list[str] = field(default_factory=list) + status: str = "disabled" # disabled|applied|staged|rejected|unreachable|not_targeted + errors: list[str] = field(default_factory=list) + warnings: list[str] = field(default_factory=list) + unknown_fields: list[str] = field(default_factory=list) + forbidden_fields: list[str] = field(default_factory=list) + restart_required: list[str] = field(default_factory=list) + hot_applied: list[str] = field(default_factory=list) + last_fetch: str | None = None + last_error: str | None = None + boot_guard: bool = False + + # The validated, cleaned overlay currently in force. Sections: + # settings / profiles / llm / mcp_servers. Each is the merge-patch to + # apply over the *local* base, never over a previous overlay. + merged: dict = field(default_factory=dict, repr=False) + # Base settings and profiles as loaded from disk, captured once, so a + # refresh can rebuild from the local config instead of stacking. + base_settings: object | None = field(default=None, repr=False) + + +_state = OverlayState() + + +def get_state() -> OverlayState: + return _state + + +def reset_state() -> None: + """Tests only: forget everything, including the captured base.""" + global _state + _state = OverlayState() + + +def status_dict() -> dict: + data = asdict(_state) + data.pop("merged", None) + data.pop("base_settings", None) + return data + + +def status_summary() -> dict: + """Compact form for the swarm announce payload (stored in hive.meta).""" + return { + "version": _state.version, + "hash": _state.hash, + "status": _state.status, + "restart_required": list(_state.restart_required), + "hot_applied": list(_state.hot_applied), + # Validation errors first (a rejected document), then the reason a + # fetch failed — an "unreachable" machine that does not say why is + # useless in /health and in the hive's status view. + "error": (_state.errors[0] if _state.errors else None) or _state.last_error, + } diff --git a/navi/overlay/util.py b/navi/overlay/util.py new file mode 100644 index 0000000..b0feb1c --- /dev/null +++ b/navi/overlay/util.py @@ -0,0 +1,21 @@ +"""Small shared helpers: stable JSON and the hash that identifies a config. + +The hash is computed over the *effective layer set* for one instance, not over +the whole hive document: publishing a canary layer must not make every other +machine re-ack identical content. The hive and the instance compute it the +same way, so an ack can be compared by string. +""" + +from __future__ import annotations + +import hashlib +import json + + +def canonical_json(value: object) -> str: + 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}" diff --git a/navi/overlay/validate.py b/navi/overlay/validate.py new file mode 100644 index 0000000..3cf7d03 --- /dev/null +++ b/navi/overlay/validate.py @@ -0,0 +1,266 @@ +"""Validate a hive overlay before any of it is applied. + +The rule is all-or-nothing for the parts that describe behaviour, and +ignore-with-a-warning for the parts that describe the document: + + - an unknown top-level section, an unknown setting, an unknown profile or an + unknown MCP server field is *forward compatibility* — a newer hive may + know more than this instance. It is dropped and reported, never fatal; + - a known field with the wrong type, an invalid route, or an MCP config that + does not parse is a broken document. Applying the other half of it would + leave the fleet in a state nobody wrote, so the whole document is + rejected and the previous one (from cache) stays in force. + +Missing fields are the normal case, not an error: they fall back to the local +config by definition of the overlay. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +from navi.overlay.merge import merge_patch +from navi.overlay.schema import DENIED_SETTINGS, KNOWN_SECTIONS + +_LLM_KEYS = {"targets", "routes", "probe"} +_TOOL_SCOPE_KEYS = {"agent", "subagent"} +_TOOL_SCOPE_INNER = {"native", "mcp"} + + +@dataclass +class ValidationResult: + merged: dict = field(default_factory=dict) + errors: list[str] = field(default_factory=list) + warnings: list[str] = field(default_factory=list) + unknown_fields: list[str] = field(default_factory=list) + forbidden_fields: list[str] = field(default_factory=list) + + @property + def ok(self) -> bool: + return not self.errors + + +def _is_int(value: object) -> bool: + return isinstance(value, int) and not isinstance(value, bool) + + +def validate_envelope(layers: object) -> tuple[list[dict], list[str]]: + """Check the transport shape; return the usable layers and warnings.""" + warnings: list[str] = [] + if not isinstance(layers, list): + raise ValueError("'layers' must be a list") + usable: list[dict] = [] + for index, layer in enumerate(layers): + if not isinstance(layer, dict): + raise ValueError(f"layer #{index} is not an object") + name = layer.get("name", f"layer{index}") + config = layer.get("config") + if not isinstance(name, str) or not name: + raise ValueError(f"layer #{index} needs a string 'name'") + if not isinstance(config, dict): + raise ValueError(f"layer '{name}' needs an object 'config'") + usable.append({"name": name, "config": config}) + return usable, warnings + + +def validate_layers(base_settings, layers: list[dict]) -> ValidationResult: + """Merge the layers and validate the result against the local base.""" + result = ValidationResult() + merged: dict = {} + for layer in layers: + merged = merge_patch(merged, layer["config"]) # type: ignore[assignment] + + unknown_sections = sorted(set(merged) - KNOWN_SECTIONS) + for name in unknown_sections: + result.warnings.append(f"unknown section '{name}' ignored") + merged.pop(name, None) + + if "settings" in merged: + merged["settings"] = _clean_settings(base_settings, merged["settings"], result) + if "profiles" in merged: + merged["profiles"] = _clean_profiles(merged["profiles"], result) + if "llm" in merged: + merged["llm"] = _clean_llm(base_settings, merged["llm"], result) + if "mcp_servers" in merged: + merged["mcp_servers"] = _clean_mcp(merged["mcp_servers"], result) + + result.merged = {k: v for k, v in merged.items() if v} + return result + + +# ── settings ──────────────────────────────────────────────────────────────── + + +def _clean_settings(base_settings, patch: object, result: ValidationResult) -> dict: + if not isinstance(patch, dict): + result.errors.append("settings must be an object") + return {} + + known = set(type(base_settings).model_fields) + cleaned: dict = {} + for name, value in patch.items(): + if name not in known: + result.unknown_fields.append(f"settings.{name}") + continue + if name in DENIED_SETTINGS: + result.forbidden_fields.append(f"settings.{name}") + continue + cleaned[name] = value + + try: + type(base_settings)(**{**base_settings.model_dump(), **cleaned}) + except Exception as exc: # validation error text names the bad field + result.errors.append(f"settings: {exc}") + return {} + return cleaned + + +# ── profiles ──────────────────────────────────────────────────────────────── + + +def _clean_profiles(patch: object, result: ValidationResult) -> dict: + if not isinstance(patch, dict): + result.errors.append("profiles must be an object") + return {} + + from navi.profiles.base import AgentProfile + from navi.profiles import ALL_PROFILES + + base_by_id = {p.id: p for p in ALL_PROFILES} + known = set(AgentProfile.model_fields) + cleaned: dict = {} + for profile_id, profile_patch in patch.items(): + if profile_id not in base_by_id: + result.warnings.append(f"unknown profile '{profile_id}' ignored") + continue + if not isinstance(profile_patch, dict): + result.errors.append(f"profiles.{profile_id} must be an object") + continue + + filtered: dict = {} + for name, value in profile_patch.items(): + if name not in known: + result.unknown_fields.append(f"profiles.{profile_id}.{name}") + continue + if name == "tools": + value = _clean_tools(profile_id, value, result) + if value is None: + continue + filtered[name] = value + if "id" in filtered and filtered["id"] != profile_id: + result.errors.append( + f"profiles.{profile_id}: 'id' cannot be changed ({filtered['id']!r})" + ) + continue + + try: + AgentProfile.model_validate( + merge_patch(base_by_id[profile_id].model_dump(), filtered) + ) + except Exception as exc: + result.errors.append(f"profiles.{profile_id}: {exc}") + continue + if filtered: # a profile patch that only held unknown fields says nothing + cleaned[profile_id] = filtered + return cleaned + + +def _clean_tools(profile_id: str, patch: object, result: ValidationResult) -> dict | None: + if not isinstance(patch, dict): + result.errors.append(f"profiles.{profile_id}.tools must be an object") + return None + out: dict = {} + for scope, scope_patch in patch.items(): + if scope not in _TOOL_SCOPE_KEYS: + result.unknown_fields.append(f"profiles.{profile_id}.tools.{scope}") + continue + if not isinstance(scope_patch, dict): + result.errors.append(f"profiles.{profile_id}.tools.{scope} must be an object") + continue + inner: dict = {} + for key, value in scope_patch.items(): + if key not in _TOOL_SCOPE_INNER: + result.unknown_fields.append( + f"profiles.{profile_id}.tools.{scope}.{key}" + ) + continue + inner[key] = value + out[scope] = inner + return out + + +# ── llm ───────────────────────────────────────────────────────────────────── + + +def _clean_llm(base_settings, patch: object, result: ValidationResult) -> dict: + if not isinstance(patch, dict): + result.errors.append("llm must be an object") + return {} + + filtered: dict = {} + for key, value in patch.items(): + if key not in _LLM_KEYS: + result.unknown_fields.append(f"llm.{key}") + continue + filtered[key] = value + + from navi.llm.config import LLMConfigError, load_llm_raw, parse_llm_config + + try: + local_raw = load_llm_raw(base_settings.llm_config_file) or {} + except Exception as exc: # a broken local file must not block a fix from above + result.warnings.append(f"local llm.json is invalid ({exc}); merged without it") + local_raw = {} + + merged = merge_patch(local_raw, filtered) + if not merged: + return {} + try: + # Secrets are resolved when the backend is actually built; doing it + # here would import the settings singleton while it is being created. + parse_llm_config(merged, source="hive-overlay", resolve_secrets=False) + except LLMConfigError as exc: + result.errors.append(f"llm: {exc}") + return {} + return filtered + + +# ── MCP servers ───────────────────────────────────────────────────────────── + + +def _clean_mcp(patch: object, result: ValidationResult) -> dict: + """Validate per-server patches against the local server definitions. + + A patch may define a server the machine does not have (the hive is the + source of truth for the fleet) or override fields of one it has. It + cannot *remove* one: ``null`` already means "drop this key" at the + merge-patch level, so there is no value left here to act on. Removing a + server stays a local-file operation. + """ + if not isinstance(patch, dict): + result.errors.append("mcp_servers must be an object") + return {} + + from navi.mcp.config import McpServerConfig, load_mcp_servers + + local = {name: cfg.model_dump() for name, cfg in load_mcp_servers().items()} + known = set(McpServerConfig.model_fields) + cleaned: dict = {} + + for server, server_patch in patch.items(): + if not isinstance(server_patch, dict): + result.errors.append(f"mcp_servers.{server} must be an object") + continue + filtered: dict = {} + for key, value in server_patch.items(): + if key not in known: + result.unknown_fields.append(f"mcp_servers.{server}.{key}") + continue + filtered[key] = value + try: + McpServerConfig.model_validate(merge_patch(local.get(server, {}), filtered)) + except Exception as exc: + result.errors.append(f"mcp_servers.{server}: {exc}") + continue + cleaned[server] = filtered + return cleaned diff --git a/navi/settings_model.py b/navi/settings_model.py new file mode 100644 index 0000000..7f29b20 --- /dev/null +++ b/navi/settings_model.py @@ -0,0 +1,315 @@ +from pathlib import Path + +from pydantic import model_validator +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class Settings(BaseSettings): + model_config = SettingsConfigDict( + env_file=".env", env_file_encoding="utf-8", extra="ignore", frozen=True + ) + + ollama_host: str = "http://localhost:11434" + ollama_api_key: str = "" + ollama_default_model: str = "gemma4:31b-cloud" + ollama_num_ctx: int = 65536 + ollama_think: bool = True + ollama_request_timeout: int = 30 + + # Embedding model for memory vector search (Ollama API) + # When embedding_ollama_host is empty, falls back to ollama_host. + embedding_ollama_host: str = "" + embedding_ollama_api_key: str = "" + embedding_model: str = "nomic-embed-text:latest" + # Route in llm.json used for embeddings (same failover as chat). Ignored + # when embedding_ollama_host pins a dedicated host. + embedding_route: str = "embedding" + embedding_dimensions: int = 768 + # Embedding models reject input past their context window with a 400, which + # costs the whole embedding — recall then falls back to ILIKE. Clip instead: + # a reaction-session prompt (a full event envelope inlined) failed 41 times. + # 0 disables the cap. + embedding_max_chars: int = 6000 + + openai_api_key: str = "" + openai_model: str = "gpt-4" + openai_base_url: str | None = None + anthropic_api_key: str = "" + + # Web search is not configured here: it is the navi-web MCP server's own + # settings (NAVI_WEB_*), mapped from .env by mcp_servers.d/navi-web.json. + + # Filesystem tool: comma-separated allowed root paths + fs_allowed_paths: str = "*" + + # Terminal tool: "*" = allow all commands (recommended for local use) + # or comma-separated list of allowed executables, e.g. "ls,cat,git" + terminal_allowed_commands: str = "*" + # Terminal allowlist for non-admin users (multi-user mode). + # Admin bypasses this restriction. + # Interpreters stay (python3/node) so scripting work is possible; package managers and + # compilers do not — the user profiles forbid installing packages or changing + # machine-wide configuration, and a toolchain is the easiest way to run foreign code. + terminal_user_allowed_commands: str = ( + "ls,cat,git,grep,egrep,fgrep,head,tail,wc,find,locate," + "ps,df,du,free,uptime,whoami,id,mkdir,touch,cp,mv,rm,chmod," + "echo,printf,sort,uniq,awk,sed,tr,cut,date,uname,hostname,which,whereis," + "python,python3,node,env,export,source" + ) + + # SSH tool: path to JSON file with named connections + ssh_hosts_file: str = "ssh_hosts.json" + + # LLM routing config: targets (hosts) and routes (ordered model@target + # chains). Relative paths resolve against CWD, then the repo root. Absent + # file = single target from OLLAMA_HOST, as in a bare dev checkout. + llm_config_file: str = "llm.json" + + # Database — PostgreSQL is required (SQLite support removed). + database_url: str = "" + + log_level: str = "INFO" + + # Directory for user-defined tools (auto-discovered at startup) + tools_dir: str = "tools" + + # Directory for user-defined context providers (auto-discovered at startup) + context_providers_dir: str = "context_providers" + + # Session file uploads + session_files_dir: str = "session_files" + session_files_max_size_mb: int = 200 + session_messages_window: int = 1000 # max hot messages per session; older -> archive + ws_replay_buffer_size: int = 500 # max events retained for WS reconnect replay + share_file_max_size_mb: int = 1024 + + # Background tasks (see docs/tasks.md). Tools in `backgroundable_tools` + # accept background=true: the call is detached into a TaskManager job and + # the tool call returns a task_id immediately. Tasks are in-memory — a + # restart loses them (same as terminal processes). + tasks_max_per_session: int = 5 # concurrent running tasks per session + tasks_max_global: int = 20 # concurrent running tasks server-wide + tasks_max_spawn: int = 2 # concurrent background spawn_agent per session + tasks_ttl_sec: int = 3600 # finished-task retention before reap + tasks_event_buffer_size: int = 50 # per-task event ring (spawn tasks: 200) + tasks_rate_limit: int = 10 # max task spawns per 5 min per session + task_notes_max_pending: int = 20 # pending completion notes per session + task_notes_per_turn: int = 5 # max notes coalesced into one turn injection + message_queue_max: int = 5 # user messages queued while a run is active + backgroundable_tools: str = "terminal,ssh_exec,peer,spawn_agent,code_exec" + parallel_tool_calls: bool = False # execute a multi-tool-call batch concurrently + + # Public base URL used by share_file tool to build download links. + # Change if the server is behind a reverse proxy or runs on a different port. + public_url: str = "http://localhost:8099" + + # --- Export: bulk history snapshot for an external consumer (n-gram tables) + # Every navi_export_interval_hours one gzip-JSONL file of message text is built + # and age-encrypted to navi_export_age_recipient. The private half of that key + # never reaches this server, so Navi cannot decrypt its own export. + # navi_export_token is the bearer token the consumer presents; only its SHA-256 + # digest is compared. Both fields must be set for the builder to run and + # the endpoints to answer at all. + navi_export_age_recipient: str = "" + navi_export_token: str = "" + navi_export_interval_hours: float = 4.0 + # Which roles make the corpus. Tool results and system prompts are opt-in: + # tool output is scraped web text and command logs (52 MB of prod's 74 MB), + # and system messages are near-duplicate boilerplate — both would drown the + # actual dialogue in an n-gram table. + navi_export_roles: str = "user,assistant" + # Rolling window, whole sessions, newest first, until this much raw text. + # A cap that never binds today (the whole corpus is ~13 MB); it exists so + # the snapshot stays bounded as history grows. + navi_export_max_bytes: int = 150 * 1024 * 1024 + navi_export_dir: str = "export_files" + + # Gmail IMAP/SMTP (App Password auth) + gmail_address: str = "" + gmail_app_password: str = "" + + # gnexus-auth OAuth integration + gnauth_base_url: str = "http://gnexus-auth.local" + gnauth_client_id: str = "" + gnauth_client_secret: str = "" + gnauth_redirect_uri: str = "http://localhost:8099/auth/callback" + gnauth_admin_role_slug: str = "navi_admin" + gnauth_user_role_slug: str = "navi_user" + gnauth_profile_path: str = "/account/profile" # appended to gnauth_base_url for profile links + + # HMAC secret for gnexus-auth webhooks (from the gnexus-auth admin panel). + # When empty and navi_auth_enabled=true, the webhook endpoint refuses to + # process unsigned events (503). When auth is disabled, unsigned webhooks + # are accepted with a warning. + gnauth_webhook_secret: str = "" + + # Master auth switch. When false, Navi skips OAuth/API-token checks and + # treats every request as the anonymous admin user. Use only for trusted + # single-user/local deployments. + navi_auth_enabled: bool = True + + # Default profile used when a client creates a session without specifying + # a profile_id. Empty string means "no default"; the client must supply one. + navi_default_profile_id: str = "" + + # Web UI switch. When false, the static mounts ("/", /assets, /images, + # /content-viewers, /content), the debug/admin panels and the navi_ui MCP + # server are not registered — pure API/WS server for terminal clients. + # The API itself is unaffected; re-enable later by editing .env only. + navi_webclient_enabled: bool = True + + # Bind address for the navi-server launcher (navi/server.py + systemd). + # 127.0.0.1 keeps the server local-only; terminals on other machines + # connect through SSH tunneling or a reverse proxy. + # 8099/8098 instead of the common 8000/8001: navi runs on shared servers + # where those ports are usually taken. + navi_host: str = "127.0.0.1" + navi_port: int = 8099 + + # Internal navi_ui MCP server — lets the agent push structured UI components + # (card_grid, form) to the webclient via the render_component tool. + navi_ui_mcp_enabled: bool = True + 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 + + # Central config overlay served by the hive (see navi/overlay/). + # These three are bootstrap: they are read from the local environment + # before the overlay is fetched, and the overlay itself can never change + # them (it would have to arrive through the channel it configures). + hive_config_enabled: bool = True + hive_config_cache_file: str = ".hive-config.cache.json" + hive_config_timeout_sec: float = 2.0 + + # Peer-to-peer swarm channel (stage 2). Incoming /peer/ask requests are + # answered by a one-shot agent run under this profile — pick a profile + # with broad but non-destructive tools (default: server_admin). + peer_ask_profile: str = "server_admin" + # Hard ceiling for a single peer ask (the LLM run inside /peer/ask). + peer_ask_timeout_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" + navi_auth_cookie_secure: bool = False + navi_auth_cookie_samesite: str = "lax" + navi_auth_cookie_max_age_days: int = 30 + + # Comma-separated list of origins allowed by CORS when navi_auth_enabled=true + # (cookie-credentialed requests need an explicit origin list — "*" is not + # accepted by browsers together with credentials). Required in auth mode: + # the app refuses to start with an empty list. Ignored when auth is disabled. + navi_allowed_origins: str = "" + + # Web push (PWA). Empty VAPID keys = push fully disabled: no endpoints + # registered, no notifications sent, the client panel shows the toggle + # as unavailable. + navi_push_vapid_public_key: str = "" + navi_push_vapid_private_key: str = "" + navi_push_vapid_subject: str = "mailto:admin@navi.local" + # Minimum seconds between two pushes for the same session (recall + agent + # runs can overlap on one session; also dedups rapid-fire turns). + navi_push_cooldown_sec: int = 30 + + # Synapse ingest — navi as an event SOURCE (outbound reaction notifications). + # Empty api key = navi cannot emit events to Synapse: the settings UI + # disables the synapse push-target options and the notify tool silently + # skips that leg of delivery. + synapse_source_url: str = "" + synapse_source_api_key: str = "" + synapse_source_name: str = "navi" + + # LLM call timeouts + # complete() is non-streaming (compression) — blocked until full response + llm_complete_timeout: int = 120 + # Non-streaming planning calls (the `plan` tool / subagent pipeline). Wider + # than llm_complete_timeout: cloud models prefill big planning prompts for + # up to a couple of minutes, and a timeout here silently kills the plan. + planning_llm_timeout_sec: int = 240 + # stream_complete(): how long to wait for the FIRST token (prefill phase) + # Large contexts can take 60-90s to prefill; 90s matches user expectation + llm_stream_first_chunk_timeout: int = 90 + # stream_complete(): max gap between any two subsequent tokens + llm_stream_chunk_timeout: int = 60 + + # Sub-agent guards. The thinking bound catches a model that degenerates + # into an endless internal loop; 12k chars was reachable in a healthy + # iteration of a cloud reasoning model, so healthy runs were killed + # mid-thought and reported as stalled. + subagent_thinking_stall_seconds: float = 60.0 + subagent_thinking_stall_chars: int = 48_000 + # Whole sub-agent run. 300s routinely cut a long ops task in half; the + # parent can still stop the run, and a background spawn reports separately. + subagent_timeout_seconds: float = 600.0 + # complete()/embed(): attempts per (server, model) on a connection error + # (timeout, refused, DNS) before the fallback gives up. A cloud backend times + # out on its own schedule, and one ReadTimeout from the only configured + # server used to abort the call outright — on prod 2026-10-08 that ended 27 + # calls in a single day, 23 of them memory summarisation. Only connection + # errors are retried; a 4xx/5xx or a missing model is a real answer, not a + # hiccup. + llm_complete_retries: int = 2 + llm_retry_backoff_sec: float = 2.0 + + # Context compression + context_compression_enabled: bool = True + context_compression_threshold: float = 0.90 # trigger at 90% of ollama_num_ctx + context_compression_target: float = 0.65 # hysteresis target — after compression the kept region should fit ~65% of the window (trigger 90% → target 65% leaves real headroom instead of skimming just below the trigger and re-firing a few messages later) + context_keep_recent: int = 4 # conversational turns to keep verbatim + context_summary_temperature: float = 0.3 + context_summary_max_tokens: int = 6000 # max output tokens for the summary LLM call + output_reserve_tokens: int = 2048 # headroom reserved for model response in context checks + # Per-message token budget for the LLM context view. A single tool/assistant + # message whose estimated size exceeds this is head/tail-truncated in the + # built context (never in stored history) so one huge tool result cannot + # alone blow the window. 0 = auto (ollama_num_ctx // 6). + context_message_token_budget: int = 0 + + # Global personality prompt prepended to every agent's system prompt. + # Multi-line values don't survive .env parsing reliably, so prefer + # navi_persona_file (path to a plain .txt file) over inline navi_persona. + navi_persona: str = "" + navi_persona_file: str = "" + + @model_validator(mode="before") + def _load_persona_from_file(cls, values: dict) -> dict: + if not values.get("navi_persona") and values.get("navi_persona_file"): + try: + values["navi_persona"] = Path(values["navi_persona_file"]).read_text( + encoding="utf-8" + ).strip() + except Exception: + pass + return values + + @property + def fs_allowed_paths_list(self) -> list[str]: + return [p.strip() for p in self.fs_allowed_paths.split(",") if p.strip()] + + @property + def navi_export_roles_list(self) -> list[str]: + return [r.strip() for r in self.navi_export_roles.split(",") if r.strip()] + + @property + def navi_export_configured(self) -> bool: + """Both a recipient key and a token are required — no half-configured export.""" + return bool(self.navi_export_age_recipient.strip() and self.navi_export_token.strip()) + + @property + def terminal_allowed_commands_list(self) -> list[str]: + return [c.strip() for c in self.terminal_allowed_commands.split(",") if c.strip()] + + @property + def terminal_user_allowed_commands_list(self) -> list[str]: + return [c.strip() for c in self.terminal_user_allowed_commands.split(",") if c.strip()] + + @property + def navi_allowed_origins_list(self) -> list[str]: + return [o.strip() for o in self.navi_allowed_origins.split(",") if o.strip()] diff --git a/navi/swarm.py b/navi/swarm.py index 9751728..2b4eb62 100644 --- a/navi/swarm.py +++ b/navi/swarm.py @@ -96,6 +96,7 @@ port: int, interval: int = 120, client: httpx.AsyncClient | None = None, + on_config_hint=None, ) -> None: self._hive_url = hive_url.rstrip("/") self._key = key @@ -103,6 +104,10 @@ self._port = port self._interval = interval self._client = client # injectable for tests; created lazily otherwise + # Called with the hive's ``config`` hint when the announce reply says + # this instance is behind. The announce loop is the reliable path: it + # runs even when a push notification was missed while offline. + self._on_config_hint = on_config_hint # state self._reachable: bool | None = None # None = never announced yet self._last_success: str | None = None @@ -112,12 +117,23 @@ @property def payload(self) -> dict: + meta = _machine_meta() + try: + from navi.overlay import status_summary + + # Which config this instance runs, so the hive can see which + # machines are behind without waiting for an ack to arrive. + meta["config"] = status_summary() + except Exception: # reporting must never break the announce + log.debug("hive.config.status_summary_failed", exc_info=True) + if self._identity.tags: + meta["tags"] = list(self._identity.tags) return { "name": self._identity.name, "instance_id": self._identity.instance_id, "version": __version__, "port": self._port, - "meta": _machine_meta(), + "meta": meta, } async def announce_once(self) -> bool: @@ -154,6 +170,21 @@ self._last_error = None self._down_since = None self._consecutive_failures = 0 + + # The reply is how a machine that was offline learns it is behind: the + # hive compares the config it serves with the one this instance just + # reported and asks for a pull. + try: + data = resp.json() + except ValueError: + data = None + hint = data.get("config") if isinstance(data, dict) else None + if isinstance(hint, dict) and hint.get("fetch") and self._on_config_hint: + log.info("hive.config.announce_hint", version=hint.get("version")) + try: + self._on_config_hint(hint) + except Exception: + log.warning("hive.config.hint_failed", exc_info=True) return True async def run(self) -> None: @@ -194,6 +225,15 @@ return _announcer +def _schedule_config_refresh(hint: dict) -> None: + """Announce says we are behind → pull the config, off the loop.""" + from navi.overlay import get_state, refresh + + if not get_state().enabled: + return + asyncio.create_task(asyncio.to_thread(refresh, "announce")) + + def build_announcer_from_settings() -> HiveAnnouncer | None: """Assemble the announcer from config; None (with a warning) if misconfigured.""" from navi.config import settings @@ -213,4 +253,5 @@ identity=ensure_identity(settings.instance_file), port=settings.navi_port, interval=settings.announce_interval_sec, + on_config_hint=_schedule_config_refresh, ) \ No newline at end of file diff --git a/tests/conftest.py b/tests/conftest.py index 5b96459..979df1b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -11,6 +11,12 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +# The config overlay talks to the hive while ``navi.config`` is imported, and +# a developer machine may well have HIVE_URL set. Tests get the local config, +# always — before any navi module is imported (several import ``settings`` at +# module level and would capture an overlaid object). +os.environ["NAVI_OVERLAY"] = "off" + @pytest.fixture(autouse=True) def _reset_settings(monkeypatch: pytest.MonkeyPatch) -> Generator[None, None, None]: diff --git a/tests/unit/overlay/test_apply.py b/tests/unit/overlay/test_apply.py new file mode 100644 index 0000000..9c4094d --- /dev/null +++ b/tests/unit/overlay/test_apply.py @@ -0,0 +1,267 @@ +"""Applying a validated overlay: settings, profiles, llm, mcp.""" + +import pytest + +import navi.config as config_mod +from navi.overlay import apply as apply_mod +from navi.overlay.apply import ( + apply_profiles_list, + apply_profiles_registry, + apply_settings_live, + build_settings, + overlay_llm_raw, + overlay_mcp_configs, + refresh, +) +from navi.overlay.fetch import Fetched +from navi.overlay.state import get_state, reset_state, status_summary +from navi.settings_model import Settings + + +@pytest.fixture(autouse=True) +def _fresh_state(): + reset_state() + yield + reset_state() + + +@pytest.fixture() +def base(tmp_path) -> Settings: + return Settings( + _env_file=None, + hive_config_cache_file=str(tmp_path / "cache.json"), + ) + + +@pytest.fixture() +def state(base): + state = get_state() + state.base_settings = base + state.enabled = True + return state + + +# ── settings ──────────────────────────────────────────────────────────────── + + +def test_build_settings_overlays_only_named_fields(base): + overlaid = build_settings(base, {"llm_complete_timeout": 99}) + assert overlaid.llm_complete_timeout == 99 + assert overlaid.navi_port == base.navi_port + assert overlaid is not base + + +def test_build_settings_without_patch_returns_the_same_object(base): + assert build_settings(base, {}) is base + + +def test_hot_setting_is_written_into_the_live_object(state): + hot, restart = apply_settings_live({"gnauth_client_secret": "rotated"}) + assert hot == ["gnauth_client_secret"] + assert restart == [] + assert config_mod.settings.gnauth_client_secret == "rotated" + + +def test_non_hot_setting_requires_a_restart(state): + before = config_mod.settings.openai_api_key + hot, restart = apply_settings_live({"openai_api_key": "sk-new"}) + assert hot == [] + assert restart == ["openai_api_key"] + # Not applied behind the operator's back. + assert config_mod.settings.openai_api_key == before + + +def test_removing_a_field_hands_it_back_to_the_local_value(state, base): + state.merged = {"settings": {"llm_complete_timeout": 99}} + apply_settings_live({"llm_complete_timeout": 99}) + assert config_mod.settings.llm_complete_timeout == 99 + + state.merged = {"settings": {"llm_complete_timeout": 99}} + hot, _ = apply_settings_live({}) + assert hot == ["llm_complete_timeout"] + assert config_mod.settings.llm_complete_timeout == base.llm_complete_timeout + + +def test_unchanged_hot_field_is_not_reported(state): + apply_settings_live({"gnauth_client_secret": "same"}) + hot, _ = apply_settings_live({"gnauth_client_secret": "same"}) + assert hot == [] + + +# ── profiles ──────────────────────────────────────────────────────────────── + + +def test_apply_profiles_list_overlays_the_named_profile(state): + from navi.profiles import ALL_PROFILES + + state.merged = {"profiles": {"assistant": {"temperature": 0.11}}} + profiles = apply_profiles_list(ALL_PROFILES) + by_id = {p.id: p for p in profiles} + assert by_id["assistant"].temperature == 0.11 + others = [p for p in profiles if p.id != "assistant"] + assert others # the untouched profiles are still all there + + +def test_apply_profiles_list_without_patch_returns_input(state): + from navi.profiles import ALL_PROFILES + + state.merged = {} + assert apply_profiles_list(ALL_PROFILES) is ALL_PROFILES + + +def test_registry_rebuild_is_idempotent_and_reversible(state): + from navi.core.registry import ProfileRegistry + from navi.profiles import ALL_PROFILES + + registry = ProfileRegistry() + for profile in ALL_PROFILES: + registry.register(profile) + original = registry.get("assistant").temperature + + state.merged = {"profiles": {"assistant": {"temperature": 0.05}}} + apply_profiles_registry(registry) + assert registry.get("assistant").temperature == 0.05 + + state.merged = {} + apply_profiles_registry(registry) + assert registry.get("assistant").temperature == original + + +def test_registry_leaves_unknown_runtime_profiles_alone(state): + from navi.core.registry import ProfileRegistry + from navi.profiles.base import AgentProfile + + registry = ProfileRegistry() + custom = AgentProfile(id="made-at-runtime", name="X", description="", system_prompt="") + registry.register(custom) + apply_profiles_registry(registry) + assert registry.get("made-at-runtime") is custom + + +# ── llm / mcp ─────────────────────────────────────────────────────────────── + + +def test_overlay_llm_raw_merges_and_keeps_local_values(state): + state.merged = {"llm": {"targets": {"home": {"host": "http://new"}}}} + raw = {"targets": {"home": {"host": "http://old"}, "cloud": {"host": "https://c"}}} + merged = overlay_llm_raw(raw) + assert merged["targets"]["home"]["host"] == "http://new" + assert merged["targets"]["cloud"]["host"] == "https://c" + + +def test_overlay_llm_raw_without_patch_is_a_passthrough(state): + state.merged = {} + raw = {"targets": {}} + assert overlay_llm_raw(raw) is raw + assert overlay_llm_raw(None) is None + + +def test_overlay_mcp_configs_merges_and_deletes(state): + from navi.mcp.config import McpServerConfig + + state.merged = { + "mcp_servers": {"srv": {"command": "new"}, "gone": None} + } + configs = { + "srv": McpServerConfig(transport="stdio", command="old"), + "gone": McpServerConfig(transport="stdio", command="x"), + "kept": McpServerConfig(transport="stdio", command="y"), + } + merged = overlay_mcp_configs(configs) + assert merged["srv"].command == "new" + assert "gone" not in merged + assert merged["kept"].command == "y" + + +# ── refresh ───────────────────────────────────────────────────────────────── + + +def test_refresh_applies_and_records(state, base, monkeypatch): + fetched = Fetched( + version=7, + layers=[{"name": "global", "config": {"settings": {"llm_complete_timeout": 33}}}], + hash="sha256:x", + ) + monkeypatch.setattr(apply_mod, "fetch_layers", lambda b: fetched) + + result = refresh("test") + assert result["status"] == "applied" + assert config_mod.settings.llm_complete_timeout == 33 + assert state.version == 7 + assert state.hash == "sha256:x" + assert state.layers == ["global"] + assert state.merged["settings"] == {"llm_complete_timeout": 33} + + +def test_refresh_rejects_and_keeps_the_previous_config(state, monkeypatch): + monkeypatch.setattr( + apply_mod, + "fetch_layers", + lambda b: Fetched( + version=1, + layers=[{"name": "g", "config": {"settings": {"llm_complete_timeout": 33}}}], + hash="sha256:good", + ), + ) + refresh("first") + good = config_mod.settings.llm_complete_timeout + + monkeypatch.setattr( + apply_mod, + "fetch_layers", + lambda b: Fetched( + version=2, + layers=[{"name": "g", "config": {"settings": {"llm_complete_timeout": "soon"}}}], + hash="sha256:bad", + ), + ) + result = refresh("second") + assert result["status"] == "rejected" + assert config_mod.settings.llm_complete_timeout == good + assert state.version == 1 + assert state.errors + + +def test_refresh_is_a_noop_when_the_channel_is_disabled(state): + state.enabled = False + assert refresh("test")["status"] == "disabled" + + +def test_refresh_survives_an_unreachable_hive(state, monkeypatch): + def boom(base_settings): + raise RuntimeError("no route to host") + + monkeypatch.setattr(apply_mod, "fetch_layers", boom) + result = refresh("test") + assert result["status"] == "unreachable" + assert state.last_error and "no route" in state.last_error + # The reason reaches /health and the hive, not only the state object. + assert "no route" in (status_summary()["error"] or "") + + # A later success clears it, so /health stops reporting an old outage. + monkeypatch.setattr( + apply_mod, + "fetch_layers", + lambda b: Fetched( + version=1, + layers=[{"name": "g", "config": {"settings": {"llm_complete_timeout": 21}}}], + hash="sha256:ok", + ), + ) + refresh("retry") + assert state.last_error is None + + +def test_refresh_staged_when_a_restart_is_needed(state, monkeypatch): + monkeypatch.setattr( + apply_mod, + "fetch_layers", + lambda b: Fetched( + version=1, + layers=[{"name": "g", "config": {"settings": {"openai_api_key": "sk-new"}}}], + hash="sha256:x", + ), + ) + result = refresh("test") + assert result["status"] == "staged" + assert "openai_api_key" in state.restart_required diff --git a/tests/unit/overlay/test_fetch.py b/tests/unit/overlay/test_fetch.py new file mode 100644 index 0000000..cc615c9 --- /dev/null +++ b/tests/unit/overlay/test_fetch.py @@ -0,0 +1,238 @@ +"""Fetching, caching and the boot-loop guard.""" + +from pathlib import Path + +import pytest + +from navi.overlay import fetch as fetch_mod +from navi.overlay.fetch import ( + CachedState, + Fetched, + confirm_startup, + load_cached, + resolve_startup, + save_cached, + store_applied, +) +from navi.overlay.util import content_hash +from navi.settings_model import Settings + + +@pytest.fixture() +def base(tmp_path) -> Settings: + return Settings( + _env_file=None, + hive_url="http://hive.test:8087", + hive_config_cache_file=str(tmp_path / "cache.json"), + hive_config_timeout_sec=0.1, + ) + + +def _doc(**settings) -> list[dict]: + return [{"name": "global", "config": {"settings": settings}}] + + +def _path(base) -> Path: + return Path(base.hive_config_cache_file) + + +def _patch_fetch(monkeypatch, result): + def fake(base_settings): + if isinstance(result, Exception): + raise result + return result + + monkeypatch.setattr(fetch_mod, "fetch_layers", fake) + + +# ── cache file ────────────────────────────────────────────────────────────── + + +def test_cache_roundtrip(base): + save_cached( + _path(base), + CachedState( + good={"version": 1, "hash": "sha256:x", "layers": _doc()}, + pending=None, + blocked_hash="", + ), + ) + loaded = load_cached(_path(base)) + assert loaded.good["version"] == 1 + assert loaded.pending is None + + +def test_cache_is_private(base): + save_cached(_path(base), CachedState(good={"version": 1, "hash": "", "layers": []})) + assert (_path(base).stat().st_mode & 0o777) == 0o600 + + +def test_broken_cache_is_treated_as_empty(base): + _path(base).write_text("{not json") + loaded = load_cached(_path(base)) + assert loaded.good is None + + +# ── startup resolution ────────────────────────────────────────────────────── + + +def test_no_hive_and_no_cache_runs_local(base, monkeypatch): + _patch_fetch(monkeypatch, RuntimeError("connection refused")) + result = resolve_startup(base) + assert result.layers == [] + assert result.source == "local" + assert "connection refused" in (result.error or "") + + +def test_hive_url_without_a_key_is_reported_as_unreachable(tmp_path): + # No monkeypatching: the real fetch runs and stops at the missing key. + settings = Settings( + _env_file=None, + hive_url="http://hive.test:8087", + swarm_key_file=str(tmp_path / "absent-key"), + hive_config_cache_file=str(tmp_path / "cache.json"), + hive_config_timeout_sec=0.1, + ) + result = resolve_startup(settings) + assert result.layers == [] + assert result.source == "local" + assert "swarm key" in (result.error or "") + + +def test_fetched_document_is_used_and_marked_pending(base, monkeypatch): + _patch_fetch( + monkeypatch, + Fetched(version=3, layers=_doc(llm_complete_timeout=30), hash="sha256:a"), + ) + result = resolve_startup(base) + assert result.source == "hive" + assert result.merged["settings"] == {"llm_complete_timeout": 30} + + cached = load_cached(_path(base)) + assert cached.pending is not None + assert cached.pending["attempts"] == 1 + assert cached.good is None + + +def test_confirm_startup_promotes_pending(base, monkeypatch): + _patch_fetch( + monkeypatch, + Fetched(version=3, layers=_doc(llm_complete_timeout=30), hash="sha256:a"), + ) + resolve_startup(base) + confirm_startup(base) + cached = load_cached(_path(base)) + assert cached.good is not None + assert cached.pending is None + + +def test_same_config_retried_once_then_blocked(base, monkeypatch): + """A config that was applied but never confirmed is retried once.""" + _patch_fetch( + monkeypatch, + Fetched(version=3, layers=_doc(llm_complete_timeout=30), hash="sha256:a"), + ) + + resolve_startup(base) # attempt 1 + resolve_startup(base) # attempt 2 + result = resolve_startup(base) # guard trips, nothing is applied + + assert result.layers == [] + assert result.boot_guard is True + cached = load_cached(_path(base)) + assert cached.blocked_hash == "sha256:a" + assert cached.pending is None + + +def test_a_new_document_clears_the_guard(base, monkeypatch): + _patch_fetch( + monkeypatch, + Fetched(version=3, layers=_doc(llm_complete_timeout=30), hash="sha256:a"), + ) + for _ in range(3): + resolve_startup(base) + + _patch_fetch( + monkeypatch, + Fetched(version=4, layers=_doc(llm_complete_timeout=45), hash="sha256:b"), + ) + result = resolve_startup(base) + assert result.source == "hive" + assert result.merged["settings"] == {"llm_complete_timeout": 45} + + +def test_fetch_failure_falls_back_to_the_last_good_cache(base, monkeypatch): + good = _doc(llm_complete_timeout=30) + save_cached( + _path(base), + CachedState(good={"version": 2, "hash": "sha256:g", "layers": good}), + ) + _patch_fetch(monkeypatch, RuntimeError("hive down")) + + result = resolve_startup(base) + assert result.source == "cache" + assert result.merged["settings"] == {"llm_complete_timeout": 30} + assert result.error and "hive down" in result.error + + +def test_invalid_fetched_document_falls_back_to_good_cache(base, monkeypatch): + good = _doc(llm_complete_timeout=30) + save_cached( + _path(base), + CachedState(good={"version": 2, "hash": "sha256:g", "layers": good}), + ) + _patch_fetch( + monkeypatch, + Fetched(version=3, layers=_doc(llm_complete_timeout="soon"), hash="sha256:b"), + ) + + result = resolve_startup(base) + assert result.source == "cache" + assert result.merged["settings"] == {"llm_complete_timeout": 30} + + +def test_invalid_document_with_no_cache_runs_local(base, monkeypatch): + _patch_fetch( + monkeypatch, + Fetched(version=3, layers=_doc(llm_complete_timeout="soon"), hash="sha256:b"), + ) + result = resolve_startup(base) + assert result.layers == [] + assert result.source == "local" + assert result.validation is not None and not result.validation.ok + + +def test_store_applied_refreshes_good_and_drops_pending(base): + save_cached( + _path(base), + CachedState( + good={"version": 1, "hash": "sha256:g", "layers": []}, + pending={"version": 2, "hash": "sha256:p", "layers": [], "attempts": 1}, + ), + ) + store_applied(base, Fetched(version=3, layers=_doc(llm_complete_timeout=30), hash="sha256:n")) + cached = load_cached(_path(base)) + assert cached.good["version"] == 3 + assert cached.pending is None + + +def test_empty_layer_answer_is_a_real_answer(base, monkeypatch): + """The hive saying "no layers" must revert to local, not to the cache.""" + save_cached( + _path(base), + CachedState( + good={ + "version": 2, + "hash": "sha256:g", + "layers": _doc(llm_complete_timeout=30), + } + ), + ) + _patch_fetch( + monkeypatch, Fetched(version=5, layers=[], hash=content_hash([])) + ) + + result = resolve_startup(base) + assert result.layers == [] + assert result.source == "hive" + assert result.merged == {} diff --git a/tests/unit/overlay/test_merge.py b/tests/unit/overlay/test_merge.py new file mode 100644 index 0000000..4544ebb --- /dev/null +++ b/tests/unit/overlay/test_merge.py @@ -0,0 +1,51 @@ +"""Merge-patch semantics (RFC 7386) — the rule every overlay section relies on.""" + +from navi.overlay.merge import merge_patch + + +def test_missing_key_leaves_target_alone(): + assert merge_patch({"a": 1, "b": 2}, {"a": 9}) == {"a": 9, "b": 2} + + +def test_null_removes_the_key(): + assert merge_patch({"a": 1, "b": 2}, {"b": None}) == {"a": 1} + + +def test_null_on_missing_key_is_a_noop(): + assert merge_patch({"a": 1}, {"zzz": None}) == {"a": 1} + + +def test_objects_merge_recursively(): + target = {"settings": {"x": 1, "nested": {"keep": True, "change": 1}}} + patch = {"settings": {"nested": {"change": 2}}} + assert merge_patch(target, patch) == { + "settings": {"x": 1, "nested": {"keep": True, "change": 2}} + } + + +def test_arrays_replace_wholesale(): + # A list is a value, not a container to merge: "append one allowed command" + # is not expressible, which is intentional — partial list edits are how a + # policy ends up with values nobody ordered. + assert merge_patch({"list": [1, 2, 3]}, {"list": [9]}) == {"list": [9]} + + +def test_scalar_patch_replaces_target(): + assert merge_patch({"a": {"deep": 1}}, {"a": "flat"}) == {"a": "flat"} + + +def test_dict_patch_over_scalar_target_starts_from_empty(): + assert merge_patch({"a": "flat"}, {"a": {"deep": 1}}) == {"a": {"deep": 1}} + + +def test_non_dict_patch_replaces_everything(): + assert merge_patch({"a": 1}, [1, 2]) == [1, 2] + assert merge_patch({"a": 1}, None) is None + + +def test_inputs_are_not_mutated(): + target = {"a": {"b": 1}} + patch = {"a": {"c": 2}} + merge_patch(target, patch) + assert target == {"a": {"b": 1}} + assert patch == {"a": {"c": 2}} diff --git a/tests/unit/overlay/test_notify_route.py b/tests/unit/overlay/test_notify_route.py new file mode 100644 index 0000000..3933a80 --- /dev/null +++ b/tests/unit/overlay/test_notify_route.py @@ -0,0 +1,85 @@ +"""POST /config/notify — the hive's push, on the instance side.""" + +import asyncio + +import pytest +from fastapi import FastAPI +from httpx import ASGITransport, AsyncClient + +from navi.api.routes import config_overlay +from navi.overlay.state import get_state, reset_state +from navi.settings_model import Settings + + +@pytest.fixture(autouse=True) +def _state(): + reset_state() + state = get_state() + state.enabled = True + state.base_settings = Settings(_env_file=None) + yield + reset_state() + + +@pytest.fixture() +def app(monkeypatch): + monkeypatch.setattr("navi.swarm.verify_swarm_key", lambda key: key == "swarm-secret") + app = FastAPI() + app.include_router(config_overlay.router) + return app + + +async def _post(app, **kwargs): + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://navi") as client: + return await client.post("/config/notify", **kwargs) + + +async def test_notify_requires_the_swarm_key(app): + assert (await _post(app)).status_code == 401 + assert (await _post(app, headers={"X-Swarm-Key": "nope"})).status_code == 403 + + +async def test_notify_schedules_a_refresh(app, monkeypatch): + calls = [] + + def fake_refresh(trigger): + calls.append(trigger) + return {"status": "applied"} + + monkeypatch.setattr(config_overlay, "refresh", fake_refresh) + resp = await _post( + app, + json={"version": 3, "hash": "sha256:x"}, + headers={"X-Swarm-Key": "swarm-secret"}, + ) + assert resp.status_code == 202 + assert resp.json()["status"] == "scheduled" + + for _ in range(100): + if calls: + break + await asyncio.sleep(0.01) + assert calls == ["notify"] + + +async def test_notify_on_a_disabled_instance_does_nothing(app): + get_state().enabled = False + resp = await _post( + app, json={"version": 3}, headers={"X-Swarm-Key": "swarm-secret"} + ) + assert resp.status_code == 202 + assert "disabled" in resp.json()["detail"] + + +async def test_status_requires_the_key_and_returns_the_state(app): + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://navi") as client: + assert (await client.get("/config/status")).status_code == 401 + body = ( + await client.get( + "/config/status", headers={"X-Swarm-Key": "swarm-secret"} + ) + ).json() + assert "status" in body + assert "hash" in body diff --git a/tests/unit/overlay/test_validate.py b/tests/unit/overlay/test_validate.py new file mode 100644 index 0000000..f8678b9 --- /dev/null +++ b/tests/unit/overlay/test_validate.py @@ -0,0 +1,266 @@ +"""Overlay validation — all-or-nothing for behaviour, warn-and-drop for shape.""" + +import pytest + +from navi.overlay.validate import validate_envelope, validate_layers +from navi.settings_model import Settings + + +@pytest.fixture() +def base() -> Settings: + return Settings(_env_file=None) + + +@pytest.fixture(autouse=True) +def _empty_mcp(monkeypatch): + # Validation merges against the local server definitions; the tests must + # not depend on whatever this checkout happens to have in mcp_servers.d. + import navi.mcp.config as mcp_config + + monkeypatch.setattr(mcp_config, "load_mcp_servers", lambda path=None: {}) + + +def _layers(*layers): + return [{"name": f"layer{i}", "config": cfg} for i, cfg in enumerate(layers)] + + +def test_empty_document_is_a_valid_empty_overlay(base): + result = validate_layers(base, []) + assert result.ok + assert result.merged == {} + + +def test_unknown_section_warns_and_is_dropped(base): + result = validate_layers(base, _layers({"nonsense": {"a": 1}})) + assert result.ok + assert result.merged == {} + assert any("nonsense" in w for w in result.warnings) + + +def test_unknown_setting_is_dropped_with_a_warning(base): + result = validate_layers(base, _layers({"settings": {"no_such_field": 1}})) + assert result.ok + assert result.merged == {} + assert "settings.no_such_field" in result.unknown_fields + + +def test_denied_setting_is_refused_not_applied(base): + result = validate_layers( + base, _layers({"settings": {"database_url": "postgresql://evil/navidb"}}) + ) + assert result.ok + assert result.merged == {} + assert "settings.database_url" in result.forbidden_fields + + +def test_invalid_setting_value_rejects_the_whole_document(base): + result = validate_layers( + base, + _layers( + {"settings": {"llm_complete_timeout": 30}}, + {"settings": {"llm_complete_timeout": "not-a-number"}}, + ), + ) + # The good layer would have been useful, but a document that is half-broken + # is not a document anyone wrote — nothing from it is applied. + assert not result.ok + assert result.merged == {} + assert any("llm_complete_timeout" in e for e in result.errors) + + +def test_layers_merge_in_order(base): + result = validate_layers( + base, + _layers( + {"settings": {"llm_complete_timeout": 30, "llm_complete_retries": 1}}, + {"settings": {"llm_complete_timeout": 45}}, + ), + ) + assert result.ok + assert result.merged["settings"] == { + "llm_complete_timeout": 45, + "llm_complete_retries": 1, + } + + +# ── profiles ──────────────────────────────────────────────────────────────── + + +def test_unknown_profile_is_ignored_with_a_warning(base): + result = validate_layers(base, _layers({"profiles": {"nobody": {"temperature": 0.1}}})) + assert result.ok + assert result.merged == {} + assert any("nobody" in w for w in result.warnings) + + +def test_unknown_profile_field_is_dropped(base): + result = validate_layers( + base, _layers({"profiles": {"assistant": {"not_a_field": True}}}) + ) + assert result.ok + assert result.merged == {} + assert "profiles.assistant.not_a_field" in result.unknown_fields + + +def test_profile_change_is_kept(base): + result = validate_layers( + base, _layers({"profiles": {"assistant": {"temperature": 0.25}}}) + ) + assert result.ok + assert result.merged["profiles"]["assistant"] == {"temperature": 0.25} + + +def test_invalid_profile_value_rejects_the_document(base): + result = validate_layers( + base, _layers({"profiles": {"assistant": {"temperature": "hot"}}}) + ) + assert not result.ok + assert result.merged == {} + + +def test_profile_id_cannot_be_changed(base): + result = validate_layers(base, _layers({"profiles": {"assistant": {"id": "other"}}})) + assert not result.ok + assert any("id" in e for e in result.errors) + + +def test_tool_lists_are_validated_by_scope(base): + result = validate_layers( + base, + _layers( + { + "profiles": { + "assistant": { + "tools": { + "agent": {"native": ["filesystem", "terminal"]}, + "bogus_scope": {"native": []}, + "subagent": {"nonsense": []}, + } + } + } + } + ), + ) + assert result.ok + assert result.merged["profiles"]["assistant"]["tools"] == { + "agent": {"native": ["filesystem", "terminal"]}, + "subagent": {}, + } + assert "profiles.assistant.tools.bogus_scope" in result.unknown_fields + assert "profiles.assistant.tools.subagent.nonsense" in result.unknown_fields + + +# ── llm ───────────────────────────────────────────────────────────────────── + + +def test_llm_patch_is_merged_over_the_local_file(base, tmp_path): + local = tmp_path / "llm.json" + local.write_text( + '{"targets": {"home": {"host": "http://127.0.0.1:11434"}},' + ' "routes": {"default": [{"model": "m", "target": "home"}]}}' + ) + base = base.model_copy(update={"llm_config_file": str(local)}) + result = validate_layers( + base, + _layers({"llm": {"targets": {"home": {"host": "http://192.168.1.129:11434"}}}}), + ) + assert result.ok + assert result.merged["llm"] == { + "targets": {"home": {"host": "http://192.168.1.129:11434"}} + } + + +def test_llm_unknown_key_is_dropped(base): + result = validate_layers(base, _layers({"llm": {"routes": {}, "extra": 1}})) + assert result.ok + assert "llm.extra" in result.unknown_fields + + +def test_broken_llm_route_rejects_the_document(base, tmp_path): + local = tmp_path / "llm.json" + local.write_text( + '{"targets": {"home": {"host": "http://127.0.0.1:11434"}},' + ' "routes": {"default": [{"model": "m", "target": "home"}]}}' + ) + base = base.model_copy(update={"llm_config_file": str(local)}) + result = validate_layers( + base, _layers({"llm": {"routes": {"default": [{"model": "m", "target": "nowhere"}]}}}) + ) + assert not result.ok + assert any("nowhere" in e for e in result.errors) + + +def test_broken_local_llm_file_does_not_block_a_fix_from_above(base, tmp_path): + local = tmp_path / "llm.json" + local.write_text("{not json") + base = base.model_copy(update={"llm_config_file": str(local)}) + result = validate_layers( + base, + _layers( + { + "llm": { + "targets": {"home": {"host": "http://127.0.0.1:11434"}}, + "routes": {"default": [{"model": "m", "target": "home"}]}, + } + } + ), + ) + assert result.ok + assert any("llm.json" in w for w in result.warnings) + + +# ── mcp ───────────────────────────────────────────────────────────────────── + + +def test_mcp_server_field_is_merged(base, monkeypatch): + import navi.mcp.config as mcp_config + + existing = mcp_config.McpServerConfig(transport="stdio", command="old") + monkeypatch.setattr( + mcp_config, "load_mcp_servers", lambda path=None: {"srv": existing} + ) + result = validate_layers( + base, _layers({"mcp_servers": {"srv": {"command": "new"}}}) + ) + assert result.ok + assert result.merged["mcp_servers"] == {"srv": {"command": "new"}} + + +def test_mcp_null_is_a_merge_patch_delete_not_a_removal(base, monkeypatch): + """``null`` drops the key at the merge level, so it never reaches _clean_mcp. + + This is documented behaviour: the overlay can add or override a server + definition, never remove one. + """ + result = validate_layers(base, _layers({"mcp_servers": {"srv": None}})) + assert result.ok + assert result.merged == {} + + +def test_mcp_unknown_field_is_dropped(base, monkeypatch): + result = validate_layers(base, _layers({"mcp_servers": {"new": {"nope": 1}}})) + assert result.ok + assert "mcp_servers.new.nope" in result.unknown_fields + + +def test_invalid_mcp_config_rejects_the_document(base, monkeypatch): + result = validate_layers(base, _layers({"mcp_servers": {"srv": {"transport": "carrier-pigeon"}}})) + assert not result.ok + + +# ── envelope ──────────────────────────────────────────────────────────────── + + +def test_envelope_rejects_non_list(): + with pytest.raises(ValueError): + validate_envelope({"layers": []}) + + +def test_envelope_rejects_layer_without_config(): + with pytest.raises(ValueError): + validate_envelope([{"name": "x"}]) + + +def test_envelope_defaults_missing_name(): + usable, _ = validate_envelope([{"config": {}}]) + assert usable == [{"name": "layer0", "config": {}}] diff --git a/tests/unit/test_swarm.py b/tests/unit/test_swarm.py index 8595804..4bce9a8 100644 --- a/tests/unit/test_swarm.py +++ b/tests/unit/test_swarm.py @@ -148,4 +148,97 @@ 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 + assert "connection refused" in text + +# ── central config reporting ──────────────────────────────────────────────── + + +def test_payload_reports_config_state_and_tags(monkeypatch): + from navi.overlay.state import get_state, reset_state + + reset_state() + state = get_state() + state.version = 4 + state.hash = "sha256:abc" + state.status = "staged" + state.restart_required = ["openai_api_key"] + + ident = InstanceIdentity( + name="lab-box", + instance_id="22222222-2222-3333-4444-555555555555", + created_at="", + tags=("lab", "gpu"), + ) + announcer = HiveAnnouncer(hive_url="http://hive.test", key="k", identity=ident, port=8099) + payload = announcer.payload + + assert payload["meta"]["config"] == { + "version": 4, + "hash": "sha256:abc", + "status": "staged", + "restart_required": ["openai_api_key"], + "hot_applied": [], + "error": None, + } + assert payload["meta"]["tags"] == ["lab", "gpu"] + reset_state() + + +async def test_announce_calls_the_config_hint_from_the_reply(): + hints = [] + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, json={"ok": True, "config": {"fetch": True, "version": 7}} + ) + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler), base_url="http://hive.test") + announcer = HiveAnnouncer( + hive_url="http://hive.test", + key="secret", + identity=IDENT, + port=8099, + client=client, + on_config_hint=hints.append, + ) + assert await announcer.announce_once() is True + assert hints == [{"fetch": True, "version": 7}] + + +async def test_announce_without_a_hint_does_not_refresh(): + hints = [] + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"ok": True, "config": None}) + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler), base_url="http://hive.test") + announcer = HiveAnnouncer( + hive_url="http://hive.test", + key="secret", + identity=IDENT, + port=8099, + client=client, + on_config_hint=hints.append, + ) + await announcer.announce_once() + assert hints == [] + + +async def test_a_failing_hint_callback_does_not_fail_the_announce(): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"ok": True, "config": {"fetch": True}}) + + def boom(hint): + raise RuntimeError("nope") + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler), base_url="http://hive.test") + announcer = HiveAnnouncer( + hive_url="http://hive.test", + key="secret", + identity=IDENT, + port=8099, + client=client, + on_config_hint=boom, + ) + assert await announcer.announce_once() is True + assert announcer.get_status()["reachable"] is True