routesapi/app/services/routing/road_sequencing_agent.py

"""
Road-Sequencing Decision Agent
==============================

Autonomous controller for the `routing_use_road_distance` feature. Instead of a
human flipping a config flag, this agent periodically *measures* whether road-aware
sequencing (road travel-time matrix + OR-Tools open-TSP) actually beats the default
straight-line ordering on REAL recent batches, and turns the feature on or off by
itself — re-checking every cycle so it self-corrects if the gain ever disappears.

Decision (with hysteresis, so it doesn't flap):
    mean travel-time gain >= routing_auto_enable_gain_pct  -> enable
    mean travel-time gain <  routing_auto_disable_gain_pct -> disable
    in between                                             -> leave as-is

Every decision is stored (auditable) in DynamicConfig under `routing_road_eval`
and exposed via GET /api/v1/ml/road-eval. Cost is bounded: it evaluates only
`routing_eval_sample_batches` batches once per `routing_eval_interval_hours`.
"""

import asyncio
import json
import logging
import threading
import time
from datetime import datetime
from typing import Any, Dict, List, Optional

import numpy as np

logger = logging.getLogger(__name__)


def _total(order_idx: List[int], matrix: List[List[float]]) -> float:
    """Total open-route cost: origin(0) -> drops in `order_idx` (1-based into matrix)."""
    seq = [0] + [i + 1 for i in order_idx]
    return sum(matrix[seq[k]][seq[k + 1]] for k in range(len(seq) - 1))


class RoadSequencingAgent:
    """Measures road-vs-aerial sequencing on real batches and auto-toggles the flag."""

    def __init__(self):
        self._opt = None  # lazy RouteOptimizer
        self._scheduler_started = False
        self._lock = threading.Lock()
        self.last_decision: Dict[str, Any] = {}

    def _optimizer(self):
        if self._opt is None:
            from app.services.routing.route_optimizer import RouteOptimizer
            self._opt = RouteOptimizer()
        return self._opt

    async def evaluate(self, sample_batches: int = 8, days: int = 14) -> Dict[str, Any]:
        """
        Compare aerial-order vs road-order travel time on sampled real batches.
        Returns mean gain % and per-batch detail. No flag changes here.
        """
        from app.services.routing.delivery_history_service import get_delivery_history_service
        from app.core.arrow_utils import calculate_haversine_matrix_vectorized

        opt = self._optimizer()
        if not opt.use_road_matrix:
            return {"evaluated": 0, "reason": "no_matrix_backend", "mean_gain_pct": 0.0}

        batches = get_delivery_history_service().sample_batches(
            days=days, limit=sample_batches
        )
        if not batches:
            return {"evaluated": 0, "reason": "no_local_batches", "mean_gain_pct": 0.0}

        per_batch: List[Dict[str, Any]] = []
        for b in batches:
            origin, drops = b["origin"], b["drops"]
            locs = [origin] + drops
            matrix = await opt._road_duration_matrix(locs)
            if matrix is None:
                continue
            # aerial order (current production behaviour)
            lats = np.array([p[0] for p in locs])
            lons = np.array([p[1] for p in locs])
            aerial = calculate_haversine_matrix_vectorized(lats, lons)
            aer = [i - 1 for i in opt._two_opt_improve(opt._solve_greedy(locs, aerial), aerial) if i != 0]
            # road order (OR-Tools open-TSP on the road travel-time matrix)
            road = [i - 1 for i in opt._solve_tsp_ortools(locs, matrix) if i != 0]

            # Human's ACTUAL route: drops are already in delivered order
            # (_load_raw_rows is ORDER BY deliverytime), so identity = what the
            # rider really drove. This is the "do we beat the humans?" baseline.
            human = list(range(len(drops)))

            t_aer = _total(aer, matrix)
            t_road = _total(road, matrix)
            t_human = _total(human, matrix)
            gain = (100.0 * (t_aer - t_road) / t_aer) if t_aer > 0 else 0.0
            gain_vs_human = (100.0 * (t_human - t_road) / t_human) if t_human > 0 else 0.0
            per_batch.append({
                "rider": b["rider"], "day": b["day"], "drops": len(drops),
                "aerial_min": round(t_aer, 1), "road_min": round(t_road, 1),
                "human_min": round(t_human, 1),
                "gain_pct": round(gain, 1),
                "gain_vs_human_pct": round(gain_vs_human, 1),
                "beat_human": t_road <= t_human + 1e-9,
            })

        if not per_batch:
            return {"evaluated": 0, "reason": "matrix_unavailable", "mean_gain_pct": 0.0}

        n = len(per_batch)
        mean_gain = sum(x["gain_pct"] for x in per_batch) / n
        mean_vs_human = sum(x["gain_vs_human_pct"] for x in per_batch) / n
        beats = sum(1 for x in per_batch if x["beat_human"])
        return {
            "evaluated": n,
            "mean_gain_pct": round(mean_gain, 2),
            "median_gain_pct": round(sorted(x["gain_pct"] for x in per_batch)[n // 2], 2),
            "human_beat_rate_pct": round(100.0 * beats / n, 1),
            "human_beat_count": f"{beats}/{n}",
            "mean_gain_vs_human_pct": round(mean_vs_human, 2),
            "per_batch": per_batch,
        }

    def decide_and_apply(self) -> Dict[str, Any]:
        """Run an evaluation and autonomously enable/disable road sequencing."""
        from app.config.dynamic_config import get_config
        cfg = get_config()

        sample = int(cfg.get("routing_eval_sample_batches", 8))
        days = int(cfg.get("routing_eval_days", 14))
        enable_thr = float(cfg.get("routing_auto_enable_gain_pct", 3.0))
        disable_thr = float(cfg.get("routing_auto_disable_gain_pct", 1.0))
        min_batches = int(cfg.get("routing_eval_min_batches", 3))
        auto_manage = bool(cfg.get("routing_auto_manage", True))

        try:
            result = asyncio.run(self.evaluate(sample_batches=sample, days=days))
        except RuntimeError:
            # An event loop is already running in this thread — use a fresh one.
            loop = asyncio.new_event_loop()
            try:
                result = loop.run_until_complete(self.evaluate(sample_batches=sample, days=days))
            finally:
                loop.close()

        current = bool(cfg.get("routing_use_road_distance", False))
        evaluated = result.get("evaluated", 0)
        gain = result.get("mean_gain_pct", 0.0)
        action = "kept"
        new_state = current

        if not auto_manage:
            action = "auto_manage_off"
        elif evaluated < min_batches:
            action = "insufficient_data"
        else:
            if gain >= enable_thr and not current:
                new_state = True
                cfg.set("routing_use_road_distance", True, source="road_agent")
                action = "enabled"
            elif gain < disable_thr and current:
                new_state = False
                cfg.set("routing_use_road_distance", False, source="road_agent")
                action = "disabled"

        decision = {
            "decided_at": datetime.utcnow().isoformat(),
            "action": action,
            "flag_before": current,
            "flag_after": new_state,
            "mean_gain_pct": gain,
            "enable_threshold_pct": enable_thr,
            "disable_threshold_pct": disable_thr,
            "evaluation": result,
        }
        self.last_decision = decision
        try:
            # Persist a compact copy for audit (without the full per-batch list).
            compact = {k: v for k, v in decision.items() if k != "evaluation"}
            compact["evaluated"] = evaluated
            compact["human_beat_rate_pct"] = result.get("human_beat_rate_pct")
            compact["mean_gain_vs_human_pct"] = result.get("mean_gain_vs_human_pct")
            cfg.set("routing_road_eval", compact, source="road_agent")
        except Exception:
            pass

        logger.info(
            f"[RoadAgent] action={action} gain_vs_aerial={gain}% evaluated={evaluated} "
            f"flag {current}->{new_state}"
        )
        if evaluated:
            logger.info(
                f"[RoadAgent] beat humans on {result.get('human_beat_count', '?')} batches "
                f"(avg {result.get('mean_gain_vs_human_pct', 0)}% faster than actual delivered order)"
            )
        return decision

    def ensure_background_agent(self, interval_hours: int = 24,
                               warmup_seconds: int = 120) -> bool:
        """Start the autonomous decision loop once (daemon thread)."""
        with self._lock:
            if self._scheduler_started:
                return False
            self._scheduler_started = True

        def _loop():
            # Let the ETA sync agent populate the local mirror first.
            time.sleep(max(0, warmup_seconds))
            logger.info(f"[RoadAgent] autonomous road-sequencing decision loop started "
                        f"(interval={interval_hours}h)")
            while True:
                try:
                    self.decide_and_apply()
                except Exception as e:
                    logger.warning(f"[RoadAgent] decision cycle failed (will retry): {e}")
                from app.config.dynamic_config import get_config
                hrs = int(get_config().get("routing_eval_interval_hours", interval_hours))
                time.sleep(max(1, hrs) * 3600)

        threading.Thread(target=_loop, daemon=True, name="road-seq-agent").start()
        return True


_agent: Optional[RoadSequencingAgent] = None
_agent_lock = threading.Lock()


def get_road_agent() -> RoadSequencingAgent:
    global _agent
    with _agent_lock:
        if _agent is None:
            _agent = RoadSequencingAgent()
        return _agent