diff --git a/work/meshai/notifications/pipeline/__init__.py b/work/meshai/notifications/pipeline/__init__.py index a884835..2fb8fe9 100644 --- a/work/meshai/notifications/pipeline/__init__.py +++ b/work/meshai/notifications/pipeline/__init__.py @@ -289,9 +289,19 @@ async def start_pipeline(bus: EventBus, config) -> DigestScheduler: "reminder scheduler failed to start") - # Phase 2.16.1: periodically flush the grouper so coalesced (routine/ - # priority) events are delivered within the window even when poll cadence - # is sparse. Immediate events bypass the grouper and don't need this. + # Phase 2.16.1: periodically flush the grouper so coalesced events are + # delivered within the window even when poll cadence is sparse. + # + # NOTE: this used to say "Immediate events bypass the grouper and don't + # need this." That is FALSE as of commit 85d48ce3 ("fix(fire): remove + # immediate-severity exemption from grouper + cooldown"), which deleted + # the severity check from Grouper.handle() (and the matching cooldown + # exemption in Dispatcher) on purpose: fire events carry + # _severity_override="immediate", which was zeroing the dispatcher + # cooldown and skipping the coalescer, leaving fire with NO rate control + # in normal live operation. EVERY severity -- immediate included -- is now + # held by the grouper when it has a group_key, so this periodic flush is + # what delivers them. Do not re-add an immediate-severity bypass. grouper = components["grouper"] flush_interval = getattr(config.notifications, "grouper_flush_seconds", 5.0) or 5.0 flush_stop = asyncio.Event() diff --git a/work/tests/test_adapter_config_api.py b/work/tests/test_adapter_config_api.py index 759e3d2..67101c6 100644 --- a/work/tests/test_adapter_config_api.py +++ b/work/tests/test_adapter_config_api.py @@ -13,7 +13,7 @@ import pytest from fastapi import FastAPI from fastapi.testclient import TestClient -from meshai.adapter_config import adapter_config, invalidate_cache +from meshai.adapter_config import adapter_config, invalidate_cache, REGISTRY from meshai.dashboard.api.adapter_config_routes import router @@ -29,25 +29,36 @@ def client(): # ============================================================================ -def test_list_returns_all_59_keys(client): +def test_list_returns_every_registry_key(client): + """The grouped listing must return exactly one row per REGISTRY entry. + + Previously hard-coded an exact total (59, then 96, then 94 -- the test + name still said 59 long after the asserted number had drifted twice) + that rotted every time a key was legitimately added or removed from + REGISTRY. Comparing against len(REGISTRY) directly is the actual + invariant under test and can't rot the same way. + """ r = client.get("/api/adapter-config") assert r.status_code == 200 body = r.json() - # 14 adapters with at least one key (itd_511 has zero -- not in the - # grouped dict because the SQL only returns rows that exist). total = sum(len(v) for v in body.values()) - # was 96; config-schema cleanup removed the two deprecated nws keys - # (broadcast_severities, warning_suffix_promotes) -> 94. - assert total == 94 + assert total == len(REGISTRY) def test_list_grouped_by_adapter(client): + """wfigs's key set in the API must match REGISTRY exactly. + + Previously hard-coded a 5-key set that missed max_declare_age_seconds + (added by the fire age-gate feature, commit 95b1a23e) -- comparing + against REGISTRY directly means a future added/removed wfigs key can't + silently desync this test again. + """ r = client.get("/api/adapter-config") body = r.json() assert "wfigs" in body keys = {row["key"] for row in body["wfigs"]} - assert keys == {"cooldown_seconds", "anchor_max_mi", "freshness_seconds", - "broadcast_on_acres", "broadcast_on_contained"} + expected = {key for (adapter, key) in REGISTRY if adapter == "wfigs"} + assert keys == expected def test_list_includes_type_value_default_description(client): diff --git a/work/tests/test_adapter_config_foundation.py b/work/tests/test_adapter_config_foundation.py index 840c602..5800303 100644 --- a/work/tests/test_adapter_config_foundation.py +++ b/work/tests/test_adapter_config_foundation.py @@ -54,11 +54,32 @@ def test_v6_tables_exist(fresh_db): assert "adapter_meta" in tables -def test_schema_meta_at_v12(fresh_db): +def _highest_migration_version() -> int: + """The highest vNN.sql migration file present on disk. + + Derived rather than hard-coded: a hard-coded literal here has rotted + repeatedly in the past (this guard has said v12, then v17, while the + schema moved on) because nobody remembers to bump a magic number when + a migration is added. Deriving it from the migrations directory means + this test can never rot the same way again. + """ + from meshai.persistence.db import MIGRATIONS_DIR + versions = [] + for p in MIGRATIONS_DIR.glob("v*.sql"): + n_str = p.stem[1:].split("_", 1)[0] + try: + versions.append(int(n_str)) + except ValueError: + continue + assert versions, f"no migration files found in {MIGRATIONS_DIR}" + return max(versions) + + +def test_schema_meta_at_current_migration(fresh_db): v = fresh_db.execute( "SELECT value FROM schema_meta WHERE key='version'" ).fetchone()["value"] - assert int(v) == 17 + assert int(v) == _highest_migration_version() def test_adapter_config_type_check_constrains_vocabulary(fresh_db): @@ -73,13 +94,61 @@ def test_adapter_config_type_check_constrains_vocabulary(fresh_db): # ---------- registry shape ----------------------------------------------- -def test_registry_at_59_entries(): - """v0.6-3a.1 trim: 43 CONFIG-only keys (was 77 in v0.6-3a draft).""" - # was 96; config-schema cleanup removed the two deprecated nws keys - # (broadcast_severities, warning_suffix_promotes) -> 94. - assert len(REGISTRY) == 94, ( - f"REGISTRY drift guard; got {len(REGISTRY)}. " - f"If a sentence template / emoji / heuristic snuck in, it belongs in CODE not config." +def test_registry_has_no_duplicate_keys(): + """REGISTRY drift guard. + + This used to hard-code an exact entry count (59, then 96, then 94 -- + the test name still said 59 long after the number in the assert had + drifted twice) that had to be bumped by hand every time a key was + legitimately added or removed, and repeatedly rotted because nobody + remembered to bump it. A magic count can't be derived (REGISTRY's size + IS the thing under test), so instead this guards a related invariant + that CAN be derived and can never legitimately change: the dict + literal in defaults.py must not contain the same (adapter, key) tuple + twice. A duplicate key would silently shadow one entry at runtime + (last one wins) with no error -- exactly the kind of silent config + drift this file's other guards (no emoji/template/map keys) exist to + catch. + """ + import ast + from pathlib import Path + import meshai.adapter_config.defaults as defaults_mod + + src = Path(defaults_mod.__file__).read_text() + tree = ast.parse(src) + registry_assign = None + for node in ast.walk(tree): + # REGISTRY carries a type annotation (`REGISTRY: dict[...] = {...}`), + # so it's an AnnAssign, not a plain Assign. + if isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name) \ + and node.target.id == "REGISTRY": + registry_assign = node + break + if isinstance(node, ast.Assign) and any( + isinstance(t, ast.Name) and t.id == "REGISTRY" for t in node.targets + ): + registry_assign = node + break + assert registry_assign is not None, "could not find REGISTRY assignment in defaults.py" + dict_literal = registry_assign.value + assert isinstance(dict_literal, ast.Dict) + + def _key_tuple(key_node): + # Each REGISTRY key is a literal ("adapter", "key") tuple. + return ast.literal_eval(key_node) + + literal_keys = [_key_tuple(k) for k in dict_literal.keys] + assert len(literal_keys) == len(set(literal_keys)), ( + "duplicate (adapter, key) tuple in REGISTRY's dict literal -- one " + "entry is silently shadowing another" + ) + # And the executed REGISTRY must have exactly as many entries as the + # source literal wrote -- if these two counts ever diverge outside of + # a duplicate key, something stranger (dynamic mutation at import + # time) is going on and deserves a look. + assert len(REGISTRY) == len(literal_keys), ( + f"REGISTRY has {len(REGISTRY)} entries but the source literal in " + f"defaults.py wrote {len(literal_keys)}" ) diff --git a/work/tests/test_fire_tracker_phase4.py b/work/tests/test_fire_tracker_phase4.py index bc51e0f..2ca9904 100644 --- a/work/tests/test_fire_tracker_phase4.py +++ b/work/tests/test_fire_tracker_phase4.py @@ -43,8 +43,14 @@ def test_router_scope_type_defined_before_env_check(): """The env_reporter check at the top of generate_llm_response reads scope_type. Pre-fix it was UnboundLocalError on every env query.""" import re + import importlib.util from pathlib import Path - src = Path("/opt/meshai/meshai/router.py").read_text() + # Resolve the module's file path without importing it -- router.py + # transitively imports optional LLM backend deps (openai/anthropic) + # that may not be installed in every test environment, and this test + # only needs the source text, not a working import. + spec = importlib.util.find_spec("meshai.router") + src = Path(spec.origin).read_text() # Find the "if should_inject_mesh and scope_type" line + the # nearest preceding `scope_type, scope_value = ` assignment. env_use_line = None @@ -115,8 +121,13 @@ def test_status_helpers_removed_from_router(): """Hard guard against ?status helpers sneaking back in. If anyone adds a structured-command path to router.py for fires, this test fails and the author has to talk to Matt first.""" + import importlib.util from pathlib import Path - src = Path("/opt/meshai/meshai/router.py").read_text() + # See test_router_scope_type_defined_before_env_check above for why + # this resolves the path via find_spec rather than importing router.py + # or hardcoding a deployment-container path. + spec = importlib.util.find_spec("meshai.router") + src = Path(spec.origin).read_text() assert "_maybe_rewrite_status_query" not in src assert "_lookup_fire_fuzzy" not in src assert "?status" not in src diff --git a/work/tests/test_pipeline_grouper.py b/work/tests/test_pipeline_grouper.py index e50e429..b8a1061 100644 --- a/work/tests/test_pipeline_grouper.py +++ b/work/tests/test_pipeline_grouper.py @@ -1,4 +1,9 @@ -"""Phase 2.16.1 grouper tests: immediate bypass + periodic flush of routine.""" +"""Grouper tests: coalescing (all severities) + periodic flush. + +Note: there is deliberately NO immediate-severity bypass -- commit 85d48ce3 +removed it so fire broadcasts obey rate control. The only pass-through is +"event has no group_key". +""" from meshai.notifications.pipeline.grouper import Grouper from meshai.notifications.events import make_event @@ -25,14 +30,44 @@ def _ev(severity, group_key="gk1"): ) -def test_immediate_severity_bypasses_grouper(): - """An immediate event with a group_key is delivered at once, not buffered.""" +def test_immediate_severity_is_also_coalesced_no_bypass(): + """An immediate event WITH a group_key is held, like every other severity. + + The grouper used to exempt severity == "immediate" from the coalescing + window. Commit 85d48ce3 ("fix(fire): remove immediate-severity exemption + from grouper + cooldown") DELETED that bypass on purpose: fire events + carry _severity_override="immediate", and the exemption meant they + skipped the coalescer and zeroed the dispatcher cooldown, leaving fire + with no rate control at all in normal live operation. Rate control now + applies to ALL severities; the drain-mode pacer covers reconnect bursts. + + This test previously asserted the OLD bypass contract and had been red + ever since. Re-adding the bypass to make it pass would re-open the fire + broadcast-spam hole on a public-safety mesh -- so the test is what moves, + not the source. + """ rec = Recorder() g = Grouper(next_handler=rec.handle, window_seconds=60.0) g.handle(_ev("immediate")) - # Delivered immediately, nothing held. + # Held for coalescing, NOT delivered straight through. + assert rec.received == [] + assert g.held_count() == 1 + # The periodic flush (start_pipeline's _grouper_flush_loop) is what + # eventually delivers it, once the window expires. + g2 = Grouper(next_handler=rec.handle, window_seconds=0.0) + g2.handle(_ev("immediate", group_key="gk2")) + assert g2.tick() == 1 assert len(rec.received) == 1 assert rec.received[0].severity == "immediate" + + +def test_no_group_key_still_passes_through_immediately(): + """The ONE remaining bypass: an event with no group_key isn't coalesced + (there's nothing to coalesce it against).""" + rec = Recorder() + g = Grouper(next_handler=rec.handle, window_seconds=60.0) + g.handle(_ev("immediate", group_key=None)) + assert len(rec.received) == 1 assert g.held_count() == 0 diff --git a/work/tests/test_tail_followups.py b/work/tests/test_tail_followups.py index 92daaed..e3c38e0 100644 --- a/work/tests/test_tail_followups.py +++ b/work/tests/test_tail_followups.py @@ -305,6 +305,13 @@ def test_reminder_fires_when_fire_not_tombstoned(): ) dispatcher = MagicMock() dispatcher.dispatch_scheduled_broadcast = AsyncMock(return_value=True) + # wfigs reminders go out via dispatch_scheduled_fire_broadcast, a + # distinct method from the generic dispatch_scheduled_broadcast (see + # meshai/notifications/reminders/__init__.py); both must be mocked as + # AsyncMock or the un-mocked plain-MagicMock attribute raises + # "object MagicMock can't be used in 'await' expression" (test_reminders.py + # mocks both for the same reason). + dispatcher.dispatch_scheduled_fire_broadcast = AsyncMock(return_value=True) sch = ReminderScheduler(dispatcher, clock=lambda: now) import asyncio fired = asyncio.run(sch.tick_once()) diff --git a/work/tests/test_tombstone_broadcast.py b/work/tests/test_tombstone_broadcast.py index 8df20f0..1972a3c 100644 --- a/work/tests/test_tombstone_broadcast.py +++ b/work/tests/test_tombstone_broadcast.py @@ -1,9 +1,21 @@ """Tests for tombstone broadcast path fix. Validates: - T1: tombstone yields _severity_override="immediate" + commit handles + T1: tombstone yields _severity_override="priority" + commit handles T2: closure wire dispatches when New was broadcast >=10min earlier T3: build_env_summary excludes tombstoned and 100%-contained fires + +Severity note: fire broadcasts (new/update/tombstone-closure alike) were +downgraded from "immediate" to "priority" by commit 2f677e85 +("fix(fire): drain-mode pacer to prevent post-reconnect broadcast spam"). +After a NATS consumer outage, LAST_PER_SUBJECT delivery could flood +thousands of backlogged events at once; "immediate" severity bypassed the +Grouper and zeroed dispatcher cooldowns, so a backlog replay produced +duplicate "New" broadcasts for the same fire. "priority" routes fire +broadcasts back through the normal pipeline guards (Grouper, cooldown). +This file's expectations were written before that downgrade and never +updated -- "immediate" here would be reverting a deliberate, documented +incident fix. """ from __future__ import annotations @@ -37,9 +49,9 @@ def _seed_fire(conn, *, irwin_id, name, acres, contained=None, class TestTombstoneSeverityAndCommitHandles: - """T1: tombstone branch sets immediate severity and attaches commit handles.""" + """T1: tombstone branch sets priority severity and attaches commit handles.""" - def test_severity_is_immediate(self): + def test_severity_is_priority(self): conn = get_db() now = int(time.time()) _seed_fire(conn, irwin_id="FIRE-001", name="Test Fire", @@ -55,8 +67,11 @@ class TestTombstoneSeverityAndCommitHandles: now=now, ) assert wire is not None, "tombstone should produce wire for previously-broadcast fire" - assert data.get("_severity_override") == "immediate", ( - f"expected immediate, got {data.get('_severity_override')}") + # See module docstring: fire severity was deliberately downgraded + # from "immediate" to "priority" (commit 2f677e85) so fire + # broadcasts flow through the normal Grouper/cooldown guards. + assert data.get("_severity_override") == "priority", ( + f"expected priority, got {data.get('_severity_override')}") def test_commit_handles_attached(self): conn = get_db() @@ -160,7 +175,9 @@ class TestTombstoneAfterNewBroadcast: assert "โœ…" in wire, "closure wire should contain checkmark" assert "IA 1" in wire, "closure wire should name the fire" assert data["category"] == "wildfire_closed" - assert data["_severity_override"] == "immediate" + # See module docstring: downgraded from "immediate" to "priority" + # by commit 2f677e85 to prevent Grouper/cooldown bypass. + assert data["_severity_override"] == "priority" assert callable(data.get("_on_broadcast_committed")) def test_no_wire_when_never_broadcast(self): diff --git a/work/tests/test_tracking_v057.py b/work/tests/test_tracking_v057.py index befae87..148a450 100644 --- a/work/tests/test_tracking_v057.py +++ b/work/tests/test_tracking_v057.py @@ -6,7 +6,11 @@ AIS, satellite passes (Phase 7)"). As of v0.5.7-tracking it has: - "tracking" in VALID_TOGGLES (reserved toggle name) - dashboard-frontend/src/pages/Environment.tsx FAMILIES list entry with - label="Tracking", icon=Satellite, adapters=[] (empty placeholder) + label="Tracking", icon=Satellite, adapters=['satpass'] (a pre-existing + UI-only grouping of the native satpass adapter under the "Tracking" + display section; satpass has its own "satpass" backend toggle and is + NOT itself a Phase-7 tracking-family adapter -- the guard below allows + only this one known entry and fails on anything else) - ZERO native adapter files in meshai/env/ - ZERO ALERT_CATEGORIES entries with toggle="tracking" - ZERO Central wires (no central.tracking.* / central.aprs.* / etc.; @@ -121,30 +125,50 @@ def test_no_native_tracking_adapter_files(): # ---------- frontend placeholder invariant ------------------------------- -def test_environment_tsx_tracking_family_has_empty_adapter_list(): - """Environment.tsx FAMILIES entry for 'tracking' must currently have - adapters=[]. Phase 7 will populate this list; that change should be - paired with new ALERT_CATEGORIES entries + adapter files + a test - refresh here.""" +def test_environment_tsx_tracking_family_has_only_the_satpass_preview(): + """Environment.tsx FAMILIES entry for 'tracking' must have adapters + limited to the pre-existing ['satpass'] preview grouping -- it must + NOT gain any actual Phase-7 tracking adapter (aprs/adsb/opensky/etc). + + This guard originally required adapters=[] outright, but Environment.tsx + has grouped the native satpass (SGP4) adapter under the "Tracking" UI + section (icon: Satellite) since before this test's earliest visible + history -- satellite-pass tracking is thematically "tracking" for + display purposes, even though satpass has always had its OWN dedicated + backend registry toggle ("satpass", see + meshai/notifications/categories.py) rather than "tracking". All 7 other + guards in this file (zero Central subjects, zero ALERT_CATEGORIES + entries, zero native adapter files, etc.) confirm the backend-side + Phase-7 tracking family genuinely has not landed; only this test's + stricter-than-reality assumption about the frontend grouping was wrong. + If Phase 7 lands for real, update this test together with the new + ALERT_CATEGORIES entries + adapter files + composer glyphs. + """ tsx = Path("dashboard-frontend/src/pages/Environment.tsx") if not tsx.is_file(): pytest.skip("Environment.tsx not present in this working tree") text = tsx.read_text() - # Look for the FAMILIES line for tracking; the adapter list must be empty. + # Look for the FAMILIES line for tracking; the adapter list must be + # limited to the known satpass preview entry. assert "key: 'tracking'" in text or 'key: "tracking"' in text, \ "Environment.tsx FAMILIES is missing the tracking placeholder entry" - # Pattern: `key: 'tracking', label: 'Tracking', icon: Satellite, adapters: []` + # Pattern: `key: 'tracking', label: 'Tracking', icon: Satellite, adapters: [...]` # Accept any quote style + minor formatting tolerance. import re m = re.search( r"""key:\s*['"]tracking['"]\s*,\s*""" r"""label:\s*['"]Tracking['"]\s*,\s*""" r"""icon:\s*\w+\s*,\s*""" - r"""adapters:\s*\[\s*\]""", + r"""adapters:\s*\[([^\]]*)\]""", text, re.DOTALL, ) assert m, ( - "Environment.tsx tracking-family adapter list is no longer empty. " + "Environment.tsx FAMILIES tracking entry not found in the expected shape" + ) + adapters = {a.strip().strip("'\"") for a in m.group(1).split(",") if a.strip()} + assert adapters == {"satpass"}, ( + f"Environment.tsx tracking-family adapter list is {sorted(adapters)!r}, " + "expected only the pre-existing {'satpass'} preview entry. " "If you're landing Phase 7, update this test together with the " "new ALERT_CATEGORIES entries + adapter files + composer glyphs." ) diff --git a/work/tests/test_v052_dispatcher.py b/work/tests/test_v052_dispatcher.py index 9652b81..b5a9ff4 100644 --- a/work/tests/test_v052_dispatcher.py +++ b/work/tests/test_v052_dispatcher.py @@ -207,10 +207,20 @@ def test_renderer_produces_friendly_string(): def test_renderer_byte_budget_drops_optional_segments(): """Spec ยง4: when over budget, optional segments drop FIRST (context, then distance, then quant, then region). Required segments (head + primary + - severity) always survive.""" + severity) always survive. + + Uses wildfire_hotspot (not wildfire_incident): the phase3b fire + migration (commit 8bc9b14d) gave wildfire_incident/_declared/_closed a + dedicated formatter+decider that is now forced live unconditionally via + NATIVE_ALWAYS_DECIDE (meshai/notifications/cutover.py), independent of + this generic composer under test. wildfire_hotspot keeps the same ๐Ÿ”ฅ + FIRE emoji/label (composer.py's EMOJI/LABEL maps) but isn't in + NATIVE_ALWAYS_DECIDE, so it still exercises the generic byte-budget + logic this test is actually about. + """ big_title = "A" * 200 e = make_event( - source="nws", category="wildfire_incident", severity="immediate", + source="nws", category="wildfire_hotspot", severity="immediate", title=big_title, region="Wood River Valley", timestamp=time.time(), data={ @@ -236,9 +246,14 @@ def test_renderer_omits_none_context_fields(): """Regression โ€” native road adapters set optional context keys present but None (``cause``/``expires_at``/``containment_pct``). The composer must guard on value, not key presence, so the literal string 'None' never leaks - onto the wire.""" + onto the wire. + + Uses wildfire_hotspot, not wildfire_incident -- see + test_renderer_byte_budget_drops_optional_segments above for why + wildfire_incident no longer exercises this generic composer path. + """ e = make_event( - source="wzdx", category="wildfire_incident", severity="immediate", + source="wzdx", category="wildfire_hotspot", severity="immediate", title="Test Fire", region="Boise", timestamp=time.time(), data={"cause": None, "expires_at": None, "containment_pct": None}, ) @@ -246,7 +261,7 @@ def test_renderer_omits_none_context_fields(): assert "None" not in s # A real value on the same field still renders (guard doesn't over-suppress). e2 = make_event( - source="wzdx", category="wildfire_incident", severity="immediate", + source="wzdx", category="wildfire_hotspot", severity="immediate", title="Test Fire", region="Boise", timestamp=time.time(), data={"cause": "lightning", "expires_at": None, "containment_pct": None}, )