fix(monitors): store freshness staleness threshold for full-week active schedules

The staleness threshold (median gap x sensitivity factor) was only assigned
inside the schedule-aware recompute branch, which requires excluded days or a
sub-daily window. Tables with an active schedule covering all 7 days at a
non-sub-daily frequency (e.g., a 12h cadence) never stored it, so
threshold_value fell back to the deadline-based upper tolerance and the first
2-3 missed updates were not flagged.

Store the staleness threshold for every active-stage schedule: full-week
schedules use the first-pass gap thresholds, which are already in business
minutes since no days are excluded. Non-active stages keep the conservative
fallback to upper_tolerance to avoid false weekend anomalies.

Co-Authored-By: Claude <noreply@anthropic.com>

Author: aarthy-dk

testgen/commands/test_thresholds_prediction.py

@@ -89,7 +89,7 @@ def run(self) -> None:
 
             # Freshness update events are fetched as secondary data only when the suite
             # is a monitor — Volume_Trend / Metric_Trend in monitor suites couple to the
-            # Freshness_Trend signal to avoid stairstep false positives. 
+            # Freshness_Trend signal to avoid stairstep false positives.
             freshness_updates_by_table: dict[tuple[str, str], list[str]] = (
                 self._fetch_freshness_updates_by_table() if self.test_suite.is_monitor else {}
             )
@@ -268,10 +268,16 @@ def compute_freshness_threshold(
                     window_end=schedule.window_end if has_window else None,
                 )
                 lower, upper = result.lower, result.upper
-                staleness = result.staleness
             except NotEnoughData:
                 pass  # Keep first-pass thresholds
 
+        # An active schedule certifies the cadence is regular, so the tight staleness
+        # threshold (median * factor) applies. Full-week schedules skip the recompute
+        # above and use the first-pass result — with no excluded days, those gaps are
+        # already in business minutes. Non-active stages keep staleness = None so the
+        # prediction SQL falls back to upper_tolerance (avoids false weekend anomalies).
+        staleness = result.staleness
+
         # Override upper threshold with schedule-based deadline (daily/weekly only)
         if schedule.frequency != "sub_daily":
             holiday_dates = resolve_holiday_dates(holiday_codes, history.index) if holiday_codes else None
@@ -344,7 +350,7 @@ def compute_volume_or_metric_threshold(
     This avoids the "stairstep" false-positive shape where inter-change plateaus collapse the SE estimate.
     The returned prediction JSON is augmented with `freshness_gated` and `baseline_value` so
     that test execution can apply dual-branch evaluation.
-    
+
     If the filtered fit fails for any reason, falls back to fit SARIMAX on
     the raw value series and emits a prediction JSON without the freshness-gating markers.
 
@@ -381,5 +387,5 @@ def compute_volume_or_metric_threshold(
         exclude_weekends=exclude_weekends,
         holiday_codes=holiday_codes,
         schedule_tz=schedule_tz,
-    )        
+    )
     return lower, upper, None, prediction

tests/unit/common/conftest.py

@@ -294,6 +294,24 @@ def _gen_subdaily_gap_schedule_phase() -> list[tuple[pd.Timestamp, float]]:
     return _to_csv_rows(_make_observations(start, end, 2, updates))
 
 
+def _gen_twice_daily_outage() -> list[tuple[pd.Timestamp, float]]:
+    """Updates every 12h (00:00/12:00 UTC), 7 days a week, 5 weeks, then updates stop.
+
+    The 12h cadence classifies as "daily" frequency with a full-week active schedule,
+    so no excluded days or sub-daily window apply — staleness must come from the
+    first-pass gap thresholds.
+    """
+    start = datetime(2025, 10, 6, 0, 0)
+    end = datetime(2025, 11, 11, 12, 0)
+    outage_start = datetime(2025, 11, 10, 0, 0)
+    updates: set[datetime] = set()
+    d = start
+    while d < outage_start:
+        updates.add(d)
+        d += timedelta(hours=12)
+    return _to_csv_rows(_make_observations(start, end, 12, updates))
+
+
 def _gen_weekly_early() -> list[tuple[pd.Timestamp, float]]:
     start = datetime(2025, 8, 7, 10, 0)
     end = datetime(2025, 11, 6, 22, 0)

tests/unit/common/test_freshness_scenarios.py

@@ -25,6 +25,7 @@
     _gen_subdaily_gap_schedule_phase,
     _gen_subdaily_regular,
     _gen_training_only,
+    _gen_twice_daily_outage,
     _gen_weekly_early,
     _run_scenario,
 )
@@ -472,3 +473,48 @@ def test_recovery_passes(self, results: list[ScenarioPoint]) -> None:
         assert post_recovery[0].result_code == 0
         # Second update after recovery should pass
         assert post_recovery[1].result_code == 1
+
+
+# ─── Scenario 9: Twice-Daily Outage (full-week schedule) ────────────
+
+
+class Test_TwiceDailyOutage:
+    """Updates every 12h around the clock for 5 weeks, then updates stop.
+
+    Full-week "daily"-frequency schedule: no excluded days and no sub-daily
+    window, so the staleness threshold must come from the first-pass gap
+    computation. With a 720-min median gap, staleness = 720 * 0.85 = 612 min,
+    so the first missed check (gap 720 min) flags Late.
+    """
+
+    @pytest.fixture(scope="class")
+    def results(self) -> list[ScenarioPoint]:
+        rows = _gen_twice_daily_outage()
+        return _run_scenario(rows, PredictSensitivity.medium, exclude_weekends=False, tz="UTC")
+
+    def test_schedule_goes_active_all_days(self, results: list[ScenarioPoint]) -> None:
+        sched = _schedule(_updates(results)[-1])
+        assert sched is not None
+        assert sched.get("schedule_stage") == "active"
+        assert sched.get("active_days") == [0, 1, 2, 3, 4, 5, 6]
+
+    def test_staleness_stored_once_active(self, results: list[ScenarioPoint]) -> None:
+        last_update = _updates(results)[-1]
+        assert last_update.staleness == pytest.approx(720 * 0.85)
+
+    def test_no_anomalies_before_outage(self, results: list[ScenarioPoint]) -> None:
+        outage_start = pd.Timestamp("2025-11-10")
+        assert all(p.timestamp >= outage_start for p in _anomalies(results))
+
+    def test_first_missed_check_flags_late(self, results: list[ScenarioPoint]) -> None:
+        anomalies = _anomalies(results)
+        assert len(anomalies) > 0
+        first = anomalies[0]
+        assert first.timestamp == pd.Timestamp("2025-11-10 00:00")
+        assert first.value == 720
+
+    def test_all_missed_checks_flag(self, results: list[ScenarioPoint]) -> None:
+        outage_start = pd.Timestamp("2025-11-10")
+        missed_checks = [p for p in results if p.timestamp >= outage_start]
+        assert len(missed_checks) == 4
+        assert all(p.result_code == 0 for p in missed_checks)