score.py

"""Priority scoring + deadline ranking — the deterministic core.

Two scores compete; the higher one is displayed:

    do_score   = (I * U * R_eff * QW) / 10
    prep_score = (I * U * (10 - R)) / 10 * 0.7 * QW

with U = max(calendar_urgency, completion_urgency), a quick-win boost QW, and a
partial readiness lift R_eff. All inputs are coerced to finite numbers and
clamped to their domains, so the score can never be NaN/Infinity. Standard
library only.
"""

from __future__ import annotations

import math
from datetime import datetime

# -- Scoring knobs ------------------------------------------------------------
READY_LIFT = 0.7   # how far urgency lifts an unready task toward its target (<1)
PREP_BIAS = 0.7    # action bias on the prep branch (<1)
SCORE_CAP = 1000   # backstop guard, above the formula's natural max
TIE_BAND = 0.10    # value-closeness band (log-bucketed for transitivity)


def clamp(x: float, lo: float, hi: float) -> float:
    return max(lo, min(hi, x))


def finite(x, fallback: float) -> float:
    """Coerce non-numeric / non-finite input to a default before the formula."""
    try:
        xf = float(x)
    except (TypeError, ValueError):
        return fallback
    return xf if math.isfinite(xf) else fallback


def readiness_of(t: dict) -> float:
    return finite(t.get("readiness"), 8)  # default: presumed mostly-ready


# -- Urgency curves -----------------------------------------------------------

def cal_urgency(d: float) -> float:
    if d <= 0:
        return min(40, 30 + (-d) * 2)
    base = 1 + 9 / (1 + (d / 7) ** 1.5)
    T, k, eps = 2, 8, 0.25
    accel = max(0, k / (d + eps) - k / (T + eps)) if d < T else 0
    return min(30, base + accel)


def comp_urgency(d: float, e_hours: float) -> float:
    if d <= 0:
        return min(40, 30 + (-d) * 2)
    e_days = max(e_hours, 0.05) / 4
    slack = d / e_days
    base = 1 + 9 / (1 + slack / 2)
    T, k, eps = 1, 4, 0.1
    accel = max(0, k / (slack + eps) - k / (T + eps)) if slack < T else 0
    return min(30, base + accel)


# -- Core score ---------------------------------------------------------------

def score_components(t: dict) -> dict:
    I = clamp(finite(t.get("impact"), 5), 1, 10)
    R = clamp(readiness_of(t), 1, 10)
    E = max(finite(t.get("effort_hours"), 0.05), 0.05)
    d = days_to_due_raw(t.get("due_date"))
    has_due = bool(t.get("due_date")) and math.isfinite(d)
    U_cal = cal_urgency(d) if has_due else 1.5
    U_comp = comp_urgency(d, E) if has_due else 1.5
    U = max(U_cal, U_comp)
    if has_due and d > 0:
        slack = d / (E / 4)
    elif d <= 0:
        slack = 0.0
    else:
        slack = math.inf
    QW = 1 + 0.6 * (I / 10) * math.exp(-E / 1.5)
    risk_on = bool(t.get("execute_anyway"))
    R_floor = clamp((U - 6) / 4 * 10, 0, 10)
    R_eff = max(R, 9) if risk_on else R + max(0, R_floor - R) * READY_LIFT
    do_score = clamp((I * U * R_eff * QW) / 10, 0.1, SCORE_CAP)
    prep_score = 0.0 if risk_on else clamp((I * U * (10 - R)) / 10 * PREP_BIAS * QW, 0, SCORE_CAP)
    defer = (not risk_on) and has_due and U_comp >= 9 and slack < 0.5 and I < 7
    display = max(do_score, prep_score)
    prep_wins = prep_score > do_score and R <= 5 and not risk_on
    return {
        "I": I, "R": R, "E": E, "d": d, "U": U, "U_cal": U_cal, "U_comp": U_comp,
        "slack": slack, "qw": QW, "R_floor": R_floor, "R_eff": R_eff,
        "do_score": do_score, "prep_score": prep_score, "defer": defer,
        "display": display, "prep_wins": prep_wins,
    }


def priority(t: dict) -> float:
    return score_components(t)["display"]


def value_bucket(v: float) -> int:
    return math.floor(math.log(max(v, 0.1)) / math.log(1 + TIE_BAND))


# -- Dates --------------------------------------------------------------------

def normalize_iso(due_date):
    """A bare date ('YYYY-MM-DD') is treated as 5pm local; a datetime passes through."""
    if not isinstance(due_date, str):
        return due_date
    return f"{due_date}T17:00:00" if len(due_date) == 10 else due_date


def _parse(due_date):
    try:
        return datetime.fromisoformat(normalize_iso(due_date))
    except (TypeError, ValueError):
        return None


def days_to_due_raw(due_date, now: datetime | None = None) -> float:
    if not due_date:
        return math.inf
    due = _parse(due_date)
    if due is None:
        return math.inf
    now = now or datetime.now()
    return (due - now).total_seconds() / 86400


def deadline_status(task: dict, now: datetime | None = None):
    """Discrete deadline tier: 'overdue' | 'today' | 'tomorrow' | None."""
    if not task or task.get("completed") or not task.get("due_date"):
        return None
    due = _parse(task["due_date"])
    if due is None:
        return None
    now = now or datetime.now()
    if due < now:
        return "overdue"
    start = lambda d: datetime(d.year, d.month, d.day)
    day_diff = round((start(due) - start(now)).total_seconds() / 86400)
    if day_diff == 0:
        return "today"
    if day_diff == 1:
        return "tomorrow"
    return None


# -- Ranking ------------------------------------------------------------------

def deadline_risk_map(sorted_active: list, now: datetime | None = None) -> dict:
    """Walk the queue in priority order, accumulate effort, and flag any task
    whose projected wall-clock finish lands past its deadline. Returns
    {id: 'at-risk' | 'tight'}."""
    out = {}
    now = now or datetime.now()
    acc = 0.0  # cumulative effort-hours so far, inclusive
    for t in sorted_active:
        if not t or t.get("completed"):
            continue
        E = max(t.get("effort_hours") or 0, 0)
        acc += E
        if not t.get("due_date"):
            continue
        due = _parse(t["due_date"])
        if due is None:
            continue
        hours_to_due = (due - now).total_seconds() / 3600
        if not (hours_to_due > 0):
            continue  # overdue — deadline_status owns that signal
        slack = hours_to_due - acc
        if slack <= 0:
            out[t["id"]] = "at-risk"
        elif slack < 0.5 * E:
            out[t["id"]] = "tight"
    return out


def rank_active(tasks: list, now: datetime | None = None) -> list:
    """Rank with deadlines as a constraint, not a blended term. Three tiers:
    0 overdue (EDF), 1 binding/at-risk (EDF, lifted above the value pack),
    2 value pack (value bucket, then sooner due breaks near-ties)."""
    now = now or datetime.now()
    active = [t for t in tasks if not t.get("completed")]
    V = {t["id"]: priority(t) for t in active}
    D = {t["id"]: days_to_due_raw(t.get("due_date"), now) for t in active}
    by_value = sorted(active, key=lambda t: V[t["id"]], reverse=True)
    risk = deadline_risk_map(by_value, now)

    def tier_of(t):
        if t.get("due_date") and D[t["id"]] <= 0:
            return 0
        return 1 if risk.get(t["id"]) == "at-risk" else 2

    def sort_key(t):
        tid = t["id"]
        tr = tier_of(t)
        if tr < 2:
            return (tr, D[tid], 0.0)  # earliest-deadline-first
        return (tr, -value_bucket(V[tid]), D[tid])

    # Stable sort: equal keys keep by_value (value-desc) order.
    return sorted(by_value, key=sort_key)


# -- Display helpers ----------------------------------------------------------

def format_score(n: float) -> str:
    rounded = round(n * 10) / 10
    return str(int(rounded)) if rounded == int(rounded) else f"{rounded:.1f}"


def explain(t: dict, now: datetime | None = None) -> str:
    """A short, deterministic 'why it ranks here' phrase, read off the same
    components that drive the score. This replaces a model-written reason: it
    costs the model no tokens to produce and can never drift from the math the
    way an LLM sentence can."""
    now = now or datetime.now()
    c = score_components(t)
    I, R, E = c["I"], c["R"], c["E"]
    status = deadline_status(t, now)
    if status == "overdue":
        return "overdue — start now"
    if c["prep_wins"]:
        return ("high-value but not ready — de-risk first" if I >= 5
                else "not ready — needs a first step")
    if status in ("today", "tomorrow"):
        return f"due {status} — time-sensitive"
    if E <= 1 and R >= 7:
        return "quick win — short and ready"
    if R <= 3:
        return "blocked — next step unclear"
    if I >= 8:
        return "high-impact work"
    if I <= 3:
        return "minor — low impact"
    return "steady priority"


def due_label(due_date, now: datetime | None = None) -> str:
    """Compact relative deadline label ('now', '3h', 'tomorrow', '2d', '5d late')."""
    if not due_date:
        return "-"
    due = _parse(due_date)
    if due is None:
        return "-"
    now = now or datetime.now()
    raw = (due - now).total_seconds() / 86400
    if raw < 0:
        ab = -raw
        return f"{max(1, round(ab * 24))}h late" if ab < 1 else f"{round(ab)}d late"
    start = lambda d: datetime(d.year, d.month, d.day)
    day_diff = round((start(due) - start(now)).total_seconds() / 86400)
    if day_diff == 0:
        return "now" if raw < 1 / 24 else f"{round(raw * 24)}h"
    if day_diff == 1:
        return "tomorrow"
    return f"{day_diff}d"