product_alert_service.py

"""
product_alert_service.py
========================
Point 2: product decision loop — gate + idempotent emit onto NodeTransport.

Backend SSOT for "should we notify?" Product emission never trusts a client
boolean; it recomputes the gate from RiskScore fields + ProductGateConfig.

States (per emission_id):
  ABSENT → EMITTED
  EMITTED + duplicate → ALREADY_EMITTED (no second transport side effect)

SUPERSEDED (same-day severity upgrade) is explicitly UNDEFINED — not implemented.
"""

from __future__ import annotations

import hashlib
import logging
from dataclasses import dataclass, field
from datetime import date, datetime, timezone
from enum import Enum
from typing import Any, Dict, Optional, Protocol, runtime_checkable

from zone_observation import AlertLevel, RiskScore
from node_transport import (
    NodeTransport,
    ProductAlert,
    alert_to_bytes,
    product_alert_from_risk,
)

logger = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# Product gate (SSOT thresholds — scorecard freeze defaults)
# ---------------------------------------------------------------------------

@dataclass(frozen=True)
class ProductGateConfig:
    """
    Product-facing gate. Distinct from training rational_termination_threshold.

    Defaults match verified backtest freeze:
      drought_risk >= 0.35 OR flood_risk >= 0.25 OR alert_level >= WARNING
    """

    drought_threshold: float = 0.35
    flood_threshold: float = 0.25
    min_alert_level: AlertLevel = AlertLevel.WARNING
    # If True, hazard thresholds alone can open the gate even when
    # AlertLevel stays ADVISORY due to confidence damping on max_risk.
    allow_hazard_override: bool = True

    def __post_init__(self) -> None:
        if not (0.0 <= self.drought_threshold <= 1.0):
            raise ValueError("drought_threshold must be in [0,1]")
        if not (0.0 <= self.flood_threshold <= 1.0):
            raise ValueError("flood_threshold must be in [0,1]")


DEFAULT_PRODUCT_GATE = ProductGateConfig()


class EmissionState(str, Enum):
    ABSENT = "absent"
    EMITTED = "emitted"


@dataclass(frozen=True)
class AlertEmissionRecord:
    emission_id: str
    zone_id: str
    valid_date: date
    state: EmissionState
    alert_level: str
    trigger: str
    drought_risk: float
    flood_risk: float
    emitted_at: datetime
    payload_sha256: str


@dataclass
class ServiceResult:
    success: bool
    outcome_code: str
    data: Dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> Dict[str, Any]:
        return {
            "success": self.success,
            "outcome_code": self.outcome_code,
            "data": dict(self.data),
        }


# ---------------------------------------------------------------------------
# Gate
# ---------------------------------------------------------------------------

def is_elevated(score: RiskScore) -> bool:
    """Internal / elevated attention: ADVISORY and above."""
    return score.alert_level.severity() >= AlertLevel.ADVISORY.severity()


def is_product_actionable(
    score: RiskScore,
    gate: ProductGateConfig = DEFAULT_PRODUCT_GATE,
) -> bool:
    if score.alert_level.severity() >= gate.min_alert_level.severity():
        return True
    if not gate.allow_hazard_override:
        return False
    if score.drought_risk >= gate.drought_threshold:
        return True
    if score.flood_risk >= gate.flood_threshold:
        return True
    return False


def product_trigger_reason(
    score: RiskScore,
    gate: ProductGateConfig = DEFAULT_PRODUCT_GATE,
) -> str:
    reasons = []
    if score.alert_level.severity() >= gate.min_alert_level.severity():
        reasons.append(f"alert_{score.alert_level.value}")
    if gate.allow_hazard_override and score.drought_risk >= gate.drought_threshold:
        reasons.append("drought_threshold")
    if gate.allow_hazard_override and score.flood_risk >= gate.flood_threshold:
        reasons.append("flood_threshold")
    return "+".join(reasons) if reasons else "none"


# ---------------------------------------------------------------------------
# Idempotency ledger (in-process SSOT for emission state)
# ---------------------------------------------------------------------------

class EmissionLedger:

    def __init__(self) -> None:
        self._records: Dict[str, AlertEmissionRecord] = {}

    def get(self, emission_id: str) -> Optional[AlertEmissionRecord]:
        return self._records.get(emission_id)

    def put(self, record: AlertEmissionRecord) -> None:
        self._records[record.emission_id] = record

    def clear(self) -> None:
        self._records.clear()

    def __len__(self) -> int:
        return len(self._records)


def make_emission_id(zone_id: str, valid_date: date) -> str:
    raw = f"{zone_id}|{valid_date.isoformat()}|product_v1"
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:32]


# ---------------------------------------------------------------------------
# Emit
# ---------------------------------------------------------------------------

async def emit_product_alert(
    transport: NodeTransport,
    score: RiskScore,
    *,
    ledger: EmissionLedger,
    gate: ProductGateConfig = DEFAULT_PRODUCT_GATE,
    valid_time: Optional[datetime] = None,
) -> ServiceResult:
    if not score.zone_id or not str(score.zone_id).strip():
        return ServiceResult(False, "INVALID_INPUT", {"error": "empty zone_id"})

    if valid_time is None:
        valid_time = datetime.now(timezone.utc)
    if valid_time.tzinfo is None:
        return ServiceResult(
            False,
            "INVALID_INPUT",
            {"error": "valid_time must be timezone-aware UTC"},
        )

    valid_date = valid_time.astimezone(timezone.utc).date()
    emission_id = make_emission_id(score.zone_id, valid_date)

    existing = ledger.get(emission_id)
    if existing is not None and existing.state == EmissionState.EMITTED:
        return ServiceResult(
            True,
            "ALREADY_EMITTED",
            {
                "emission_id": emission_id,
                "zone_id": score.zone_id,
                "valid_date": valid_date.isoformat(),
                "prior_trigger": existing.trigger,
            },
        )

    if not is_product_actionable(score, gate):
        return ServiceResult(
            True,
            "NOT_ACTIONABLE",
            {
                "emission_id": emission_id,
                "zone_id": score.zone_id,
                "valid_date": valid_date.isoformat(),
                "alert_level": score.alert_level.value,
                "drought_risk": score.drought_risk,
                "flood_risk": score.flood_risk,
            },
        )

    trigger = product_trigger_reason(score, gate)
    alert = product_alert_from_risk(
        score.zone_id,
        alert_level=score.alert_level.value,
        product_actionable=True,
        elevated=is_elevated(score),
        drought_risk=score.drought_risk,
        flood_risk=score.flood_risk,
        max_risk=max(
            score.drought_risk,
            score.flood_risk,
            score.supply_risk_composite,
            score.quality_risk_composite,
            score.fungi_contamination_prob,
        ),
        confidence=score.confidence,
        trigger=trigger,
    )
    payload = alert_to_bytes(alert)
    payload_hash = hashlib.sha256(payload).hexdigest()

    try:
        await transport.send_alert(score.zone_id, payload)
    except Exception as e:
        logger.exception("emit_product_alert transport failed: %s", e)
        return ServiceResult(
            False,
            "TRANSPORT_FAILED",
            {
                "emission_id": emission_id,
                "zone_id": score.zone_id,
                "error": str(e),
            },
        )

    record = AlertEmissionRecord(
        emission_id=emission_id,
        zone_id=score.zone_id,
        valid_date=valid_date,
        state=EmissionState.EMITTED,
        alert_level=score.alert_level.value,
        trigger=trigger,
        drought_risk=score.drought_risk,
        flood_risk=score.flood_risk,
        emitted_at=datetime.now(timezone.utc),
        payload_sha256=payload_hash,
    )
    ledger.put(record)

    return ServiceResult(
        True,
        "EMITTED",
        {
            "emission_id": emission_id,
            "zone_id": score.zone_id,
            "valid_date": valid_date.isoformat(),
            "trigger": trigger,
            "alert_level": score.alert_level.value,
            "payload_sha256": payload_hash,
        },
    )


async def score_and_maybe_emit(
    transport: NodeTransport,
    score: RiskScore,
    *,
    ledger: EmissionLedger,
    gate: ProductGateConfig = DEFAULT_PRODUCT_GATE,
    valid_time: Optional[datetime] = None,
) -> ServiceResult:
    return await emit_product_alert(
        transport,
        score,
        ledger=ledger,
        gate=gate,
        valid_time=valid_time,
    )


# ---------------------------------------------------------------------------
# Self-test
# ---------------------------------------------------------------------------

def _self_test() -> None:
    import asyncio
    from node_transport import LocalTransport

    print("product_alert_service.py self-test")

    def _score(
        zone: str,
        alert: AlertLevel,
        drought: float = 0.0,
        flood: float = 0.0,
    ) -> RiskScore:
        return RiskScore(
            zone_id=zone,
            scored_at=datetime.now(timezone.utc),
            drought_risk=drought,
            flood_risk=flood,
            alert_level=alert,
            confidence=0.63,
        )

    gate = ProductGateConfig()

    # Gate: ADVISORY low hazards → not product
    s_adv = _score("z1", AlertLevel.ADVISORY, drought=0.1, flood=0.05)
    assert is_elevated(s_adv)
    assert not is_product_actionable(s_adv, gate)
    print("  ADVISORY not product OK")

    # Gate: hazard override
    s_haz = _score("z1", AlertLevel.ADVISORY, drought=0.40, flood=0.01)
    assert is_product_actionable(s_haz, gate)
    assert "drought_threshold" in product_trigger_reason(s_haz, gate)
    print("  hazard override OK")

    # Gate: WARNING level
    s_w = _score("z1", AlertLevel.WARNING, drought=0.1, flood=0.05)
    assert is_product_actionable(s_w, gate)
    print("  WARNING level OK")

    async def _emit_tests() -> None:
        tr = LocalTransport()
        ledger = EmissionLedger()
        vt = datetime(2023, 8, 15, 12, 0, tzinfo=timezone.utc)

        # not actionable
        r0 = await emit_product_alert(tr, s_adv, ledger=ledger, gate=gate, valid_time=vt)
        assert r0.outcome_code == "NOT_ACTIONABLE"
        assert await tr.recv_alert("z1") is None
        print("  NOT_ACTIONABLE OK")

        # emit
        r1 = await emit_product_alert(tr, s_haz, ledger=ledger, gate=gate, valid_time=vt)
        assert r1.success and r1.outcome_code == "EMITTED"
        assert await tr.recv_alert("z1") is not None
        assert len(ledger) == 1
        print("  EMITTED OK")

        # duplicate
        r2 = await emit_product_alert(tr, s_haz, ledger=ledger, gate=gate, valid_time=vt)
        assert r2.outcome_code == "ALREADY_EMITTED"
        assert len(ledger) == 1
        print("  ALREADY_EMITTED OK")

        # invalid time
        r3 = await emit_product_alert(
            tr, s_haz, ledger=ledger, gate=gate,
            valid_time=datetime(2023, 8, 15),  # naive
        )
        assert r3.outcome_code == "INVALID_INPUT"
        print("  INVALID_INPUT naive time OK")

    asyncio.run(_emit_tests())
    print("All product_alert_service self-tests passed.")


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    _self_test()