mirror of
https://github.com/zvx-echo6/meshai.git
synced 2026-08-26 17:31:34 +00:00
Foundation for making all hazard formatting+gating source-agnostic. ZERO behavior change — the formatter/decider registries are empty (get_formatter/ get_decider return None → existing precomposed/Mode-B path preserved), and the shadow comparator is off unless MESHAI_SHADOW_CATEGORIES is set. - notifications/formatters/ (registry+dispatch with family fallback), gating/ (GateResult + deferred-commit contract), both empty registries. - notifications/clock.py determinism seam; route wfigs/quake/nws gating time reads through it (identical values) so goldens can freeze time. - formatters/_budget.py = copy of central/budget.py; central/budget.py is now a re-export shim (import-smoke test guards it). - compose_mesh_message consults the registry first (verbatim, no Mode-B re-cap), falls back to legacy; _resolve_budget injects per-category budget. - notifications/shadow.py + two DRY-RUN hooks (consumer._normalize, dispatcher render): compute the new result and diff-log SHADOW_MISMATCH JSONL, but NEVER commit/emit/write tables and always broadcast the OLD result. Inert by default. - tests/harness (pinned_time/pinned_tz, byte-golden, gate-sequence) + scripts/capture_fixtures.py (ephemeral read-only NATS capture); tzdata pinned. Tests: +60 (18 scaffold + 42 harness/shadow); 0 new failures (34 baseline). Co-authored-by: Matt Johnson <mj@k7zvx.com> Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
38 lines
1.2 KiB
Python
38 lines
1.2 KiB
Python
"""Phase-0 import smoke test: every meshai/central/*_handler.py module
|
|
must be importable without error after the budget shim refactor.
|
|
|
|
This guards against a broken import chain (e.g. circular imports or a bad
|
|
re-export in the shim) that would silently break all handlers.
|
|
"""
|
|
|
|
import glob
|
|
import importlib
|
|
import os
|
|
|
|
|
|
def _handler_modules():
|
|
"""Collect meshai.central.*_handler module names by globbing the source tree."""
|
|
pattern = os.path.join(
|
|
os.path.dirname(__file__),
|
|
"..", "meshai", "central", "*_handler.py",
|
|
)
|
|
paths = sorted(glob.glob(pattern))
|
|
assert paths, "No *_handler.py files found — check the glob path"
|
|
modules = []
|
|
for p in paths:
|
|
name = os.path.basename(p)[:-3] # strip .py
|
|
modules.append(f"meshai.central.{name}")
|
|
return modules
|
|
|
|
|
|
def test_all_central_handlers_importable():
|
|
"""Each *_handler module must import cleanly (no ImportError / circular deps)."""
|
|
failed = []
|
|
for mod_name in _handler_modules():
|
|
try:
|
|
importlib.import_module(mod_name)
|
|
except ImportError as exc:
|
|
failed.append(f"{mod_name}: {exc}")
|
|
assert not failed, (
|
|
"The following handler modules raised ImportError:\n" + "\n".join(failed)
|
|
)
|