"""Connection-risk derivation for the flight-assist precheck.

Where `wake_rules.py` detects events from single-flight snapshot deltas
and `phase_markers.py` detects them from wall-clock time, this module
detects them from CROSS-flight context — leg-1's live arrival vs
leg-2's scheduled departure on the same trip.

The capability (README #4): for multi-leg itineraries with tight
transfer windows, alert when leg-1 is running late so a rebook
decision is possible BEFORE leg-1 takes off (when the user still has
agency on the ground).

Fires once per leg-2 flight (the `connection_at_risk_fired` phase
marker on leg-2's state). The caller keeps a wake pending until its
successful send is acknowledged. After confirmed delivery, the marker
suppresses later alerts even if the delay magnitude grows further.

Inputs are loaded by the precheck from existing on-disk per-flight
state — no fresh byAir call is needed here, because the cadence ladder
already keeps each leg's snapshot at the appropriate freshness. The
per-flight `trip_id` stored at sync time is the grouping key.

Pure functions: no I/O, no state mutation. The caller (precheck.py)
sets the marker flag only from a confirmed receipt. A wake alone
does not consume this gate.

stdlib-only: `datetime` per `coding-policy: dependency-management`.
"""

from __future__ import annotations

from datetime import UTC, datetime, timedelta

DEFAULT_MIN_TRANSFER_MINUTES = 45

# Don't fire farther out than this from leg-1's scheduled departure —
# delay projections that far in advance are typically speculative, and
# a rebook decision needs day-of (or eve-of) certainty to be useful.
_LEG1_LOOKAHEAD_HOURS = 24

# Status values at which leg-1 connection-risk is no longer actionable.
# `landed`: the user already did or did not make the connection (the
# answer is observable). `cancelled` / `diverted`: a more specific
# alert path fires via wake_rules.
_LEG1_TERMINAL_STATUSES = frozenset({"landed", "cancelled", "diverted"})


def detect_connection_risks(
    *,
    flight_states: list[dict],
    now_utc: datetime,
    min_transfer_minutes: int = DEFAULT_MIN_TRANSFER_MINUTES,
) -> list[tuple[int, dict]]:
    """Return per-leg-2 connection-risk events for tight transfers.

    Returns a list of `(leg2_flight_id, event_dict)` tuples. The caller
    (precheck.py) queues these notices, then copies confirmed receipts into
    each leg-2 record's `phase_markers["connection_at_risk_fired"]`.
    Emitting a wake leaves that gate unconsumed.

    `flight_states` is the full on-disk list of per-flight records
    (each matches the shape in state-schema.md). The function groups by
    `trip_id`, sorts each group by `scheduled_dep_time`, and walks
    consecutive pairs where leg-1's arrival airport equals leg-2's
    departure airport.

    A pair fires `connection_at_risk` when ALL of:

    - Leg-1's scheduled departure is within `_LEG1_LOOKAHEAD_HOURS` of `now_utc`
    - Leg-1 status is not in {landed, cancelled, diverted}
    - Leg-2's `connection_at_risk_fired` marker is False
    - The projected transfer window is less than `min_transfer_minutes`

    The projected transfer window is computed as:

        scheduled_dep(leg-2) - projected_arr(leg-1)

    where `projected_arr(leg-1)` comes from leg-1's live `arr_time`
    snapshot when present, otherwise from `scheduled_arr_time` (no live
    update yet means scheduled is the best estimate).

    Raises ValueError if `min_transfer_minutes` is not a plain non-negative
    int — `bool` is rejected (it's a subclass of `int` in Python so
    `True` would silently behave as `1 min`), matching the validator
    discipline applied to schema_version / flight_id / config fields
    elsewhere in this plugin.
    """
    if (
        not isinstance(min_transfer_minutes, int)
        or isinstance(min_transfer_minutes, bool)
        or min_transfer_minutes < 0
    ):
        raise ValueError(
            f"detect_connection_risks: min_transfer_minutes must be a "
            f"non-negative int, got {type(min_transfer_minutes).__name__} "
            f"{min_transfer_minutes!r}"
        )
    events: list[tuple[int, dict]] = []
    for trip_group in _group_by_trip(flight_states):
        events.extend(
            _detect_for_trip(
                trip_group=trip_group,
                now_utc=now_utc,
                min_transfer_minutes=min_transfer_minutes,
            )
        )
    return events


def _group_by_trip(flight_states: list[dict]) -> list[list[dict]]:
    """Group flight states by `trip_id`, sorted by scheduled_dep_time inside each group.

    Single-leg trips (one flight per trip_id) are excluded since they
    have no connection to evaluate. Flights with trip_id == 0 (the sync
    fallback) or non-int trip_id are skipped. Sort key is the parsed
    UTC datetime — raw RFC3339 string ordering breaks for multi-timezone
    itineraries (a leg in JST "2026-05-17T08:00:00+09:00" sorts before
    a leg in PDT "2026-05-17T10:00:00-07:00" by string, but the JST leg
    is actually 8h earlier in UTC). Legs missing or with unparseable
    scheduled_dep_time sort last via a sentinel max-future.
    """
    by_trip: dict[int, list[dict]] = {}
    for state in flight_states:
        trip_id = state.get("trip_id")
        if not isinstance(trip_id, int) or isinstance(trip_id, bool) or trip_id == 0:
            continue
        by_trip.setdefault(trip_id, []).append(state)
    groups: list[list[dict]] = []
    sentinel = datetime.max.replace(tzinfo=UTC)
    for trip_id in sorted(by_trip):
        legs = by_trip[trip_id]
        if len(legs) < 2:
            continue
        legs_sorted = sorted(
            legs, key=lambda s: _parse_iso8601(s.get("scheduled_dep_time")) or sentinel
        )
        groups.append(legs_sorted)
    return groups


def _detect_for_trip(
    *,
    trip_group: list[dict],
    now_utc: datetime,
    min_transfer_minutes: int,
) -> list[tuple[int, dict]]:
    """Walk one trip's sorted legs, emit events for tight connections."""
    events: list[tuple[int, dict]] = []
    for leg1, leg2 in zip(trip_group, trip_group[1:], strict=False):
        event = _evaluate_pair(
            leg1=leg1,
            leg2=leg2,
            now_utc=now_utc,
            min_transfer_minutes=min_transfer_minutes,
        )
        if event is not None:
            events.append((leg2["flight_id"], event))
    return events


def _evaluate_pair(
    *,
    leg1: dict,
    leg2: dict,
    now_utc: datetime,
    min_transfer_minutes: int,
) -> dict | None:
    """Return a connection_at_risk event for the (leg1, leg2) pair, or None."""
    if leg1.get("arr_airport_id") != leg2.get("dep_airport_id"):
        return None

    leg2_markers = leg2.get("phase_markers") or {}
    if leg2_markers.get("connection_at_risk_fired"):
        return None

    leg1_snapshot = leg1.get("last_snapshot") or {}
    leg1_status = leg1_snapshot.get("computed_status")
    if leg1_status in _LEG1_TERMINAL_STATUSES:
        return None

    leg1_scheduled_dep = _parse_iso8601(leg1.get("scheduled_dep_time"))
    if leg1_scheduled_dep is None:
        return None
    if leg1_scheduled_dep - now_utc > timedelta(hours=_LEG1_LOOKAHEAD_HOURS):
        return None

    projected_arr_str = leg1_snapshot.get("arr_time") or leg1.get("scheduled_arr_time")
    leg2_dep_str = leg2.get("scheduled_dep_time")
    if not projected_arr_str or not leg2_dep_str:
        return None

    projected_arr = _parse_iso8601(projected_arr_str)
    leg2_dep = _parse_iso8601(leg2_dep_str)
    if projected_arr is None or leg2_dep is None:
        return None

    transfer_minutes = int((leg2_dep - projected_arr).total_seconds() // 60)
    if transfer_minutes >= min_transfer_minutes:
        return None

    scheduled_arr = _parse_iso8601(leg1.get("scheduled_arr_time"))
    scheduled_layover_minutes: int | None
    if scheduled_arr is not None:
        scheduled_layover_minutes = int((leg2_dep - scheduled_arr).total_seconds() // 60)
    else:
        scheduled_layover_minutes = None

    return {
        "reason": "connection_at_risk",
        "leg1_code": leg1.get("code") or leg1_snapshot.get("code"),
        "leg2_code": leg2.get("code") or (leg2.get("last_snapshot") or {}).get("code"),
        "leg1_flight_id": leg1.get("flight_id"),
        "connecting_airport_id": leg2.get("dep_airport_id"),
        # transfer_minutes_remaining can be <= 0 when projected leg-1 arrival
        # is AT or AFTER leg-2 scheduled departure (severe delay; the
        # connection is structurally lost). missed_connection flags that
        # case so SKILL.md can compose a "rebook required" notification
        # instead of rendering "boards in -15 min".
        "transfer_minutes_remaining": transfer_minutes,
        "missed_connection": transfer_minutes <= 0,
        "scheduled_layover_minutes": scheduled_layover_minutes,
        "min_transfer_minutes": min_transfer_minutes,
        "projected_leg1_arr_time": projected_arr_str,
        "scheduled_leg2_dep_time": leg2_dep_str,
    }


def _parse_iso8601(value: str | None) -> datetime | None:
    if not value:
        return None
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=UTC)
    return parsed