from dataclasses import dataclass, field from datetime import date from typing import Dict, List @dataclass class DecisionItem: decision_id: str issue_id: str entity_id: str created_at: date title: str urgency: float recommendation: str confidence: float evidence: List[Dict] = field(default_factory=list) 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: return "Review warehouse capacity, picking rate, and routing load within 72 hours." 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