Jyotisha/scripts/rectification/decision_policy.py

from __future__ import annotations

from collections.abc import Sequence
from decimal import ROUND_FLOOR, ROUND_HALF_UP, Decimal
from typing import Any
from uuid import NAMESPACE_URL, uuid5

from scripts.active_rectification_events import CandidateScoreRow
from scripts.rectification.contracts import (
    EVENT_CONTRACT_VERSION,
    RectificationRequest,
    is_scoreable_event,
)
from scripts.rectification.scoring_service import precision_weight

POLICY_VERSION = "rectification-candidate-policy-v2"
RECEIPT_VERSION = "candidate-decision-receipt-v2"
EXECUTION_LEDGER_VERSION = "rectification-execution-ledger-v2"
SCORE_QUANTUM = Decimal("0.0001")
TIE_ABSOLUTE_TOLERANCE = Decimal("0.0001")
MIN_ACCEPTANCE_EVENTS = 3
MIN_ACCEPTANCE_DOMAINS = 2
MIN_DATE_QUALITY_MEAN = Decimal("0.65")
MIN_DIAGNOSTIC_RETENTION = Decimal("0.75")
MIN_ACCEPTANCE_MARGIN_PERCENT = Decimal("10")

_BAD_DATE_RELIABILITY = frozenset({"low", "uncertain", "unreliable"})
_CLEAR_DATE_CONFLICT = frozenset({"", "none", "resolved", "no_conflict"})


def _decimal(value: Any, default: str = "0") -> Decimal:
    if isinstance(value, bool):
        return Decimal(default)
    try:
        return Decimal(str(value))
    except Exception:
        return Decimal(default)


def _quantized_score(row: CandidateScoreRow) -> Decimal:
    return _decimal(row.get("score")).quantize(SCORE_QUANTUM, rounding=ROUND_HALF_UP)


def _relative_support(scores: Sequence[Decimal]) -> list[int]:
    if not scores:
        return []
    weights = [max(score, Decimal(0)) for score in scores]
    total = sum(weights, Decimal(0))
    if total == 0:
        base, remainder = divmod(100, len(scores))
        return [base + (1 if index < remainder else 0) for index in range(len(scores))]

    exact = [weight * Decimal(100) / total for weight in weights]
    floors = [int(value.to_integral_value(rounding=ROUND_FLOOR)) for value in exact]
    remaining = 100 - sum(floors)
    order = sorted(
        range(len(scores)),
        key=lambda index: (-(exact[index] - Decimal(floors[index])), index),
    )
    for index in order[:remaining]:
        floors[index] += 1
    return floors


def build_candidate_decisions(
    rows: Sequence[CandidateScoreRow],
    *,
    result_id: str,
) -> list[dict[str, Any]]:
    ranked = sorted(rows, key=lambda row: (-_quantized_score(row), row["time"]))
    public_rows = ranked[:3]
    supports = _relative_support([_quantized_score(row) for row in public_rows])
    all_scores = [_quantized_score(row) for row in ranked]
    decisions = []
    for index, row in enumerate(public_rows):
        score = _quantized_score(row)
        tied_minute_count = sum(
            abs(score - other) <= TIE_ABSOLUTE_TOLERANCE
            for other in all_scores
        )
        decisions.append({
            "candidate_id": str(uuid5(NAMESPACE_URL, f"{POLICY_VERSION}:{result_id}:{row['time']}")),
            "rank": index + 1,
            "time": row["time"],
            "relative_support": supports[index],
            "tied_minute_count": tied_minute_count,
        })
    return decisions


def _gate(passed: bool, **details: Any) -> dict[str, Any]:
    return {"passed": passed, **details}


def _date_quality(events: Sequence[dict[str, Any]]) -> dict[str, Any]:
    weights = [_decimal(precision_weight(str(event["precision"]))) for event in events]
    total = sum(weights, Decimal(0))
    mean = total / Decimal(len(weights)) if weights else Decimal(0)
    low_reliability = sorted(
        event["id"]
        for event in events
        if str(event.get("date_reliability") or "").strip().lower() in _BAD_DATE_RELIABILITY
    )
    unresolved_conflicts = sorted(
        event["id"]
        for event in events
        if str(event.get("date_conflict_status") or "").strip().lower() not in _CLEAR_DATE_CONFLICT
    )
    passed = bool(events) and mean >= MIN_DATE_QUALITY_MEAN and not low_reliability and not unresolved_conflicts
    return _gate(
        passed,
        precision_weight_total=float(total),
        precision_weight_mean=float(mean.quantize(SCORE_QUANTUM, rounding=ROUND_HALF_UP)),
        minimum_precision_weight_mean=float(MIN_DATE_QUALITY_MEAN),
        low_reliability_event_ids=low_reliability,
        unresolved_conflict_event_ids=unresolved_conflicts,
    )


def _diagnostic_quality(diagnostics: dict[str, Any]) -> dict[str, Any]:
    retention_names = (
        "leave_one_event_out_retention_rate",
        "leave_one_domain_out_retention_rate",
        "date_sensitivity_retention_rate",
    )
    retentions = {name: _decimal(diagnostics.get(name)) for name in retention_names}
    margin = _decimal(diagnostics.get("primary_secondary_margin_percent"))
    passed = (
        all(value >= MIN_DIAGNOSTIC_RETENTION for value in retentions.values())
        and margin >= MIN_ACCEPTANCE_MARGIN_PERCENT
    )
    return _gate(
        passed,
        minimum_retention=float(MIN_DIAGNOSTIC_RETENTION),
        minimum_margin_percent=float(MIN_ACCEPTANCE_MARGIN_PERCENT),
        margin_percent=float(margin),
        **{name: float(value) for name, value in retentions.items()},
    )


def build_decision_receipt(
    request: RectificationRequest,
    candidate_decisions: Sequence[dict[str, Any]],
    built: dict[str, Any],
    diagnostics: dict[str, Any],
) -> dict[str, Any]:
    scoreable_events = [event for event in request["events"] if is_scoreable_event(event)]
    domains = sorted({event["domain"] for event in scoreable_events})
    candidate_presence = _gate(bool(candidate_decisions), candidate_count=len(candidate_decisions))
    event_quality = _gate(
        len(scoreable_events) >= MIN_ACCEPTANCE_EVENTS,
        scoreable_event_count=len(scoreable_events),
        minimum=MIN_ACCEPTANCE_EVENTS,
    )
    domain_diversity = _gate(
        len(domains) >= MIN_ACCEPTANCE_DOMAINS,
        scoreable_domain_count=len(domains),
        minimum=MIN_ACCEPTANCE_DOMAINS,
        domains=domains,
    )
    date_quality = _date_quality(scoreable_events)
    top_tied_count = candidate_decisions[0]["tied_minute_count"] if candidate_decisions else 0
    unique_top = _gate(top_tied_count == 1, tied_minute_count=top_tied_count)
    diagnostic_quality = _diagnostic_quality(diagnostics)
    required_layers = _gate(
        not built.get("missing_layers"),
        missing_layers=sorted(built.get("missing_layers") or []),
    )

    acceptance_allowed = all((
        candidate_presence["passed"],
        event_quality["passed"],
        domain_diversity["passed"],
        date_quality["passed"],
        unique_top["passed"],
        diagnostic_quality["passed"],
    ))
    margin = _decimal(diagnostics.get("primary_secondary_margin_percent"))
    if acceptance_allowed and margin >= Decimal("20"):
        overall_confidence = "high"
    elif acceptance_allowed:
        overall_confidence = "medium"
    else:
        overall_confidence = "low"

    acceptance_reasons: list[str] = []
    for passed, reason in (
        (candidate_presence["passed"], "no_candidates"),
        (event_quality["passed"], "insufficient_events"),
        (domain_diversity["passed"], "insufficient_domain_diversity"),
        (date_quality["passed"], "low_date_quality"),
        (unique_top["passed"], "tied_top_score"),
        (diagnostic_quality["passed"], "insufficient_diagnostic_stability"),
    ):
        if not passed:
            acceptance_reasons.append(reason)
    confirmation_reasons = []
    if not required_layers["passed"]:
        confirmation_reasons.append("missing_mandatory_layers")
    confirmation_reasons.extend(("engine_exact_confirmation_not_granted", "external_validation_not_passed"))
    reasons = [*acceptance_reasons, *confirmation_reasons]

    representative = candidate_decisions[0] if candidate_decisions else None
    exact_confirmation = {
        "passed": False,
        "fail_closed": True,
        "engine_granted": False,
        "external_validation_status": "not_evaluated",
        "required_scoreable_events": 4,
        "required_scoreable_domains": 3,
        "reason": "engine_and_external_validation_must_explicitly_pass",
    }
    receipt = {
        "receipt_version": RECEIPT_VERSION,
        "contract_version": "v2",
        "event_contract_version": EVENT_CONTRACT_VERSION,
        "policy_version": POLICY_VERSION,
        "decision_policy_version": POLICY_VERSION,
        "display_allowed": bool(candidate_decisions),
        "selection_allowed": acceptance_allowed,
        "acceptance_allowed": acceptance_allowed,
        "confirmation_allowed": False,
        "accept_allowed": acceptance_allowed,
        "confirm_allowed": False,
        "representative_candidate_id": representative["candidate_id"] if representative else None,
        "representative_time": representative["time"] if representative else None,
        "overall_confidence": overall_confidence,
        "margin_percent": float(margin),
        "reasons": reasons,
        "acceptance_reasons": acceptance_reasons,
        "confirmation_reasons": confirmation_reasons,
        "tie_policy": {
            "score_quantum": float(SCORE_QUANTUM),
            "absolute_tolerance": float(TIE_ABSOLUTE_TOLERANCE),
            "rounding": "ROUND_HALF_UP",
        },
        "gates": {
            "candidate_presence": candidate_presence,
            "event_quality": event_quality,
            "domain_diversity": domain_diversity,
            "date_quality": date_quality,
            "unique_top": unique_top,
            "diagnostic_quality": diagnostic_quality,
            "required_layers": required_layers,
            "exact_confirmation": exact_confirmation,
        },
    }
    return receipt


def build_execution_ledger(
    request: RectificationRequest,
    built: dict[str, Any],
    diagnostics: dict[str, Any],
    candidate_decisions: Sequence[dict[str, Any]],
) -> list[dict[str, Any]]:
    matrix = built.get("matrix") or {}
    date_sensitivity = {
        item.get("event_id"): item
        for item in built.get("date_sensitivity") or []
        if isinstance(item, dict)
    }
    entries: list[dict[str, Any]] = []
    all_layers: set[str] = set()
    for event in request["events"]:
        candidates = matrix.get(event["id"], {})
        layers = sorted({
            layer
            for contribution in candidates.values()
            for layer in contribution.get("technique_layers", [])
        })
        all_layers.update(layers)
        sensitivity = date_sensitivity.get(event["id"], {})
        scoreable = is_scoreable_event(event)
        entries.append({
            "ledger_version": EXECUTION_LEDGER_VERSION,
            "stage": "event_scoring",
            "status": "executed" if scoreable and candidates else "not_executed" if scoreable else "retained_not_scored",
            "source": "python-engine",
            "event_id": event["id"],
            "domain": event["domain"],
            "event_kind": event["event_kind"],
            "date_precision": event["precision"],
            "precision_weight": precision_weight(event["precision"]),
            "sampled_date_count": len(sensitivity.get("sample_dates") or []),
            "candidate_count": len(candidates),
            "technique_layers": layers,
        })
    for layer in sorted(all_layers):
        entries.append({
            "ledger_version": EXECUTION_LEDGER_VERSION,
            "stage": "technique_layer",
            "method": layer,
            "status": "executed",
            "source": "python-engine",
        })
    entries.extend((
        {
            "ledger_version": EXECUTION_LEDGER_VERSION,
            "stage": "candidate_ranking",
            "status": "executed" if candidate_decisions else "not_executed",
            "source": "python-decision-policy",
            "candidate_count": len(candidate_decisions),
        },
        {
            "ledger_version": EXECUTION_LEDGER_VERSION,
            "stage": "diagnostics",
            "status": "executed" if diagnostics else "not_executed",
            "source": "python-engine",
            "metrics": sorted(diagnostics),
        },
    ))
    return entries