fix: never suppress alerts during a real outage-share spike

_should_suppress_alert only checked prev_overall != curr_overall before
folding upstream-flap deltas into the noise-suppression path. If overall
has already been parked on one bucket (e.g. DEGRADED) for an extended
period and a new outage manifests as more 5xx-flavored probes without
moving that bucket, the alert -- including to the LOCOSP #health mirror --
could get suppressed as "just more flap."

Wire the existing outage-share signal (same fraction --watch already uses
for outage-aware backoff) into the suppression decision: if outage_share
>= outage_threshold, never suppress, regardless of classification. New
params default to a no-op (0.0 >= 1.0 is always false) so existing 3-arg
callers and the pre-existing test suite are unaffected.

Author: HiroAlleyCat

test_suppression.py

@@ -195,5 +195,56 @@ def test_unclassified_present_does_not_suppress(self):
         self.assertIn("unclassified", reason)
 
 
+class TestOutageShareOverridesSuppression(unittest.TestCase):
+    """A real outage must never be classified away as upstream flap just
+    because `overall` was already sitting on the same bucket (e.g. DEGRADED
+    for weeks) when the outage hit. See BrainVault Meta note on wdgo-health
+    silence: the LOCOSP mirror only hears about state that reaches this
+    function as non-suppressed, so this is the "is the API down" gate."""
+
+    def _summary(self, **kw) -> dict:
+        base = {"improved": 0, "regressed": 0, "sideways": 0,
+                "upstream_flap_count": 0, "total_classified": 0,
+                "unclassified": 0}
+        base.update(kw)
+        return base
+
+    def test_outage_share_at_threshold_overrides_a_would_be_suppression(self):
+        # Same shape as test_all_flaps_no_net_regression_suppresses, which
+        # normally suppresses -- but a 50% outage share must win.
+        s = self._summary(upstream_flap_count=4, total_classified=4,
+                          improved=2, regressed=2)
+        suppress, reason = _should_suppress_alert(
+            "DEGRADED", "DEGRADED", s, outage_share=0.5, outage_threshold=0.30)
+        self.assertFalse(suppress)
+        self.assertIn("outage-share", reason)
+
+    def test_outage_share_below_threshold_preserves_existing_behavior(self):
+        s = self._summary(upstream_flap_count=4, total_classified=4,
+                          improved=2, regressed=2)
+        suppress, reason = _should_suppress_alert(
+            "DEGRADED", "DEGRADED", s, outage_share=0.1, outage_threshold=0.30)
+        self.assertTrue(suppress)
+        self.assertIn("upstream flap", reason)
+
+    def test_default_params_preserve_old_call_signature(self):
+        # Existing callers (and the 90/90 suite predating this change) call
+        # with 3 positional args -- defaults must be a no-op.
+        s = self._summary(upstream_flap_count=3, total_classified=3,
+                          improved=3)
+        suppress, _ = _should_suppress_alert("DEGRADED", "DEGRADED", s)
+        self.assertTrue(suppress)
+
+    def test_outage_share_wins_even_on_overall_state_change(self):
+        # Belt-and-suspenders: outage-share check runs first, so its reason
+        # takes priority even when the overall-changed check would also fire.
+        s = self._summary(upstream_flap_count=3, total_classified=3,
+                          improved=3)
+        suppress, reason = _should_suppress_alert(
+            "HEALTHY", "OUTAGE", s, outage_share=0.8, outage_threshold=0.30)
+        self.assertFalse(suppress)
+        self.assertIn("outage-share", reason)
+
+
 if __name__ == "__main__":
     unittest.main()

wdgwars_api_tester.py

@@ -909,15 +909,25 @@ def _annotate_deltas(deltas: list[str]) -> tuple[list[str], dict]:
 
 
 def _should_suppress_alert(prev_overall: str, curr_overall: str,
-                            delta_summary: dict) -> tuple[bool, str]:
+                            delta_summary: dict,
+                            outage_share: float = 0.0,
+                            outage_threshold: float = 1.0) -> tuple[bool, str]:
     """Decide whether to suppress the webhook + exec-on-change for this tick.
 
     Suppress when:
       - overall state didn't change AND
       - all classified deltas are upstream flaps AND
       - net direction isn't getting worse (regressed <= improved)
+    Never suppress when this sweep's outage_share (fraction of 429/transport-
+    error verdicts, the same signal --watch uses for outage-aware backoff)
+    meets outage_threshold — a real outage must never be classified away as
+    upstream flap just because `overall` hadn't moved yet.
     Returns (suppress: bool, reason: str).
     """
+    if outage_share >= outage_threshold:
+        return False, (f"outage-share {outage_share:.0%} >= "
+                        f"{outage_threshold:.0%} threshold (real outage, "
+                        "never suppress)")
     if prev_overall != curr_overall:
         return False, "overall state changed"
     if delta_summary["unclassified"] > 0:
@@ -2094,7 +2104,8 @@ def emit(results: list[Result], s: dict) -> None:
                                  "sent" if ok else "FAILED")
                     _, _dsum = _annotate_deltas(deltas)
                     suppress, reason = _should_suppress_alert(
-                        last_overall, overall, _dsum)
+                        last_overall, overall, _dsum,
+                        _outage_share(results), args.outage_backoff_threshold)
                     if suppress:
                         log.info("  alert SUPPRESSED: %s", reason)
                         if args.silent_webhook: