Stage 211: forecast.threshold_crossed (infra/service.py)

infra/service.py

@@ -1538,6 +1538,208 @@ class OrgStateService:
         self._forecast_summary_cache[cache_key] = (now, payload)
         return payload
 
+    # --- Stage 211 — forecast.threshold_crossed --------------------
+
+    # Default critical-score threshold matches
+    # core.forecast.severity_from_score default. Same constant the
+    # ForecastBadge / EntityDrilldown use client-side.
+    _CRITICAL_SCORE_THRESHOLD = 0.75
+    # Default early-warning window for forecast.threshold_crossed.
+    # Operators wanting tighter/looser windows pass days_threshold.
+    _DEFAULT_DAYS_THRESHOLD = 7
+    # Audit dedupe window — entities already fired in the past
+    # ``_THRESHOLD_DEDUPE_DAYS`` are skipped on the next check call
+    # so a daily cron doesn't paste the same five entities into
+    # Slack every morning. 7 days matches the default early-warning
+    # window — once you've been warned about an entity, you have a
+    # week before another reminder.
+    _THRESHOLD_DEDUPE_DAYS = 7
+
+    @staticmethod
+    def _compute_days_to_critical(
+        current_score: float,
+        slope_per_day: float,
+        critical_threshold: float = 0.75,
+    ) -> Optional[int]:
+        """Return the projected days until the score crosses the
+        critical threshold at the current slope. Returns None when
+        the answer isn't meaningful (already past threshold, or
+        not worsening).
+
+        Uses a small epsilon on the ceil() to absorb floating-
+        point precision artifacts — e.g., (0.75 - 0.60) / 0.03
+        evaluates to 5.000000000000001 in IEEE-754, which would
+        round up to 6 days without the epsilon.
+        """
+        if current_score >= critical_threshold:
+            return 0
+        if slope_per_day <= 0:
+            return None
+        import math as _math
+        raw = (critical_threshold - current_score) / slope_per_day
+        return int(_math.ceil(raw - 1e-9))
+
+    def check_forecast_thresholds(
+        self,
+        tenant_id: str,
+        *,
+        days_threshold: int = _DEFAULT_DAYS_THRESHOLD,
+        confidence_min: float = 0.2,
+        entity_cap: int = 500,
+        actor: str = "system",
+    ) -> dict:
+        """Stage 211 — scan every entity in the tenant for a forecast
+        whose projected days_to_critical falls inside
+        ``days_threshold``, fire ``forecast.threshold_crossed`` for
+        any NEW entrants (per the audit-log dedupe window), and
+        record an audit row per firing.
+
+        "New entrant" = not already fired-and-audited within the
+        past _THRESHOLD_DEDUPE_DAYS. This is computed from the
+        audit log so it survives process restarts.
+
+        Returns ``{"checked": N, "fired": [list of entity payloads],
+        "skipped_recent": [entity_ids], "ts": iso}`` for the caller
+        to log / display.
+
+        Intended to be invoked by a daily cron (POST endpoint exists
+        for manual trigger; a cron stage can wrap it).
+        """
+        self._require_tenant(tenant_id)
+        if days_threshold < 1 or days_threshold > 90:
+            raise ValueError(
+                f"days_threshold must be in [1, 90], got "
+                f"{days_threshold!r}",
+            )
+        if not (0.0 <= confidence_min <= 1.0):
+            raise ValueError(
+                "confidence_min must be in [0.0, 1.0], "
+                f"got {confidence_min!r}",
+            )
+
+        # 1. Enumerate entities + their forecasts at the standard
+        # 7-day horizon (the days_to_critical computation is
+        # horizon-independent — it uses slope_per_day directly).
+        entity_ids = self.runs.list_distinct_entity_ids(
+            tenant_id, limit=entity_cap,
+        )
+        if not entity_ids:
+            from datetime import datetime as _dt
+            from datetime import timezone as _tz
+            return {
+                "checked": 0, "fired": [], "skipped_recent": [],
+                "ts": _dt.now(_tz.utc).isoformat(),
+            }
+        forecasts = self.get_forecasts_batch(
+            tenant_id, entity_ids, horizon_days=7,
+        ) if len(entity_ids) <= 200 else [
+            self.get_entity_forecast(tenant_id, eid, horizon_days=7)
+            for eid in entity_ids
+        ]
+
+        # 2. Build the dedupe set: entity_ids that already fired
+        # forecast.threshold_crossed within the past window. Query
+        # is per-tenant so cross-tenant noise can't leak in.
+        from datetime import datetime as _dt
+        from datetime import timedelta as _td
+        from datetime import timezone as _tz
+        cutoff = (
+            _dt.now(_tz.utc) - _td(days=self._THRESHOLD_DEDUPE_DAYS)
+        ).isoformat()
+        recent_rows = self.audit.list(
+            action="forecast.threshold_crossed",
+            tenant_id=tenant_id,
+            since=cutoff,
+            limit=10000,
+        )
+        already_fired: set = set()
+        for r in recent_rows:
+            payload = r.get("payload") or {}
+            eid = payload.get("entity_id") if isinstance(payload, dict) else None
+            if isinstance(eid, str):
+                already_fired.add(eid)
+
+        # 3. Walk forecasts, fire on qualifying NEW entrants.
+        fired: List[dict] = []
+        skipped_recent: List[str] = []
+        ts_iso = _dt.now(_tz.utc).isoformat()
+        for f in forecasts:
+            conf = f.get("confidence") or 0.0
+            if conf < confidence_min:
+                continue
+            current = f.get("current_score") or 0.0
+            slope = f.get("slope_per_day") or 0.0
+            dtc = self._compute_days_to_critical(
+                current, slope, self._CRITICAL_SCORE_THRESHOLD,
+            )
+            if dtc is None or dtc > days_threshold:
+                continue
+            eid = f.get("entity_id")
+            if not isinstance(eid, str):
+                continue
+            if eid in already_fired:
+                skipped_recent.append(eid)
+                continue
+            entry = {
+                "entity_id": eid,
+                "days_to_critical": dtc,
+                "current_score": current,
+                "projected_score": f.get("projected_score"),
+                "slope_per_day": slope,
+                "confidence": conf,
+            }
+            self._fire_forecast_threshold_crossed_event(
+                tenant_id=tenant_id, entry=entry, ts=ts_iso,
+            )
+            self.audit.log(
+                actor, "forecast.threshold_crossed",
+                tenant_id=tenant_id,
+                payload=entry,
+            )
+            fired.append(entry)
+            # Avoid double-firing within the same call if list
+            # somehow re-encounters the entity.
+            already_fired.add(eid)
+
+        return {
+            "checked": len(forecasts),
+            "fired": fired,
+            "skipped_recent": skipped_recent,
+            "ts": ts_iso,
+        }
+
+    def _fire_forecast_threshold_crossed_event(
+        self, *,
+        tenant_id: str,
+        entry: dict,
+        ts: str,
+    ) -> None:
+        """Stage 211 — best-effort emission of
+        forecast.threshold_crossed. Same swallow-and-log discipline
+        as the other event fires; a delivery failure must never
+        roll back the audit-log write (the audit row is the source
+        of truth for the dedupe window)."""
+        payload = {
+            "event": "forecast.threshold_crossed",
+            "tenant_id": tenant_id,
+            "entity_id": entry["entity_id"],
+            "days_to_critical": entry["days_to_critical"],
+            "current_score": entry["current_score"],
+            "projected_score": entry["projected_score"],
+            "slope_per_day": entry["slope_per_day"],
+            "confidence": entry["confidence"],
+            "ts": ts,
+        }
+        try:
+            self.deliver_webhooks(tenant_id, payload)
+        except Exception as exc:                          # noqa: BLE001
+            import logging as _log
+            _log.getLogger(__name__).warning(
+                "forecast.threshold_crossed webhook emission "
+                "failed for tenant %r entity %r: %s",
+                tenant_id, entry.get("entity_id"), exc,
+            )
+
     # --- Stage 192 — peer benchmarking ---------------------------------
 
     def get_entity_peer_comparison(
@@ -3191,6 +3393,12 @@ class OrgStateService:
         # Subscribers can pick this INSTEAD of decision.status_changed
         # to cut Slack/Pagerduty noise on large sweeps.
         "decisions.bulk_changed",
+        # Stage 211 — fired when an entity's projected days-to-critical
+        # falls below the threshold (default 7). Operators subscribe
+        # for early-warning Slack pages instead of reading the
+        # trajectory chip manually each morning. Opt-in like the bulk
+        # event — NULL subscriptions don't get it.
+        "forecast.threshold_crossed",
     )
 
     @staticmethod
@@ -3222,7 +3430,10 @@ class OrgStateService:
     # piggy-backs on the same bulk endpoint that already fires N
     # individual decision.status_changed events; a NULL subscription
     # that received both would double-count every bulk sweep.
-    _OPT_IN_ONLY_EVENTS = frozenset({"decisions.bulk_changed"})
+    _OPT_IN_ONLY_EVENTS = frozenset({
+        "decisions.bulk_changed",
+        "forecast.threshold_crossed",  # Stage 211
+    })
 
     def _webhook_skips_event(self, webhook_id: str,
                                     event: Optional[str]) -> bool: