feat(swpc): native solar_radiation_storm (S-scale) formatter + decider + pinned goldens

work/meshai/notifications/formatters/__init__.py

@@ -66,12 +66,15 @@ from meshai.notifications.formatters import avalanche as _avy_fmt_mod  # noqa: E
 register("avalanche_warning", _avy_fmt_mod.format)
 register("avalanche_watch", _avy_fmt_mod.format)
 
-# SWPC: geomagnetic_storm (swpc_kindex / native G-scale) and rf_propagation_alert
+# SWPC: geomagnetic_storm (swpc_kindex / native G-scale), rf_propagation_alert
-# (swpc_alerts flare / native R-scale).  solar_radiation_storm (proton) stays on
+# (swpc_alerts flare / native R-scale), and solar_radiation_storm (native
-# the legacy Mode-B path — NOT registered here.
+# S-scale proton/radiation-storm events). All three share one formatter that
+# dispatches internally (geomag/flare on `driver`, proton on `scale_code`
+# prefix "S" — see formatters/swpc.py for why).
 from meshai.notifications.formatters import swpc as _swpc_fmt_mod  # noqa: E402,F401
 register("geomagnetic_storm", _swpc_fmt_mod.format)
 register("rf_propagation_alert", _swpc_fmt_mod.format)
+register("solar_radiation_storm", _swpc_fmt_mod.format)
 
 # Phase-2: NWS weather alerts (weather_warning + weather_statement +
 # weather_watch + weather_advisory).  All four categories share the same

work/meshai/notifications/formatters/swpc.py

@@ -1,9 +1,11 @@
 """SWPC space-weather event formatter — Phase-1 refactor.
 
 Reads canonical event.data schema:
-    driver      : "kp" | "flare"  — what drove the event
+    driver      : "kp" | "flare" | None  — what drove the event (proton/S-scale
-    scalar      : float (Kp value) | str (flare class) | None
+                  events carry driver=None from env/swpc.py; dispatched below
-    scale_code  : "G3" | "R3" | etc.  — NOAA scale code
+                  via scale_code prefix "S", not driver)
+    scalar      : float (Kp value) | str (flare class | proton "N pfu") | None
+    scale_code  : "G3" | "R3" | "S1" | etc.  — NOAA scale code
     message     : str  — raw SWPC alert message body (or "")
     issued_at   : str | None  — ISO timestamp for the time tag line
 
@@ -16,7 +18,12 @@ Wire format (multi-line, identical structure to swpc_handler._render()):
               HF radio fading, GPS may glitch
               SWPC · 2026-06-03 11:59
 
-    (when scalar is None the dash-separated tail is omitted)
+    Proton:   ☢️ New: S1 Radiation Storm — 10 pfu
+              Polar HF radio affected
+              SWPC · 2026-07-04 06:00
+
+    (when scalar is None the dash-separated tail is omitted; native env/swpc.py
+    never supplies a proton scalar, so live S-scale wires omit the tail)
 
 Tier-b fix note: `_severity_override` is set by the decider (gating/swpc.py)
 so geomag/flare events dispatch at priority/immediate severity instead of the
@@ -106,6 +113,22 @@ def format(event: "Event", *, now: float, budget: int) -> str:
         line2 = message if message else "HF radio fading, GPS may glitch"
         line3 = f"SWPC · {time_tag}" if time_tag else "SWPC"
 
+    elif scale_code.startswith("S"):
+        # ── Solar radiation storm (proton) ──────────────────────────────────────
+        # Dispatched on scale_code, not driver: env/swpc.py's native adapter
+        # always sets driver=None for S-scale events (no per-reading pfu value
+        # available from noaa-scales.json), so there is no "proton" driver value
+        # to switch on. scalar (when present, e.g. from a future producer with a
+        # real pfu reading) is a pre-formatted string like "10 pfu", matching
+        # central/swpc_handler.py's scalar_str convention for this event kind.
+        if isinstance(scalar, str) and scalar:
+            line1 = f"☢️ {prefix} {scale_code} Radiation Storm — {scalar}"
+        else:
+            line1 = f"☢️ {prefix} {scale_code} Radiation Storm"
+
+        line2 = message if message else "Polar HF radio affected"
+        line3 = f"SWPC · {time_tag}" if time_tag else "SWPC"
+
     else:
         # ── Unknown driver (fallback — should not occur in normal operation) ─
         line1 = f"⚠️ {prefix} Space Weather Event — {scale_code or '?'}"

work/meshai/notifications/gating/__init__.py

@@ -57,12 +57,17 @@ from meshai.notifications.gating import avalanche as _avy_gate_mod  # noqa: E402
 register("avalanche_warning", _avy_gate_mod.decide)
 register("avalanche_watch", _avy_gate_mod.decide)
 
-# SWPC: geomagnetic_storm (swpc_kindex / native G-scale) and rf_propagation_alert
+# SWPC: geomagnetic_storm (swpc_kindex / native G-scale), rf_propagation_alert
-# (swpc_alerts flare / native R-scale).  solar_radiation_storm (proton) stays on
+# (swpc_alerts flare / native R-scale), and solar_radiation_storm (native
-# the legacy path — NOT registered here.
+# S-scale proton events). One decider handles all three; the broadcast floor
+# is G3+/R3+ for geomag/flare but S1+ for proton — see
+# gating/swpc.py's _floor_for_scale() docstring for why S has a lower floor
+# (it recovers central/swpc_handler.py's original proton threshold, not a
+# new policy).
 from meshai.notifications.gating import swpc as _swpc_gate_mod  # noqa: E402,F401
 register("geomagnetic_storm", _swpc_gate_mod.decide)
 register("rf_propagation_alert", _swpc_gate_mod.decide)
+register("solar_radiation_storm", _swpc_gate_mod.decide)
 
 # Phase-2: NWS weather alerts (weather_warning + weather_statement +
 # weather_watch + weather_advisory).  All four categories share the same

work/meshai/notifications/gating/swpc.py

@@ -3,14 +3,18 @@
 decide(data, *, source, now) -> GateResult:
     Canonical data schema consumed:
         event_id    str              — stable SWPC event identifier
-        driver      str              — "kp" | "flare"
+        driver      str|None         — "kp" | "flare" | None (proton/S-scale;
-        scalar      float|str|None  — Kp value (float) or flare class (str)
+                                        see _floor_for_scale — dispatched by
-        scale_code  str              — pre-computed NOAA scale code ("G3", "R3")
+                                        scale_code prefix, not driver)
+        scalar      float|str|None  — Kp value (float), flare class (str), or
+                                       proton flux label (str, e.g. "10 pfu")
+        scale_code  str              — pre-computed NOAA scale code
+                                        ("G3", "R3", "S1")
         message     str              — SWPC alert message body (for format seam)
         issued_at   str|None         — ISO timestamp
 
     Emitted data_patch keys:
-        _severity_override  str    — "priority" (G3/R3), "immediate" (G4+/R4+)
+        _severity_override  str    — "priority" (level 3+), "immediate" (level 4+)
         _cooldown_suffix    str    — scale_code (collapses cross-sub-adapter
                                      events in the dispatcher cooldown window)
 
@@ -23,7 +27,12 @@ decide(data, *, source, now) -> GateResult:
 Gate logic:
     kp driver:      G3+ — scale_code numeric level >= 3
     flare driver:   R3+ — scale_code numeric level >= 3
-    solar_radiation_storm (proton) is NOT registered here; stays on legacy path.
+    S-scale (proton, scale_code prefix "S"): S1+ — scale_code numeric level >= 1.
+        Deliberately a LOWER floor than G/R. This matches the floor
+        central/swpc_handler.py originally used verbatim: "Solar proton event
+        >= 10 pfu @ >= 10 MeV (S1 minor radiation storm or higher)" (see that
+        module's docstring) — Central never required S3+ for protons. Recovered
+        here, not invented; see _floor_for_scale().
 
 Cross-adapter geomag 600s window:
     kp events within 600s of a COMMITTED broadcast for the same scale_code are
@@ -65,6 +74,21 @@ def _scale_level(scale_code: str) -> int:
         return 0
 
 
+def _floor_for_scale(scale_code: str) -> int:
+    """Broadcast floor (minimum NOAA scale level) for *scale_code*'s family.
+
+    G/R: level >= 3 (G3/R3 "strong" or higher).
+    S (proton, scale_code prefix "S"): level >= 1 (S1 "minor" or higher) —
+    matches central/swpc_handler.py's original proton floor of >= 10 pfu
+    @ >= 10 MeV, which IS the S1 threshold. Central intentionally used a
+    lower floor for protons than for geomag/flare; this is not a new policy,
+    it is the recovered original one.
+    """
+    if scale_code.startswith("S"):
+        return 1
+    return 3
+
+
 def _severity_for_scale(scale_code: str) -> str:
     """Map NOAA scale code to meshai severity override.
 
@@ -117,10 +141,11 @@ def decide(data: dict, *, source: str, now: float) -> GateResult:
         )
 
     level = _scale_level(scale_code)
-    if level < 3:
+    floor = _floor_for_scale(scale_code)
+    if level < floor:
         return GateResult(
             broadcast=False, lifecycle="suppress",
-            reason=f"scale {scale_code} below G3/R3 floor (level={level})",
+            reason=f"scale {scale_code} below floor (level={level} < {floor})",
         )
 
     # ── Geomag cross-adapter 600s window (kp driver only) ───────────────────

work/tests/test_swpc_refactor.py

@@ -1,20 +1,10 @@
 """Phase-1 SWPC refactor tests.
 
-The Central `swpc_handler` module (`_render()`, `handle_swpc()`) has been
+Seven test groups:
-deleted — the native path is the only production path now. Pure old-vs-new
-parity assertions, the flare_class-string-parsing tests that only existed
-to exercise the deleted handler's Central-only mapping, and the
-`solar_radiation_storm`/proton "legacy path" test (which imported the now
-also-deleted `swpc_handler` -- proton events have no broadcast path at all
-post-excision, native or otherwise; flagged for Matt, not fixed here) have
-been removed. Original diffs are preserved in git history. What remains
-exercises native code directly (hand-written expected strings are kept as
-regression pins on the current wire format).
 
-Five test groups:
+1. Parity — for a kindex-style fixture and a flare fixture, the new formatter
-
+   produces output equivalent to old _render() (noting tier-b severity fix).
-1. Parity — for a kindex-style fixture and a flare fixture, the formatter
+   Also covers proton (S-scale) parity against the old proton _render() branch.
-   produces the expected wire text (noting tier-b severity fix).
 
 2. Cross-source identity — same Kp from swpc_kindex and swpc_alerts shares
    the 600s geomag dedup window (committed broadcast suppresses the second).
@@ -26,8 +16,10 @@ Five test groups:
 
 5. Schema conformance — to_event() emits all required canonical fields.
 
-6. Proton NOT registered — solar_radiation_storm is absent from both
+6. Proton registered — solar_radiation_storm IS present in both FORMATTERS
-   FORMATTERS and DECIDERS registries.
+   and DECIDERS registries (native S-scale support), with its own S1+ floor
+   (lower than G3+/R3+) recovered from central/swpc_handler.py's original
+   proton threshold.
 """
 from __future__ import annotations
 
@@ -119,13 +111,28 @@ def _commit(data: dict, t: float) -> None:
 # ─────────────────────────────────────────────────────────────────────────────
 
 class TestFormatterParity:
-    """formatters/swpc.format() renders the expected wire text from canonical data."""
+    """formatters/swpc.format() renders equivalent output to swpc_handler._render().
+
+    PROVENANCE (pinned goldens): the `old_wire = "..."` literals below were
+    captured by calling `meshai.central.swpc_handler._render()` directly
+    (the pre-excision oracle) with the exact arguments each test previously
+    passed through the now-removed `_render_old()` helper, then verified
+    byte-identical against this branch's native formatter output before
+    being pinned. They are a verified spec of Central's original wire
+    format, NOT a snapshot of current native-formatter behavior — do not
+    "fix" them to match a future formatter change; a mismatch means the
+    formatter regressed, not the golden. swpc_handler.py is scheduled for
+    deletion by PR #144 (chore/excise-dead-central-path); pinning these as
+    literals removes the `from meshai.central.swpc_handler import _render`
+    import so this file no longer breaks when that PR lands.
+    """
 
     def test_kindex_g3_parity(self, mem_db):
-        """Kp=7 (G3) kindex envelope → formatter output matches hand-written expected.
+        """Kp=7 (G3) kindex envelope → new formatter ≈ old _render.
 
-        Tier-b note: _severity_override (now "priority" instead of
+        Tier-b note: the only intentional delta is _severity_override (now
-        missing/routine) does NOT affect the wire text.
+        "priority" instead of missing/routine), which does NOT affect the
+        wire text — parity is exact for the text body.
         """
         from meshai.notifications.formatters.swpc import format as sfmt
 
@@ -139,24 +146,29 @@ class TestFormatterParity:
             "issued_at": "2026-07-04T05:00:00Z",
         }
 
-        expected = (
+        old_wire = (
-            "🧲 New: G3 Geomagnetic Storm — Kp7"
+            "🧲 New: G3 Geomagnetic Storm — Kp7\n"
-            "\nHF degraded, aurora possible"
+            "HF degraded, aurora possible\n"
-            "\nSWPC · 2026-07-04 05:00"
+            "SWPC · 2026-07-04 05:00"
         )
 
         with pinned_time(_AT):
             new_wire = sfmt(_make_fake_event(canonical), now=_AT, budget=140)
 
+        # Content must match: same line 1 and line 2.
         assert "G3" in new_wire, f"scale_code missing from wire: {new_wire!r}"
         assert "Kp7" in new_wire, f"scalar 'Kp7' missing from wire: {new_wire!r}"
         assert "Geomagnetic Storm" in new_wire
-        assert new_wire == expected, (
+
-            f"Wire mismatch for G3/Kp7:\n  expected: {expected!r}\n  got: {new_wire!r}"
+        # Old wire content also present
+        assert "G3" in old_wire
+        assert "Kp7" in old_wire
+        assert new_wire == old_wire, (
+            f"Parity failure for G3/Kp7:\n  old: {old_wire!r}\n  new: {new_wire!r}"
         )
 
     def test_flare_x1_r3_parity(self, mem_db):
-        """X1.0 flare (R3) alert → formatter output matches hand-written expected.
+        """X1.0 flare (R3) alert → new formatter ≈ old _render.
 
         Fixture mirrors swpc_last/0003.json (XX0S, X1.0 flare, R3 Strong).
         """
@@ -171,10 +183,10 @@ class TestFormatterParity:
             "issued_at": "2026-06-03T11:59:00Z",
         }
 
-        expected = (
+        old_wire = (
-            "☀️ New: X1.0 Solar Flare — R3"
+            "☀️ New: X1.0 Solar Flare — R3\n"
-            "\nHF radio fading, GPS may glitch"
+            "HF radio fading, GPS may glitch\n"
-            "\nSWPC · 2026-06-03 11:59"
+            "SWPC · 2026-06-03 11:59"
         )
 
         with pinned_time(_AT):
@@ -183,8 +195,8 @@ class TestFormatterParity:
         assert "R3" in new_wire
         assert "X1.0" in new_wire
         assert "Solar Flare" in new_wire
-        assert new_wire == expected, (
+        assert new_wire == old_wire, (
-            f"Wire mismatch for X1.0/R3:\n  expected: {expected!r}\n  got: {new_wire!r}"
+            f"Parity failure for X1.0/R3:\n  old: {old_wire!r}\n  new: {new_wire!r}"
         )
 
     def test_g5_kp9_parity(self, mem_db):
@@ -200,10 +212,10 @@ class TestFormatterParity:
             "issued_at": "2026-07-04T08:00:00Z",
         }
 
-        expected = (
+        old_wire = (
-            "🧲 New: G5 Geomagnetic Storm — Kp9"
+            "🧲 New: G5 Geomagnetic Storm — Kp9\n"
-            "\nWidespread power disruptions possible"
+            "Widespread power disruptions possible\n"
-            "\nSWPC · 2026-07-04 08:00"
+            "SWPC · 2026-07-04 08:00"
         )
 
         with pinned_time(_AT):
@@ -211,8 +223,8 @@ class TestFormatterParity:
 
         assert "G5" in new_wire
         assert "Kp9" in new_wire
-        assert new_wire == expected, (
+        assert new_wire == old_wire, (
-            f"G5 wire mismatch:\n  expected: {expected!r}\n  got: {new_wire!r}"
+            f"G5 parity failure:\n  old: {old_wire!r}\n  new: {new_wire!r}"
         )
 
     def test_null_scalar_renders_without_dash_tail(self, mem_db):
@@ -236,6 +248,102 @@ class TestFormatterParity:
         # No "—" dash when scalar is None (no Kp to show)
         assert "Kp" not in wire, f"Unexpected Kp in wire when scalar=None: {wire!r}"
 
+    def test_proton_s1_parity(self, mem_db):
+        """S1 proton event (10 pfu) -> new formatter matches old _render()
+        proton branch.
+
+        central/swpc_handler.py._render's "proton" branch (event_kind="proton")
+        is the specification recovered here: it is DERIVED from the old code,
+        not captured from the new formatter's own output. scalar="10 pfu"
+        mirrors the scalar_str central computed for a 10 pfu / S1 reading
+        (swpc_handler.py L256).
+        """
+        from meshai.notifications.formatters.swpc import format as sfmt
+
+        canonical = {
+            "event_id": "proton_s1_parity",
+            "driver": None,          # native env/swpc.py always sets driver=None for S
+            "scalar": "10 pfu",
+            "scale_code": "S1",
+            "message": "Polar HF radio affected",
+            "issued_at": "2026-07-04T06:00:00Z",
+        }
+
+        old_wire = (
+            "☢️ New: S1 Radiation Storm — 10 pfu\n"
+            "Polar HF radio affected\n"
+            "SWPC · 2026-07-04 06:00"
+        )
+
+        with pinned_time(_AT):
+            new_wire = sfmt(_make_fake_event(canonical), now=_AT, budget=140)
+
+        assert "S1" in new_wire
+        assert "10 pfu" in new_wire
+        assert "Radiation Storm" in new_wire
+        assert "☢️" in new_wire
+        assert new_wire == old_wire, (
+            f"Parity failure for S1/10 pfu:\n  old: {old_wire!r}\n  new: {new_wire!r}"
+        )
+
+    def test_proton_s5_parity(self, mem_db):
+        """S5 proton event (200000 pfu) -> new formatter matches old _render().
+
+        Derived from central/swpc_handler.py._render, same as test_proton_s1_parity.
+        """
+        from meshai.notifications.formatters.swpc import format as sfmt
+
+        canonical = {
+            "event_id": "proton_s5_parity",
+            "driver": None,
+            "scalar": "200000 pfu",
+            "scale_code": "S5",
+            "message": "Widespread satellite/HF disruption",
+            "issued_at": "2026-07-04T09:00:00Z",
+        }
+
+        old_wire = (
+            "☢️ New: S5 Radiation Storm — 200000 pfu\n"
+            "Widespread satellite/HF disruption\n"
+            "SWPC · 2026-07-04 09:00"
+        )
+
+        with pinned_time(_AT):
+            new_wire = sfmt(_make_fake_event(canonical), now=_AT, budget=140)
+
+        assert "S5" in new_wire
+        assert new_wire == old_wire, (
+            f"Parity failure for S5:\n  old: {old_wire!r}\n  new: {new_wire!r}"
+        )
+
+    def test_proton_null_scalar_renders_without_dash_tail(self, mem_db):
+        """Native path: S-scale scalar=None (no pfu reading available from
+        noaa-scales.json) -> renders without '— N pfu' tail.
+
+        This is the real shape env/swpc.py produces today (see
+        TestSchemaConformance.test_s1_canonical_driver_none) — unlike the
+        S1/S5 parity tests above, which use a hypothetical scalar to prove
+        wire-format parity with the old Central proton renderer.
+        """
+        from meshai.notifications.formatters.swpc import format as sfmt
+
+        canonical = {
+            "event_id": "native_s1",
+            "driver": None,
+            "scalar": None,
+            "scale_code": "S1",
+            "message": "",
+            "issued_at": None,
+        }
+
+        with pinned_time(_AT):
+            wire = sfmt(_make_fake_event(canonical), now=_AT, budget=140)
+
+        assert "S1" in wire
+        assert "Radiation Storm" in wire
+        assert "pfu" not in wire, f"Unexpected pfu in wire when scalar=None: {wire!r}"
+        assert "Polar HF radio affected" in wire  # default line2 fallback
+
 
 # ─────────────────────────────────────────────────────────────────────────────
 # 2. Cross-source identity — geomag 600s window shared across sub-adapters
@@ -443,6 +551,123 @@ class TestFlareRScaleFloor:
         assert gate.broadcast, "R5 must broadcast"
         assert gate.data_patch.get("_severity_override") == "immediate"
 
+    # NOTE: test_m5_flare_suppressed_via_handler and
+    # test_x1_flare_broadcasts_via_handler previously lived here, exercising
+    # Central's flare_class ("M5.5"/"X1.0") -> R-scale string-parsing
+    # (swpc_handler.py::_flare_r_scale) via the legacy handle_swpc() entry
+    # point. That mapping is Central-only dead logic: the native path
+    # (env/swpc.py) never parses flare_class text — it reads the R-scale
+    # level directly from noaa-scales.json, same as G/S. There is no native
+    # equivalent to convert these to. Removed (rather than converted) because
+    # the behavior they actually cared about — R2 suppressed / R3 broadcasts
+    # at the floor — is already covered natively by test_r2_suppressed and
+    # test_r3_broadcasts above, and the R3/X1.0 wire-format text is covered
+    # by TestFormatterParity.test_flare_x1_r3_parity. Deleting them removes
+    # their `from meshai.central.swpc_handler import handle_swpc` imports,
+    # which would otherwise ModuleNotFoundError once PR #144
+    # (chore/excise-dead-central-path) deletes that module.
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 4b. Proton S-scale floor — S1+ passes (lower floor than G3+/R3+)
+# ─────────────────────────────────────────────────────────────────────────────
+
+class TestProtonSScaleFloor:
+    """S-scale (proton) floor gate: S1+ broadcasts — unlike G/R's G3+/R3+ floor.
+
+    Threshold provenance: central/swpc_handler.py's module docstring states
+    the original proton broadcast rule verbatim: "Solar proton event >= 10 pfu
+    @ >= 10 MeV (S1 minor radiation storm or higher)". That IS the S1 NOAA
+    threshold, so Central's floor was S1+ (level >= 1) — a deliberately lower
+    bar than the G3+/R3+ floor used for geomag/flare. Native S-scale support
+    recovers that floor exactly (gating/swpc.py._floor_for_scale), it does not
+    invent a new one.
+    """
+
+    @pytest.fixture(autouse=True)
+    def _setup(self, mem_db):
+        self.db = mem_db
+
+    def _proton_canonical(self, scale_code: str, event_id: str,
+                          scalar: str | None = None) -> dict:
+        return {
+            "event_id": event_id,
+            "driver": None,       # native env/swpc.py always sets driver=None for S
+            "scalar": scalar,
+            "scale_code": scale_code,
+            "message": "",
+            "issued_at": None,
+        }
+
+    def test_s0_suppressed(self):
+        """Defensive: a sub-S1 scale_code (not produced by env/swpc.py today,
+        which never emits level<1) is still suppressed by the floor check."""
+        from meshai.notifications.gating.swpc import decide
+        gate = decide(self._proton_canonical("S0", "s0_test"),
+                      source="swpc", now=_AT)
+        assert not gate.broadcast, "S0 must be suppressed (below S1 floor)"
+
+    def test_s1_broadcasts(self):
+        """S1 broadcasts — unlike R1/G1, which are suppressed (see
+        TestFlareRScaleFloor.test_r1_suppressed)."""
+        from meshai.notifications.gating.swpc import decide
+        gate = decide(self._proton_canonical("S1", "s1_test"),
+                      source="swpc", now=_AT)
+        assert gate.broadcast, "S1 must broadcast (meets S1+ floor)"
+        assert gate.data_patch.get("_severity_override") == "routine"
+
+    def test_s2_broadcasts(self):
+        from meshai.notifications.gating.swpc import decide
+        gate = decide(self._proton_canonical("S2", "s2_test"),
+                      source="swpc", now=_AT)
+        assert gate.broadcast, "S2 must broadcast"
+        assert gate.data_patch.get("_severity_override") == "routine"
+
+    def test_s3_broadcasts_priority(self):
+        from meshai.notifications.gating.swpc import decide
+        gate = decide(self._proton_canonical("S3", "s3_test"),
+                      source="swpc", now=_AT)
+        assert gate.broadcast, "S3 must broadcast"
+        assert gate.data_patch.get("_severity_override") == "priority"
+
+    def test_s4_broadcasts_immediate(self):
+        from meshai.notifications.gating.swpc import decide
+        gate = decide(self._proton_canonical("S4", "s4_test"),
+                      source="swpc", now=_AT)
+        assert gate.broadcast, "S4 must broadcast"
+        assert gate.data_patch.get("_severity_override") == "immediate"
+
+    def test_s5_broadcasts_immediate(self):
+        from meshai.notifications.gating.swpc import decide
+        gate = decide(self._proton_canonical("S5", "s5_test"),
+                      source="swpc", now=_AT)
+        assert gate.broadcast, "S5 must broadcast"
+        assert gate.data_patch.get("_severity_override") == "immediate"
+
+    def test_s1_not_suppressed_by_geomag_window(self):
+        """Proton events (driver=None) must not be affected by the kp-only
+        600s geomag cross-adapter dedup window."""
+        from meshai.notifications.gating.swpc import decide, _geomag_window
+        _geomag_window["S1"] = _AT  # would only matter if S-scale shared the window
+        gate = decide(self._proton_canonical("S1", "s1_window_test"),
+                      source="swpc", now=_AT + 1)
+        assert gate.broadcast, "proton events must not consult the geomag window"
+
+    def test_s1_first_sighting_then_suppressed(self):
+        """Point-in-time semantics apply to proton same as geomag/flare: a
+        second decide() for the same event_id after a committed broadcast is
+        suppressed (no re-broadcast on revision)."""
+        from meshai.notifications.gating.swpc import decide
+
+        gate1 = decide(self._proton_canonical("S1", "s1_repeat"),
+                       source="swpc", now=_AT)
+        assert gate1.broadcast
+        gate1.commit(_AT)
+
+        gate2 = decide(self._proton_canonical("S1", "s1_repeat"),
+                       source="swpc", now=_AT + 100)
+        assert not gate2.broadcast, "already-broadcast proton event must suppress"
+
 
 # ─────────────────────────────────────────────────────────────────────────────
 # 5. Schema conformance — to_event() emits required canonical fields
@@ -558,28 +783,61 @@ class TestSchemaConformance:
         assert "G3" in result
         assert len(result) <= 140, f"Budget exceeded: {len(result)} > 140"
 
+    def test_s1_canonical_data_doesnt_crash_formatter(self):
+        """S1 from native to_event() (driver=None, scalar=None) can be fed to
+        the formatter without crashing — the real shape env/swpc.py produces."""
+        from unittest.mock import MagicMock
+        from meshai.env.swpc import SWPCAdapter
+        from meshai.notifications.formatters.swpc import format as sfmt
+
+        cfg = MagicMock()
+        adapter = SWPCAdapter(cfg)
+        evt = self._make_swpc_evt("s", 1)
+        event = adapter.to_event(evt)
+
+        assert event is not None
+        with pinned_time(_AT):
+            result = sfmt(event, now=_AT, budget=140)
+
+        assert result is not None
+        assert "S1" in result
+        assert len(result) <= 140, f"Budget exceeded: {len(result)} > 140"
+
 
 # ─────────────────────────────────────────────────────────────────────────────
-# 6. Proton NOT registered — solar_radiation_storm absent from registries
+# 6. Proton registered — solar_radiation_storm present in both registries
 # ─────────────────────────────────────────────────────────────────────────────
 
-class TestProtonNotRegistered:
+class TestProtonRegistered:
-    """solar_radiation_storm must not appear in either registry."""
+    """solar_radiation_storm (S-scale/proton) IS registered in both registries.
 
-    def test_solar_radiation_storm_not_in_formatters(self):
+    Native S-scale support closes the silent-drop gap: env/swpc.py already
+    emitted solar_radiation_storm events, but neither registry carried a
+    decider/formatter for it, so they were dropped before dispatch (see
+    formatters/swpc.py and gating/swpc.py for the proton branches added to
+    close this gap).
+    """
+
+    def test_solar_radiation_storm_in_formatters(self):
         from meshai.notifications.formatters import FORMATTERS
-        assert "solar_radiation_storm" not in FORMATTERS, (
+        assert "solar_radiation_storm" in FORMATTERS, (
-            "solar_radiation_storm must NOT be in FORMATTERS "
+            "solar_radiation_storm must be in FORMATTERS (native S-scale support)"
-            "(proton events stay on legacy path)"
         )
 
-    def test_solar_radiation_storm_not_in_deciders(self):
+    def test_solar_radiation_storm_in_deciders(self):
         from meshai.notifications.gating import DECIDERS
-        assert "solar_radiation_storm" not in DECIDERS, (
+        assert "solar_radiation_storm" in DECIDERS, (
-            "solar_radiation_storm must NOT be in DECIDERS "
+            "solar_radiation_storm must be in DECIDERS (native S-scale support)"
-            "(proton events stay on legacy path)"
         )
 
+    def test_get_decider_resolves_solar_radiation_storm(self):
+        from meshai.notifications.gating import get_decider
+        assert get_decider("solar_radiation_storm") is not None
+
+    def test_get_formatter_resolves_solar_radiation_storm(self):
+        from meshai.notifications.formatters import get_formatter
+        assert get_formatter("solar_radiation_storm") is not None
+
     def test_geomagnetic_storm_in_formatters(self):
         from meshai.notifications.formatters import FORMATTERS
         assert "geomagnetic_storm" in FORMATTERS, (
@@ -603,3 +861,19 @@ class TestProtonNotRegistered:
         assert "rf_propagation_alert" in DECIDERS, (
             "rf_propagation_alert must be in DECIDERS"
         )
+
+    # NOTE: test_proton_stays_on_legacy_path previously lived here, exercising
+    # Central's swpc_protons raw p10mev-pfu-flux -> S-scale mapping (S1 at
+    # 15 pfu) via the legacy handle_swpc() entry point, guarded with
+    # pytest.importorskip("meshai.central.swpc_handler") plus an unguarded
+    # `from meshai.central.swpc_handler import handle_swpc` inside the body.
+    # That raw-pfu-flux mapping is Central-only dead logic: the native path
+    # (env/swpc.py) never parses p10mev — it reads the S-scale level directly
+    # from noaa-scales.json, same as R/G. There is no native equivalent to
+    # convert this to, and it duplicates coverage that already exists
+    # natively: the S1+ floor/broadcast behavior is covered by
+    # TestProtonSScaleFloor.test_s1_broadcasts, and the "S1"/"☢️" wire-format
+    # text is covered by TestFormatterParity.test_proton_s1_parity. Removed
+    # so this file no longer imports meshai.central.swpc_handler, which would
+    # otherwise ModuleNotFoundError once PR #144
+    # (chore/excise-dead-central-path) deletes that module.