Stage 188: escalation recommendation + urgency boost

core/decisions.py

@@ -17,8 +17,39 @@ class DecisionItem:
     status: str = "open"
 
 
+# --- recommendation tiers --------------------------------------------
+#
+# Stage 188 — when the α (acceleration) signal fires, the recommendation
+# is a different KIND of action, not a louder version of the same
+# action. A steady decliner needs to be reviewed in the next planning
+# cycle; an accelerating decliner needs escalation to a senior owner
+# THIS WEEK or the trajectory gets unrecoverable. So `acceleration`
+# in the evidence trail short-circuits the metric-specific suggestion
+# and returns the escalation tier instead.
+
+_ESCALATION_RECOMMENDATION = (
+    "Escalate to a senior owner now: the rate of degradation is itself "
+    "accelerating. Schedule a same-week intervention; do not defer "
+    "to the next planning cycle."
+)
+
+
 def recommendation_for_issue(entity_type: str, evidence: List[Dict]) -> str:
+    """Produce a recommendation string for the issue based on its
+    evidence trail. The text is English-only; the dashboard's
+    i18n.translateRecommendation maps it to Hebrew at render time.
+
+    Stage 188: when the ``acceleration`` evidence label is present,
+    the escalation-tier text takes precedence over any metric-
+    specific recommendation. This applies regardless of the metric
+    that's accelerating — the engine math doesn't know which metric
+    triggered α, only that the second derivative crossed the
+    threshold.
+    """
     labels = {e.get("label") for e in evidence}
+    # Acceleration trumps everything — the "kind" of action differs.
+    if "acceleration" in labels:
+        return _ESCALATION_RECOMMENDATION
     if "inventory_availability_drop" in labels:
         return "Audit inventory replenishment and check local stock allocation within 7 days."
     if "backlog_increase" in labels:
@@ -26,3 +57,20 @@ def recommendation_for_issue(entity_type: str, evidence: List[Dict]) -> str:
     if "sla_response_increase" in labels:
         return "Review service team workload and escalation policy within 72 hours."
     return "Assign an operations owner to review the detected drift and validate root cause."
+
+
+def is_escalation(decision_or_evidence) -> bool:
+    """Return True if the given evidence trail (or DecisionItem)
+    contains the acceleration label — i.e. this is the escalation
+    tier. Used by the dashboard to render an "escalation required"
+    badge without having to re-parse the recommendation string."""
+    if isinstance(decision_or_evidence, DecisionItem):
+        evidence = decision_or_evidence.evidence
+    else:
+        evidence = decision_or_evidence
+    if not evidence:
+        return False
+    for e in evidence:
+        if e.get("label") == "acceleration":
+            return True
+    return False

core/pipeline.py

@@ -304,13 +304,26 @@ def run_pipeline(observations: List[Observation],
             evidence=evidence,
         )
         result.issues.append(issue)
+        # Stage 188 — α-flagged issues are escalation-tier. Their
+        # urgency gets boosted so they sort to the top of the queue
+        # ahead of equal-severity steady decliners. Multiplier 1.5
+        # (vs 1.25 for non-accelerating) keeps it bounded in [0,1]
+        # while making the ordering decisive. An entity with
+        # drift_score 0.6 + α evidence (urgency 0.9) outranks an
+        # entity with drift_score 0.7 + no acceleration (urgency
+        # 0.875) — which mirrors the human intuition that an
+        # accelerating decliner deserves attention before a steadier
+        # but slightly-worse one.
+        alpha_value = state["signals"].get("alpha", 0.0)
+        alpha_boost = 1.5 if alpha_value >= 0.5 else 1.25
+        urgency = round(min(1.0, state["drift_score"] * alpha_boost), 4)
         result.decisions.append(DecisionItem(
             decision_id=f"decision_{idx}",
             issue_id=issue.issue_id,
             entity_id=state["entity_id"],
             created_at=state["day"],
             title=f"Review {state['entity_id']} drift",
-            urgency=round(min(1.0, state["drift_score"] * 1.25), 4),
+            urgency=urgency,
             recommendation=recommendation_for_issue(state["entity_type"], evidence),
             confidence=state["confidence"],
             evidence=evidence,