diff --git a/work/meshai/dashboard/api/config_routes.py b/work/meshai/dashboard/api/config_routes.py index 1682b92..fc183f7 100644 --- a/work/meshai/dashboard/api/config_routes.py +++ b/work/meshai/dashboard/api/config_routes.py @@ -18,23 +18,28 @@ logger = logging.getLogger(__name__) router = APIRouter(tags=["config"]) # Sections that require restart when changed. -# v0.6-tail-3: environmental added. Per Central v0.10.2 OR-not-AND -# verification (Spokane fix), env_store rebuild and CentralConsumer -# subscribe both happen only at boot. A live PUT to -# environmental..feed_source / enabled writes to disk but the -# running process keeps polling the existing native adapters AND newly -# subscribing to Central until the container restarts -- a transient -# AND-mode that violates the architecture for as long as the user -# delays the restart. +# v0.16-env-hot-reload: "environmental" and "generic_sources" REMOVED. +# Central is retired (all-native deployments, central.enabled=false, +# CentralConsumer inert), so the transient AND-mode this rule originally +# guarded against (env_store rebuild + CentralConsumer subscribe both +# boot-only, per Central v0.10.2's OR-not-AND / Spokane fix) can no longer +# happen in practice -- there is no live Central subscription to race. A PUT +# to environmental/generic_sources is now hot-applied via +# EnvironmentalStore.apply_config() (see _refresh_environmental below): +# unchanged per-adapter configs are left alone, changed native adapters are +# rebuilt in place, and store-level dedup/seen state survives because it is +# keyed by event source, not by adapter object. The ONE case that still +# needs a restart is a single adapter's feed_source flipping to/from +# "central" -- CentralConsumer's (un)subscribe is still boot-only -- and +# that is now reported per-adapter (see apply_config()'s "restart_required" +# result), not by blanket section membership here. RESTART_REQUIRED_SECTIONS = { "connection", "llm", "mesh_sources", "meshmonitor", "dashboard", - "environmental", "coverage", - "generic_sources", } # Valid config section names @@ -172,6 +177,7 @@ async def update_config_section(section: str, request: Request): # actually restarts -- otherwise the runtime would silently # switch into the transient AND-mode this commit exists to # prevent. + adapter_results = None if not restart_required and getattr(request.app.state, "config", None) is not None: try: setattr(request.app.state.config, section, new_value) @@ -179,17 +185,35 @@ async def update_config_section(section: str, request: Request): pass if section == "context": _refresh_mesh_context(request.app, new_value) + elif section == "environmental": + adapter_results = _refresh_environmental(request.app, new_value) + elif section == "generic_sources": + current_env_cfg = getattr(request.app.state.config, "environmental", None) + adapter_results = _refresh_environmental( + request.app, current_env_cfg, generic_sources=new_value) + + # A specific adapter's feed_source flip to/from "central" still + # needs a restart (CentralConsumer (un)subscribe is boot-only) -- + # apply_config() reports it per-adapter rather than by blanket + # section membership, so surface it here. + if adapter_results and any( + v == "restart_required" for v in adapter_results.values()): + restart_required = True logger.info( - "Config section %r updated, restart_required=%s changed_keys=%s", + "Config section %r updated, restart_required=%s changed_keys=%s%s", section, restart_required, changed_keys, + f" adapter_results={adapter_results}" if adapter_results is not None else "", ) - return { + response = { "saved": True, "restart_required": restart_required, "changed_keys": changed_keys, } + if adapter_results is not None: + response["adapter_results"] = adapter_results + return response except ValueError as e: raise HTTPException(status_code=422, detail=str(e)) @@ -261,6 +285,34 @@ def _refresh_toggle_filter(app) -> bool: return False +def _refresh_environmental(app, new_env_cfg, generic_sources=None): + """Best-effort live hot-reload of the running EnvironmentalStore after an + "environmental" or "generic_sources" config PUT. Delegates the actual + diff-and-swap to ``EnvironmentalStore.apply_config()`` (env/store.py): + only adapters whose OWN config changed are rebuilt in place; everything + else -- including store-level dedup/seen state -- is left untouched. + + ``generic_sources``, when omitted, defaults to the live config's current + list (an "environmental" PUT doesn't touch generic_sources); a + "generic_sources" PUT passes its own new value explicitly. + + Returns the ``{adapter_name: "reloaded"|"unchanged"|"restart_required"}`` + map from ``apply_config()``, or ``None`` when the store isn't up yet + (environmental feeds disabled, or early startup/tests). Never raises. + """ + try: + store = getattr(app.state, "env_store", None) + if store is None or new_env_cfg is None: + return None + if generic_sources is None: + config = getattr(app.state, "config", None) + generic_sources = getattr(config, "generic_sources", None) if config else None + return store.apply_config(new_env_cfg, generic_sources=generic_sources) + except Exception: + logger.exception("environmental store refresh failed") + return None + + def _refresh_mesh_context(app, new_ctx_cfg) -> bool: """Best-effort live refresh of the running MeshContext after a context config PUT. Returns True when the refresh actually fired, False if the diff --git a/work/meshai/env/store.py b/work/meshai/env/store.py index 58672a3..ffa4bd5 100644 --- a/work/meshai/env/store.py +++ b/work/meshai/env/store.py @@ -155,64 +155,33 @@ class EnvironmentalStore: # on a LATER poll broadcast. Keyed by the bare configured source name. self._generic_seeded: set[str] = set() - # Create adapter instances with error isolation - self._register_adapter("nws", config.nws, ".nws", "NWSAlertsAdapter", - lambda cfg: (cfg, self._coverage_for("nws"))) - self._register_adapter("swpc", config.swpc, ".swpc", "SWPCAdapter", - lambda cfg: (cfg,)) - self._register_adapter("ducting", config.ducting, ".ducting", "DuctingAdapter", - lambda cfg: (cfg, self._coverage_for("ducting"))) - self._register_adapter("nifc", config.fires, ".fires", "NICFFiresAdapter", - lambda cfg: (cfg, self._region_anchors, self._coverage_for("fires"))) - self._register_adapter("avalanche", config.avalanche, ".avalanche", "AvalancheAdapter", - lambda cfg: (cfg, self._coverage_for("avalanche"))) - self._register_adapter("usgs", config.usgs, ".usgs", "USGSStreamsAdapter", - lambda cfg: (cfg, self._coverage_for("usgs"))) - self._register_adapter("usgs_quake", config.usgs_quake, ".usgs_quake", "USGSQuakeAdapter", - lambda cfg: (cfg, self._coverage_for("usgs_quake"))) - self._register_adapter("traffic", config.traffic, ".traffic", "TomTomTrafficAdapter", - lambda cfg: (cfg, self._coverage_for("traffic"))) - self._register_adapter("roads511", config.roads511, ".roads511", "Roads511Adapter", - lambda cfg: (cfg, self._coverage_for("roads511"))) - self._register_adapter("wzdx", config.wzdx, ".wzdx", "WZDxAdapter", - lambda cfg: (cfg, self._coverage_for("wzdx"))) - # Native satpass TLE fetcher (storage-only: populates sat_tles, emits - # no events). Gated on satpass.feed_source=="native" like the rest. - self._register_adapter("satpass_tle", config.satpass, ".tle_fetch", "TLEFetchAdapter", - lambda cfg: (cfg,)) - # Native SGP4 pass predictor (broadcasts consolidated passes locally, - # no Central dependency). SEPARATE from the satpass_tle fetcher above; - # both are gated on satpass.enabled and feed_source=="native". - self._register_adapter("satpass", config.satpass, ".satpass", "SatpassAdapter", - lambda cfg: (cfg,)) + # Per-adapter config snapshot, keyed by the EnvironmentalConfig field + # name (e.g. "nws", "fires", "satpass" -- NOT the adapter registration + # name; "satpass" backs BOTH the satpass_tle and satpass adapters). + # Captured for every registry entry regardless of enabled/feed_source + # so a later apply_config() call can detect what changed, including + # an adapter that starts out disabled/central and is later flipped + # on. Read by apply_config()'s diff-and-swap hot-reload path. + self._configs: dict = {} + + # Create adapter instances with error isolation. The registry table + # (name, config field, module, class, arg-builder) is shared with + # apply_config()'s hot-reload path so both go through the exact same + # construction logic. + for _name, _field, _mod, _cls, _args_fn in self._adapter_registry(): + _cfg = getattr(config, _field) + self._configs[_field] = _cfg + self._register_adapter(_name, _cfg, _mod, _cls, _args_fn) # FIRMS needs reference to NIFC adapter for cross-referencing - if config.firms.enabled and config.firms.feed_source == "native": - try: - from .firms import FIRMSAdapter - fires_adapter = self._adapters.get("nifc") - self._firms = FIRMSAdapter(config.firms, self._region_anchors, fires_adapter, coverage=self._coverage_for("firms")) - self._adapters["firms"] = self._firms - except Exception as e: - err_msg = f"{type(e).__name__}: {e}" - logger.warning("Failed to initialize firms adapter: %s", err_msg) - self._failed_adapters["firms"] = err_msg + self._configs["firms"] = config.firms + self._construct_firms(config.firms) # Universal config-driven REST/GeoJSON sources. ONE adapter instance # handles every source in config.generic_sources; construct it only # when at least one source is configured. Guarded like firms so a bad # import/config never takes the whole store down. - if self._generic_sources: - try: - from .generic_http import GenericHttpAdapter - self._generic = GenericHttpAdapter( - self._generic_sources, - self._coverage_for("generic_http")) - self._adapters["generic_http"] = self._generic - except Exception as e: - err_msg = f"{type(e).__name__}: {e}" - logger.warning("Failed to initialize generic_http adapter: %s", err_msg) - self._failed_adapters["generic_http"] = err_msg + self._construct_generic(self._generic_sources) _central = [n for n in ("nws", "swpc", "ducting", "fires", "avalanche", "usgs", "usgs_quake", "traffic", "roads511", "wzdx", "firms", "satpass") if getattr(getattr(config, n, None), "feed_source", "native") == "central"] @@ -228,6 +197,56 @@ class EnvironmentalStore: self._seed_from_persistent() + def _adapter_registry(self): + """(name, config field, module path, class name, arg-builder) table + driving BOTH __init__'s adapter construction and apply_config()'s + diff-and-swap hot-reload -- the single source of truth for how each + named adapter is built from its config. Not a module-level constant: + several arg-builders close over ``self`` (``self._coverage_for``, + ``self._region_anchors``), which must reflect the store's OWN + coverage/region state (not a stale copy) on every call, including a + rebuild long after __init__. + + ``config field`` is the ``EnvironmentalConfig`` attribute name the + adapter is built from -- e.g. adapter "nifc" is built from + ``config.fires``, and BOTH the "satpass_tle" and "satpass" adapters + are built from the single ``config.satpass`` field (a change there + must rebuild both). + """ + return [ + ("nws", "nws", ".nws", "NWSAlertsAdapter", + lambda cfg: (cfg, self._coverage_for("nws"))), + ("swpc", "swpc", ".swpc", "SWPCAdapter", + lambda cfg: (cfg,)), + ("ducting", "ducting", ".ducting", "DuctingAdapter", + lambda cfg: (cfg, self._coverage_for("ducting"))), + ("nifc", "fires", ".fires", "NICFFiresAdapter", + lambda cfg: (cfg, self._region_anchors, self._coverage_for("fires"))), + ("avalanche", "avalanche", ".avalanche", "AvalancheAdapter", + lambda cfg: (cfg, self._coverage_for("avalanche"))), + ("usgs", "usgs", ".usgs", "USGSStreamsAdapter", + lambda cfg: (cfg, self._coverage_for("usgs"))), + ("usgs_quake", "usgs_quake", ".usgs_quake", "USGSQuakeAdapter", + lambda cfg: (cfg, self._coverage_for("usgs_quake"))), + ("traffic", "traffic", ".traffic", "TomTomTrafficAdapter", + lambda cfg: (cfg, self._coverage_for("traffic"))), + ("roads511", "roads511", ".roads511", "Roads511Adapter", + lambda cfg: (cfg, self._coverage_for("roads511"))), + ("wzdx", "wzdx", ".wzdx", "WZDxAdapter", + lambda cfg: (cfg, self._coverage_for("wzdx"))), + # Native satpass TLE fetcher (storage-only: populates sat_tles, + # emits no events). Gated on satpass.feed_source=="native" like + # the rest. + ("satpass_tle", "satpass", ".tle_fetch", "TLEFetchAdapter", + lambda cfg: (cfg,)), + # Native SGP4 pass predictor (broadcasts consolidated passes + # locally, no Central dependency). SEPARATE from the satpass_tle + # fetcher above; both are gated on satpass.enabled and + # feed_source=="native". + ("satpass", "satpass", ".satpass", "SatpassAdapter", + lambda cfg: (cfg,)), + ] + def _register_adapter(self, name: str, cfg, module_path: str, class_name: str, args_fn): """Register a single adapter with error isolation.""" if not cfg.enabled or cfg.feed_source != "native": @@ -241,6 +260,170 @@ class EnvironmentalStore: logger.warning("Failed to initialize %s adapter: %s", name, err_msg) self._failed_adapters[name] = err_msg + def _construct_firms(self, cfg) -> None: + """(Re)build the "firms" adapter from ``cfg`` (a FIRMSConfig), wiring + the CURRENT "nifc" adapter instance in as its cross-reference. Shared + by __init__ and apply_config()'s firms cascade so both paths build it + identically. No-ops (leaves ``self._adapters``/``self._failed_adapters`` + untouched for "firms") when disabled or not native -- callers that + need a clean slot for a rebuild must pop it first. + """ + if not (cfg.enabled and cfg.feed_source == "native"): + return + try: + from .firms import FIRMSAdapter + fires_adapter = self._adapters.get("nifc") + self._firms = FIRMSAdapter( + cfg, self._region_anchors, fires_adapter, + coverage=self._coverage_for("firms")) + self._adapters["firms"] = self._firms + except Exception as e: + err_msg = f"{type(e).__name__}: {e}" + logger.warning("Failed to initialize firms adapter: %s", err_msg) + self._failed_adapters["firms"] = err_msg + + def _construct_generic(self, sources: list) -> None: + """(Re)build the single "generic_http" adapter from ``sources`` (the + full ``config.generic_sources`` list) -- one instance handles every + configured source. Shared by __init__ and apply_config(). No-ops + (leaves any existing "generic_http" slot untouched) when ``sources`` + is empty -- callers that need a clean slot for a rebuild must pop it + first. + """ + if not sources: + return + try: + from .generic_http import GenericHttpAdapter + self._generic = GenericHttpAdapter( + sources, self._coverage_for("generic_http")) + self._adapters["generic_http"] = self._generic + except Exception as e: + err_msg = f"{type(e).__name__}: {e}" + logger.warning("Failed to initialize generic_http adapter: %s", err_msg) + self._failed_adapters["generic_http"] = err_msg + + def apply_config(self, new_cfg: "EnvironmentalConfig", *, generic_sources: list = None) -> dict: + """Hot-reload the environmental config: rebuild only what CHANGED. + + Called from the dashboard's config PUT handler in place of the old + restart-required rule. For each named adapter (per + ``_adapter_registry()``): + + * unchanged config -> the adapter object is left completely alone + (identity preserved); result ``"unchanged"``. + * changed config, still (or newly) ``feed_source == "native"`` -> + rebuilt in place via the SAME construction path __init__ uses; + result ``"reloaded"``. This covers both a field edit (e.g. + ``feed_url``) on an already-native adapter AND flipping + ``enabled`` off (the adapter is simply dropped) or on. + * ``feed_source`` toggled to/from ``"central"`` -> NOT hot-swapped + (Central (un)subscribe only happens at boot -- see + config_routes.RESTART_REQUIRED_SECTIONS' historical comment); + result ``"restart_required"``. + + Store-level dedup/seen state (``self._seen``, ``self._seeded``, + ``self._events``, ``self._fires_seeded``, ...) is keyed by event + SOURCE, not by adapter object, and is never touched here -- swapping + an adapter instance preserves it automatically. + + The "firms" adapter holds a hard reference to the "nifc" adapter + instance, so whenever "nifc" is rebuilt, "firms" is ALSO rebuilt + (even if firms' own config is unchanged) to re-fetch a fresh + reference -- otherwise it would orphan a stale nifc object. + + ``generic_sources``, when given and different from the store's + current list, rebuilds the single GenericHttpAdapter wholesale (it + holds all generic sources as one instance). + + Returns a ``{name: "reloaded"|"unchanged"|"restart_required"}`` map + covering every named adapter plus "firms" and "generic_http". + """ + old_configs = dict(self._configs) # snapshot BEFORE any mutation + result: dict = {} + nifc_reloaded = False + + for name, field, mod, cls, args_fn in self._adapter_registry(): + cfg_new = getattr(new_cfg, field) + cfg_old = old_configs.get(field) + outcome = self._apply_named_adapter( + name, field, mod, cls, args_fn, cfg_new, cfg_old) + result[name] = outcome + if name == "nifc" and outcome == "reloaded": + nifc_reloaded = True + + result["firms"] = self._apply_firms( + new_cfg.firms, old_configs.get("firms"), force=nifc_reloaded) + + if generic_sources is not None and list(generic_sources) != self._generic_sources: + self._adapters.pop("generic_http", None) + self._failed_adapters.pop("generic_http", None) + self._generic_sources = list(generic_sources) + self._construct_generic(self._generic_sources) + result["generic_http"] = "reloaded" + else: + result["generic_http"] = "unchanged" + + return result + + @staticmethod + def _feed_source_of(cfg) -> str: + return getattr(cfg, "feed_source", "native") if cfg is not None else "native" + + def _apply_named_adapter(self, name, field, mod, cls, args_fn, cfg_new, cfg_old) -> str: + """Diff one registry entry's config and rebuild in place iff needed.""" + if cfg_new == cfg_old: + return "unchanged" + + old_source = self._feed_source_of(cfg_old) + new_source = self._feed_source_of(cfg_new) + if old_source != new_source and "central" in (old_source, new_source): + # Crossing the Central boundary either direction: CentralConsumer + # (un)subscribe only happens at boot. Leave both the adapter AND + # self._configs[field] untouched so this keeps reporting + # restart_required on every subsequent PUT until an actual + # restart resolves it. + return "restart_required" + + self._configs[field] = cfg_new + if new_source != "native": + # Stayed central (or some other non-native source) both before + # and after -- no native adapter exists for this name and none + # should; nothing to hot-reload. + return "unchanged" + + # Drop any existing instance first: a disable (enabled=False) or a + # config that otherwise no longer qualifies must not leave a stale + # adapter object running, and _register_adapter() itself is a no-op + # (doesn't touch self._adapters) when the new config isn't + # enabled+native. + self._adapters.pop(name, None) + self._failed_adapters.pop(name, None) + self._register_adapter(name, cfg_new, mod, cls, args_fn) + return "reloaded" + + def _apply_firms(self, cfg_new, cfg_old, *, force: bool) -> str: + """Diff+rebuild the "firms" adapter; ``force=True`` (nifc was just + rebuilt) always rebuilds so firms re-fetches a fresh nifc reference, + even when firms' own config is byte-identical.""" + own_changed = cfg_new != cfg_old + if not (own_changed or force): + return "unchanged" + + # Only a genuine change to firms' OWN config can cross the Central + # boundary -- a pure nifc-cascade rebuild (own_changed=False) must + # never spuriously demand a restart for an untouched firms config. + if own_changed: + old_source = self._feed_source_of(cfg_old) + new_source = self._feed_source_of(cfg_new) + if old_source != new_source and "central" in (old_source, new_source): + return "restart_required" + self._configs["firms"] = cfg_new + + self._adapters.pop("firms", None) + self._failed_adapters.pop("firms", None) + self._construct_firms(cfg_new) + return "reloaded" + def _coverage_for(self, adapter: str): """Derived coverage scope for a NATIVE adapter, or None to use its own config (override/fallback). diff --git a/work/meshai/main.py b/work/meshai/main.py index fbc5b1d..82c6a9e 100644 --- a/work/meshai/main.py +++ b/work/meshai/main.py @@ -706,6 +706,16 @@ class MeshAI: else: self.env_store = None + # v0.16-env-hot-reload: expose the store to the dashboard API so a + # config PUT to /api/config/environmental (or /generic_sources) can + # hot-reload it live -- mirrors the .state.bus/.state.config stash + # above. + try: + from meshai.dashboard.server import app as _dash_app + _dash_app.state.env_store = self.env_store + except Exception: + logger.debug('dashboard app.state env_store stash skipped') + # Knowledge base (optional - Qdrant with SQLite fallback) kb_cfg = self.config.knowledge self.knowledge = None diff --git a/work/tests/test_env_hot_reload.py b/work/tests/test_env_hot_reload.py new file mode 100644 index 0000000..286cae6 --- /dev/null +++ b/work/tests/test_env_hot_reload.py @@ -0,0 +1,179 @@ +"""Config hot-reload tests for EnvironmentalStore.apply_config(). + +A config PUT to the "environmental" (or "generic_sources") section should +take effect LIVE -- only the adapter(s) whose own config actually changed are +rebuilt in place, on the SAME running store, with no container restart: + + * an unchanged adapter's object identity is preserved (untouched); + * a changed adapter (still feed_source=="native") is rebuilt in place and + the NEW instance carries the new config values; + * store-level dedup/seen state (self._seen, self._seeded, ...) is keyed by + event source, not by adapter object, and survives a swap unchanged; + * flipping feed_source to/from "central" is the one case that still needs + a restart -- it must NOT hot-swap; + * rebuilding "nifc" cascades to also rebuild "firms" (it holds a hard + reference to the nifc adapter instance). +""" +from __future__ import annotations + +import dataclasses + +from meshai.config import EnvironmentalConfig +from meshai.env.store import EnvironmentalStore + + +def _cfg(**overrides) -> EnvironmentalConfig: + """A minimal EnvironmentalConfig with usgs_quake enabled+native (has a + plain feed_url field, no network I/O at construction) so tests have a + real adapter to diff/swap, plus whatever overrides the test needs. + """ + cfg = EnvironmentalConfig() + cfg.usgs_quake = dataclasses.replace( + cfg.usgs_quake, enabled=True, feed_source="native", + feed_url="https://example.invalid/quakes-a.geojson", + ) + for key, value in overrides.items(): + setattr(cfg, key, value) + return cfg + + +def test_unchanged_adapter_config_is_a_noop(): + store = EnvironmentalStore(_cfg()) + nws_before = store._adapters["nws"] + usgs_quake_before = store._adapters["usgs_quake"] + + same_cfg = _cfg() # byte-identical field values, fresh dataclass instances + result = store.apply_config(same_cfg) + + assert result["nws"] == "unchanged" + assert result["usgs_quake"] == "unchanged" + assert store._adapters["nws"] is nws_before + assert store._adapters["usgs_quake"] is usgs_quake_before + + +def test_changed_feed_url_swaps_adapter_and_preserves_dedup_state(): + store = EnvironmentalStore(_cfg()) + nws_before = store._adapters["nws"] + usgs_quake_before = store._adapters["usgs_quake"] + assert usgs_quake_before._feed_url == "https://example.invalid/quakes-a.geojson" + + # Store-level dedup state, keyed by event SOURCE -- must survive a swap. + store._seen["usgs_quake"] = {"usgs_quake\x1eeid:us1000aaaa"} + store._seeded.add("usgs_quake") + + new_cfg = _cfg() + new_cfg.usgs_quake = dataclasses.replace( + new_cfg.usgs_quake, feed_url="https://example.invalid/quakes-b.geojson") + + result = store.apply_config(new_cfg) + + assert result["usgs_quake"] == "reloaded" + assert result["nws"] == "unchanged" + + usgs_quake_after = store._adapters["usgs_quake"] + assert usgs_quake_after is not usgs_quake_before, "changed adapter must be a NEW object" + assert usgs_quake_after._feed_url == "https://example.invalid/quakes-b.geojson", \ + "the new instance must carry the new config value" + + # Sibling adapter untouched (identity preserved). + assert store._adapters["nws"] is nws_before + + # Store-level dedup/seen state survives the swap unchanged. + assert store._seen["usgs_quake"] == {"usgs_quake\x1eeid:us1000aaaa"} + assert "usgs_quake" in store._seeded + + +def test_disabling_an_adapter_removes_it_live(): + store = EnvironmentalStore(_cfg()) + assert "usgs_quake" in store._adapters + + new_cfg = _cfg() + new_cfg.usgs_quake = dataclasses.replace(new_cfg.usgs_quake, enabled=False) + result = store.apply_config(new_cfg) + + assert result["usgs_quake"] == "reloaded" + assert "usgs_quake" not in store._adapters + + +def test_feed_source_flip_to_central_requires_restart_and_does_not_swap(): + store = EnvironmentalStore(_cfg()) + nws_before = store._adapters["nws"] + + new_cfg = _cfg() + new_cfg.nws = dataclasses.replace(new_cfg.nws, feed_source="central") + result = store.apply_config(new_cfg) + + assert result["nws"] == "restart_required" + assert store._adapters["nws"] is nws_before, "must NOT hot-swap across the Central boundary" + + # Repeated PUTs with the same (still-unapplied) value keep reporting + # restart_required rather than silently settling to "unchanged". + result2 = store.apply_config(new_cfg) + assert result2["nws"] == "restart_required" + assert store._adapters["nws"] is nws_before + + +def test_feed_source_flip_from_central_to_native_requires_restart(): + # satpass defaults feed_source="central"; flip it to native. + store = EnvironmentalStore(_cfg()) + assert "satpass" not in store._adapters # central by default -> no native instance + + new_cfg = _cfg() + new_cfg.satpass = dataclasses.replace( + new_cfg.satpass, enabled=True, feed_source="native") + result = store.apply_config(new_cfg) + + assert result["satpass"] == "restart_required" + assert result["satpass_tle"] == "restart_required" + assert "satpass" not in store._adapters + assert "satpass_tle" not in store._adapters + + +def test_nifc_rebuild_cascades_to_firms(): + cfg = _cfg() + cfg.fires = dataclasses.replace(cfg.fires, enabled=True, feed_source="native") + cfg.firms = dataclasses.replace( + cfg.firms, enabled=True, feed_source="native", map_key="test-key") + + store = EnvironmentalStore(cfg) + nifc_before = store._adapters["nifc"] + firms_before = store._adapters["firms"] + assert firms_before._fires_adapter is nifc_before + + # Change something on the "fires" (nifc) config only; firms' OWN config + # is untouched. + new_cfg = _cfg() + new_cfg.fires = dataclasses.replace(cfg.fires, state="US-NV") + new_cfg.firms = cfg.firms # byte-identical + + result = store.apply_config(new_cfg) + + assert result["nifc"] == "reloaded" + assert result["firms"] == "reloaded", \ + "firms must cascade-rebuild even though its own config didn't change" + + nifc_after = store._adapters["nifc"] + firms_after = store._adapters["firms"] + assert nifc_after is not nifc_before + assert firms_after is not firms_before + assert firms_after._fires_adapter is nifc_after, \ + "firms must hold the FRESH nifc reference, not the stale one" + + +def test_generic_sources_change_rebuilds_generic_http_only(): + store = EnvironmentalStore( + _cfg(), generic_sources=[{"name": "a", "url": "https://a.example"}]) + generic_before = store._adapters.get("generic_http") + nws_before = store._adapters["nws"] + + same_cfg = _cfg() + result = store.apply_config( + same_cfg, generic_sources=[{"name": "a", "url": "https://a.example"}]) + assert result["generic_http"] == "unchanged" + assert store._adapters.get("generic_http") is generic_before + + result2 = store.apply_config( + same_cfg, generic_sources=[{"name": "a", "url": "https://a-changed.example"}]) + assert result2["generic_http"] == "reloaded" + assert result2["nws"] == "unchanged" + assert store._adapters["nws"] is nws_before diff --git a/work/tests/test_or_arch_continuous.py b/work/tests/test_or_arch_continuous.py index 67190db..cf8746b 100644 --- a/work/tests/test_or_arch_continuous.py +++ b/work/tests/test_or_arch_continuous.py @@ -98,17 +98,42 @@ def config_app(tmp_path, monkeypatch): return app -def test_environmental_in_restart_required_sections(): +def _config_app_with_store(tmp_path): + """config_routes app with a REAL EnvironmentalStore on app.state so the + hot-reload path (apply_config) actually runs on a PUT. Mirrors config_app + but adds the store main.py stashes at boot.""" + from meshai.dashboard.api.config_routes import router as config_router + from meshai.env.store import EnvironmentalStore + from meshai.config import EnvironmentalConfig + + cfg_dir = tmp_path / "cfg" + cfg_dir.mkdir() + (cfg_dir / "config.yaml").write_text("# stub\n") + + app = FastAPI() + app.state.config = Config() + app.state.config_path = str(cfg_dir / "config.yaml") + app.state.env_store = EnvironmentalStore(EnvironmentalConfig(), event_bus=None) + app.include_router(config_router, prefix="/api") + return app + + +def test_environmental_not_in_restart_required_sections(): + # v0.16-env-hot-reload: environmental (and generic_sources) are now + # hot-applied via EnvironmentalStore.apply_config() rather than demanding a + # restart. Only a per-adapter feed_source->central flip still needs one, and + # that is reported per-adapter (see below), not by section membership. from meshai.dashboard.api.config_routes import RESTART_REQUIRED_SECTIONS - assert "environmental" in RESTART_REQUIRED_SECTIONS + assert "environmental" not in RESTART_REQUIRED_SECTIONS + assert "generic_sources" not in RESTART_REQUIRED_SECTIONS -def test_put_environmental_returns_restart_required_with_changed_keys(config_app): - """A PUT to environmental that changes feed_source returns - restart_required=true + the dotted changed_keys list.""" - client = TestClient(config_app) +def test_put_environmental_feed_source_to_central_still_requires_restart(tmp_path): + """The one surviving restart case: flipping an adapter's feed_source to + "central" is boot-only (CentralConsumer subscribe). apply_config reports it + per-adapter as "restart_required", which the PUT handler surfaces.""" + client = TestClient(_config_app_with_store(tmp_path)) - # Fetch current environmental. cur = client.get("/api/config/environmental") assert cur.status_code == 200 body = cur.json() @@ -120,12 +145,31 @@ def test_put_environmental_returns_restart_required_with_changed_keys(config_app result = r.json() assert result["saved"] is True assert result["restart_required"] is True - # The dotted key must be present. + assert result["adapter_results"]["firms"] == "restart_required" assert any(k.endswith("firms.feed_source") for k in result["changed_keys"]), ( f"changed_keys missing firms.feed_source: {result['changed_keys']}" ) +def test_put_environmental_native_field_change_hot_reloads_no_restart(tmp_path): + """A normal native-adapter field edit (nws.tick_seconds) hot-reloads that + adapter in place: restart_required=false, adapter reported "reloaded".""" + client = TestClient(_config_app_with_store(tmp_path)) + + cur = client.get("/api/config/environmental") + assert cur.status_code == 200 + body = cur.json() + # nws is enabled+native by default; bump its poll cadence. + body["nws"]["tick_seconds"] = int(body["nws"].get("tick_seconds", 60)) + 60 + + r = client.put("/api/config/environmental", json=body) + assert r.status_code == 200 + result = r.json() + assert result["restart_required"] is False + assert result["adapter_results"]["nws"] == "reloaded" + assert any(k.endswith("nws.tick_seconds") for k in result["changed_keys"]) + + def test_put_non_restart_section_returns_restart_required_false(config_app): """A PUT to bot (not restart-required) returns restart_required=false.""" client = TestClient(config_app)