core/decisions.py

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