#!/usr/bin/env python3
from __future__ import annotations
"""
CodexBar ESP32 companion proxy.

Reads OAuth tokens from the local keychain / auth files and proxies usage data
from Anthropic (Claude) and optionally OpenAI (Codex) to the ESP32.

Usage:
    python server.py

The ESP32 polls:  GET http://<mac-ip>:7842/usage
"""

import json
import os
import subprocess
import threading
import time
import urllib.request
import urllib.error
from urllib.parse import quote, urlparse
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

try:
    from companion import pace  # imported as `companion.server` (tests, smoke_test_hooks)
except ImportError:
    import pace  # run directly: `python3 companion/server.py` (companion/ is sys.path[0])

PORT = 7842
CACHE_TTL_SEC = 30

# ── Attention state ───────────────────────────────────────────────────────

NTFY_TOPIC             = os.environ.get("CODEXBAR_NTFY_TOPIC", "codingAgentWaitingOnYou").strip()
NTFY_USAGE_RESET_TOPIC = os.environ.get("CODEXBAR_USAGE_RESET_NTFY_TOPIC", "codexbarUsageResets").strip()
NTFY_SSH_HOST          = os.environ.get("CODEXBAR_SSH_HOST", "").strip()
NTFY_SSH_USER          = os.environ.get("CODEXBAR_SSH_USER", os.environ.get("USER", "")).strip()
NTFY_NOTIFY_DONE       = os.environ.get("CODEXBAR_NOTIFY_DONE", "1").strip().lower() not in ("0", "false", "no", "off")
QUESTION_HEURISTIC_ENABLED = os.environ.get("CODEXBAR_QUESTION_HEURISTIC", "1").strip().lower() \
    not in ("0", "false", "no", "off")

DISPLAY_ATTENTION_PROVIDERS = ("Claude", "Codex")
NOTIFY_ONLY_PROVIDERS = ("LM Studio",)
ATTENTION_PROVIDERS = DISPLAY_ATTENTION_PROVIDERS + NOTIFY_ONLY_PROVIDERS

STATE_PRIORITY = {
    "needs_user": 1,
    "error": 2,
    "done": 3,
    "running": 4,
    "idle": 5,
    "unknown": 6,
}

EVENT_MAP = {
    "SessionStart": ("running", "session_start"),
    "PreToolUse": ("running", "tool_use"),
    "PostToolUse": ("running", "tool_use"),
    "PermissionRequest": ("needs_user", "permission_request"),
    "Notification": ("needs_user", "notification"),
    "Elicitation": ("needs_user", "elicitation"),
    "Stop": ("done", "stop"),
    "TaskComplete": ("done", "task_complete"),
    "UserPromptSubmit": ("running", "user_prompt"),
    "SessionEnd": ("idle", "session_end"),
}

ATTENTION_TIMEOUTS_SEC = {
    "needs_user": 3600,
    "error": 1800,
    "done": 300,
    "unknown": 1200,
    "idle": 0,
}
ATTENTION_REASON_TIMEOUTS_SEC = {
    # Notification hooks are useful as an attention nudge, but they are less
    # authoritative than an explicit permission/elicitation request.
    "notification": 300,
}
RUNNING_TO_UNKNOWN_SEC = 600
RUNNING_REMOVE_SEC = 1800

_attention_lock = threading.Lock()
_attention_records: dict[str, dict] = {}
_attention_seq = 0


QUESTION_PHRASES = (
    "which option",
    "which approach",
    "what would you like",
    "what do you want",
    "do you want me to",
    "would you like me to",
    "should i",
    "should we",
    "please confirm",
    "need your input",
    "waiting for your",
)

AUTO_APPROVAL_PHRASES = (
    "automatic approval",
    "auto-approved",
    "auto approved",
    "approved by auto",
)


def looks_like_auto_approval(text: str | None) -> bool:
    if not text:
        return False
    lowered = str(text).lower()
    return any(phrase in lowered for phrase in AUTO_APPROVAL_PHRASES)


def _normalize_provider(value: str) -> str | None:
    v = (value or "").strip().lower()
    if v == "claude":
        return "Claude"
    if v == "codex":
        return "Codex"
    if v in ("lmstudio", "lm studio", "lm-studio", "lm_studio"):
        return "LM Studio"
    return None


def _project_name_from_cwd(cwd: str) -> str:
    path = os.path.abspath(os.path.expanduser(cwd))
    name = os.path.basename(path.rstrip(os.sep))
    return name or path


def _attention_key(provider: str, session_id: str, cwd: str) -> str:
    return f"{provider}:{session_id or cwd}:{cwd}"


def _empty_attention(now: int | None = None) -> dict:
    ts = int(time.time()) if now is None else now
    return {
        "state": "idle",
        "count": 0,
        "primary_project": "",
        "projects": [],
        "reason": "",
        "updated_at": ts,
    }


def _invalidate_usage_cache() -> None:
    global _cached_response, _cache_expires_at
    with _cache_lock:
        _cached_response = None
        _cache_expires_at = 0


def _build_ssh_action(label: str = "Open Terminal") -> str | None:
    """Return an ntfy Actions header value that opens an SSH connection, or None if not configured."""
    if not NTFY_SSH_HOST or not NTFY_SSH_USER:
        return None
    return f"view, {label}, ssh://{NTFY_SSH_USER}@{NTFY_SSH_HOST}"


def _send_ntfy_async(provider: str, project_name: str, reason: str, *, kind: str = "needs_user") -> None:
    if not NTFY_TOPIC:
        return

    def worker() -> None:
        try:
            if kind == "done":
                message = f"{provider} finished: {project_name}"
                headers: dict[str, str] = {
                    "Title": "Agent task complete",
                    "Priority": "low",
                    "Tags": "white_check_mark",
                }
            else:
                message = f"{provider} waiting: {project_name}"
                if reason:
                    message += f" ({reason})"
                headers = {
                    "Title": "Agent waiting for you",
                    "Priority": "high",
                    "Tags": "bell",
                }
                action = _build_ssh_action()
                if action:
                    headers["Actions"] = action
            req = urllib.request.Request(
                f"https://ntfy.sh/{quote(NTFY_TOPIC, safe='')}",
                data=message.encode(),
                headers=headers,
                method="POST",
            )
            with urllib.request.urlopen(req, timeout=5):
                pass
        except Exception as e:
            print(f"[ntfy] notification failed: {e}")

    threading.Thread(target=worker, daemon=True).start()


def looks_like_user_question(text: str | None) -> bool:
    if not text:
        return False
    lowered = " ".join(str(text).lower().split())
    if any(phrase in lowered for phrase in QUESTION_PHRASES):
        return True
    if "?" in lowered:
        return True
    lines = [line.strip().lower() for line in str(text).splitlines() if line.strip()]
    choice_lines = 0
    for line in lines[-8:]:
        if len(line) >= 3 and (
            (line[0].isdigit() and line[1] in (".", ")"))
            or (line[0] in "abcdefghijklmnopqrstuvwxyz" and line[1] in (".", ")"))
            or line.startswith("- ")
        ):
            choice_lines += 1
    return choice_lines >= 2


def _expire_attention_locked(now: int) -> None:
    remove_keys = []
    for key, rec in _attention_records.items():
        state = rec.get("state", "unknown")
        updated_at = int(rec.get("updated_at", now))
        age = now - updated_at

        if state == "running":
            if age >= RUNNING_REMOVE_SEC:
                remove_keys.append(key)
            elif age >= RUNNING_TO_UNKNOWN_SEC:
                rec["state"] = "unknown"
                rec["reason"] = "stale_running"
                rec.setdefault("stale_at", updated_at + RUNNING_TO_UNKNOWN_SEC)
            continue

        if state == "unknown":
            stale_at = int(rec.get("stale_at", updated_at))
            if now - stale_at >= ATTENTION_TIMEOUTS_SEC["unknown"]:
                remove_keys.append(key)
            continue

        timeout = ATTENTION_REASON_TIMEOUTS_SEC.get(rec.get("reason", ""))
        if timeout is None:
            timeout = ATTENTION_TIMEOUTS_SEC.get(state)
        if timeout is not None and age >= timeout:
            remove_keys.append(key)

    for key in remove_keys:
        _attention_records.pop(key, None)


def ingest_attention_event(payload: dict) -> tuple[bool, dict]:
    global _attention_seq
    now = int(time.time())
    provider = _normalize_provider(str(payload.get("provider", "")))
    if provider is None:
        return False, {"ok": False, "error": "unknown provider"}

    event = str(payload.get("event", "")).strip()
    if event not in EVENT_MAP:
        return False, {"ok": False, "error": "unknown event"}

    cwd = str(payload.get("cwd", "")).strip()
    if not cwd:
        return False, {"ok": False, "error": "missing cwd"}

    session_id = str(payload.get("session_id", "")).strip() or f"{provider}:{cwd}"
    state, reason = EVENT_MAP[event]
    if event == "Notification" and looks_like_auto_approval(payload.get("message")):
        state, reason = "running", "tool_use"
    elif event == "Stop" and QUESTION_HEURISTIC_ENABLED and looks_like_user_question(payload.get("last_assistant_message")):
        state, reason = "needs_user", "possible_question"
    project_name = _project_name_from_cwd(cwd)
    key = _attention_key(provider, session_id, cwd)
    should_notify = False
    should_notify_done = False

    with _attention_lock:
        _expire_attention_locked(now)
        _attention_seq += 1
        seq = _attention_seq

        if event == "SessionStart":
            for old_key, rec in list(_attention_records.items()):
                if rec.get("provider") == provider and rec.get("cwd") == cwd:
                    _attention_records.pop(old_key, None)

        prev_state = _attention_records.get(key, {}).get("state")
        if state == "idle":
            _attention_records.pop(key, None)
        else:
            _attention_records[key] = {
                "provider": provider,
                "session_id": session_id,
                "cwd": cwd,
                "project_name": project_name,
                "state": state,
                "reason": reason,
                "updated_at": now,
                "seq": seq,
                "last_running_at": now if state == "running" else _attention_records.get(key, {}).get("last_running_at"),
            }
            # permission_request is handled by the agent's own approval UI (e.g. guardian_subagent);
            # paging the user is redundant and fires for every auto-approved tool use.
            should_notify = (
                state == "needs_user"
                and prev_state != "needs_user"
                and reason != "permission_request"
            )
            # done notification fires when agent finishes a task (was actively running/waiting).
            should_notify_done = (
                NTFY_NOTIFY_DONE
                and state == "done"
                and (
                    prev_state in ("running", "needs_user", "unknown")
                    or provider in NOTIFY_ONLY_PROVIDERS
                )
            )

    _invalidate_usage_cache()

    if should_notify:
        _send_ntfy_async(provider, project_name, reason)
    if should_notify_done:
        _send_ntfy_async(provider, project_name, reason, kind="done")

    return True, {"ok": True}


def _dedupe_provider_records(records: list[dict]) -> list[dict]:
    by_cwd: dict[str, dict] = {}
    for rec in records:
        cwd = rec.get("cwd", "")
        cur = by_cwd.get(cwd)
        if cur is None:
            by_cwd[cwd] = rec
            continue
        cur_rank = (
            STATE_PRIORITY.get(cur.get("state", "unknown"), 99),
            -int(cur.get("updated_at", 0)),
            -int(cur.get("seq", 0)),
        )
        rec_rank = (
            STATE_PRIORITY.get(rec.get("state", "unknown"), 99),
            -int(rec.get("updated_at", 0)),
            -int(rec.get("seq", 0)),
        )
        if rec_rank < cur_rank:
            by_cwd[cwd] = rec
    return list(by_cwd.values())


def get_attention_by_provider() -> dict[str, dict]:
    now = int(time.time())
    with _attention_lock:
        _expire_attention_locked(now)
        snapshot = list(_attention_records.values())

    grouped: dict[str, list] = {provider: [] for provider in ATTENTION_PROVIDERS}
    for rec in snapshot:
        provider = rec.get("provider")
        if provider in grouped:
            grouped[provider].append(rec)

    result = {}
    for provider, records in grouped.items():
        records = _dedupe_provider_records(records)
        if not records:
            result[provider] = _empty_attention(now)
            continue

        best_state = min(
            (rec.get("state", "unknown") for rec in records),
            key=lambda s: STATE_PRIORITY.get(s, 99),
        )
        selected = [rec for rec in records if rec.get("state", "unknown") == best_state]
        selected.sort(
            key=lambda rec: (int(rec.get("updated_at", 0)), int(rec.get("seq", 0))),
            reverse=True,
        )
        primary = selected[0]
        projects = []
        for rec in selected:
            name = rec.get("project_name", "")
            if name and name not in projects:
                projects.append(name)
            if len(projects) >= 3:
                break

        result[provider] = {
            "state": best_state,
            "count": len(selected),
            "primary_project": primary.get("project_name", ""),
            "projects": projects,
            "reason": primary.get("reason", ""),
            "updated_at": int(primary.get("updated_at", now)),
        }

    return result


def get_status_payload() -> dict:
    now = int(time.time())
    attention = get_attention_by_provider()
    return {
        "ok": True,
        "updated_at": now,
        "providers": [
            {"name": "Claude", "attention": attention.get("Claude", _empty_attention(now))},
            {"name": "Codex", "attention": attention.get("Codex", _empty_attention(now))},
        ],
    }

# ── Claude ────────────────────────────────────────────────────────────────

ANTHROPIC_USAGE_URL = "https://api.anthropic.com/api/oauth/usage"
TOKEN_REFRESH_URL   = "https://platform.claude.com/v1/oauth/token"
OAUTH_CLIENT_ID     = "9d1c250a-e61b-44d9-88ed-5944d1962f5e"


def read_credentials() -> dict:
    """Read Claude CLI OAuth credentials from macOS keychain."""
    result = subprocess.run(
        ["security", "find-generic-password", "-s", "Claude Code-credentials", "-w"],
        capture_output=True, text=True, timeout=10,
    )
    if result.returncode != 0:
        raise RuntimeError(
            "Could not read Claude credentials from keychain.\n"
            "Make sure you're logged in: run `claude` and authenticate."
        )
    raw = result.stdout.strip()
    creds = json.loads(raw)
    return creds.get("claudeAiOauth", creds)


def get_access_token() -> tuple[str, str | None]:
    creds = read_credentials()
    return creds["accessToken"], creds.get("refreshToken")


def refresh_access_token(refresh_token: str) -> str:
    payload = json.dumps({
        "grant_type": "refresh_token",
        "refresh_token": refresh_token,
        "client_id": OAUTH_CLIENT_ID,
    }).encode()
    req = urllib.request.Request(
        TOKEN_REFRESH_URL,
        data=payload,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=15) as resp:
        data = json.loads(resp.read())
    return data["access_token"]


def fetch_claude_usage(access_token: str) -> dict:
    req = urllib.request.Request(
        ANTHROPIC_USAGE_URL,
        headers={
            "Authorization": f"Bearer {access_token}",
            "Accept": "application/json",
            "Content-Type": "application/json",
            "anthropic-beta": "oauth-2025-04-20",
            "User-Agent": "claude-code/2.1.0",
        },
    )
    with urllib.request.urlopen(req, timeout=15) as resp:
        return json.loads(resp.read())


def parse_resets_in_sec(resets_at_iso: str | None) -> int:
    if not resets_at_iso:
        return -1
    try:
        dt = datetime.fromisoformat(resets_at_iso.replace("Z", "+00:00"))
        delta = (dt - datetime.now(timezone.utc)).total_seconds()
        return max(0, int(delta))
    except Exception:
        return -1


def build_claude_windows(raw: dict) -> list:
    windows = []
    window_map = [
        ("five_hour",      "5h",   18000),
        ("seven_day",      "7d",   604800),
        ("seven_day_opus", "Opus", 604800),
    ]
    for key, label, duration_sec in window_map:
        w = raw.get(key)
        if w is None:
            continue
        utilization = w.get("utilization")
        # API returns utilization as a percent value (0–100+), not a 0–1 ratio.
        # Use floor so 0.6% → 0; also clamp 1% to 0 since it's indistinguishable
        # from noise at the start of a window and shows as a misleading redlined arc.
        pct = int(utilization) if utilization is not None else -1
        pct = min(pct, 100)
        if pct == 1:
            pct = 0
        windows.append({
            "label": label,
            "pct": pct,
            "resets_in_sec": parse_resets_in_sec(w.get("resets_at")),
            "duration_sec": duration_sec,
        })
    return windows


# ── Codex ─────────────────────────────────────────────────────────────────

CODEX_USAGE_URL   = "https://chatgpt.com/backend-api/wham/usage"
CODEX_REFRESH_URL = "https://auth.openai.com/oauth/token"
CODEX_CLIENT_ID   = "app_EMoamEEZ73f0CkXaXp7hrann"


def _codex_auth_path() -> str:
    codex_home = os.environ.get("CODEX_HOME", "").strip()
    if not codex_home:
        codex_home = os.path.expanduser("~/.codex")
    return os.path.join(codex_home, "auth.json")


def read_codex_credentials() -> dict | None:
    """Read Codex OAuth credentials from ~/.codex/auth.json. Returns None if unavailable."""
    path = _codex_auth_path()
    if not os.path.exists(path):
        return None
    try:
        with open(path) as f:
            data = json.load(f)
    except Exception:
        return None

    # Some configs store a bare OPENAI_API_KEY
    api_key = (data.get("OPENAI_API_KEY") or "").strip()
    if api_key:
        return {"access_token": api_key, "refresh_token": "", "account_id": None}

    tokens = data.get("tokens", {})
    access_token  = (tokens.get("access_token")  or tokens.get("accessToken",  "")).strip()
    refresh_token = (tokens.get("refresh_token") or tokens.get("refreshToken", "")).strip()
    account_id    =  tokens.get("account_id")    or tokens.get("accountId")

    return {"access_token": access_token, "refresh_token": refresh_token, "account_id": account_id} \
        if access_token else None


def refresh_codex_token(refresh_token: str) -> str:
    payload = json.dumps({
        "client_id": CODEX_CLIENT_ID,
        "grant_type": "refresh_token",
        "refresh_token": refresh_token,
        "scope": "openid profile email",
    }).encode()
    req = urllib.request.Request(
        CODEX_REFRESH_URL,
        data=payload,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=15) as resp:
        data = json.loads(resp.read())
    return data["access_token"]


def fetch_codex_usage(access_token: str, account_id: str | None = None) -> dict:
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Accept": "application/json",
        "User-Agent": "CodexBar",
    }
    if account_id:
        headers["ChatGPT-Account-Id"] = account_id
    req = urllib.request.Request(CODEX_USAGE_URL, headers=headers)
    with urllib.request.urlopen(req, timeout=15) as resp:
        return json.loads(resp.read())


def _window_label(limit_window_seconds: int) -> str:
    if limit_window_seconds == 18000:   return "5h"   # 5 hours
    if limit_window_seconds == 604800:  return "7d"   # 7 days
    if limit_window_seconds < 7200:     return f"{limit_window_seconds // 3600}h"
    return f"{limit_window_seconds // 86400}d"


def build_codex_windows(raw: dict) -> list:
    """Convert Codex API response to ESP32 window list."""
    rate_limit = raw.get("rate_limit") or {}
    windows = []
    for key in ("primary_window", "secondary_window"):
        w = rate_limit.get(key)
        if not w:
            continue
        pct = w.get("used_percent", -1)
        if pct < 0:
            continue
        pct = min(int(pct), 100)  # floor, matching Claude behavior
        if pct == 1:
            pct = 0
        reset_at = w.get("reset_at", 0)
        resets_in_sec = max(0, int(reset_at - time.time())) if reset_at else -1
        windows.append({
            "label": _window_label(w.get("limit_window_seconds", 0)),
            "pct": pct,
            "resets_in_sec": resets_in_sec,
            "duration_sec": w.get("limit_window_seconds", 0),
        })
    return windows


# ── Pace detection ────────────────────────────────────────────────────────

# Fire when the 7d window shows zero usage and at least this many seconds remain —
# "you haven't used anything all week and still have time to do background work."
UNDERPACE_PCT_THRESHOLD     = int(os.environ.get("CODEXBAR_UNDERPACE_PCT_THRESHOLD",     "25"))
UNDERPACE_MIN_REMAINING_SEC = int(os.environ.get("CODEXBAR_UNDERPACE_MIN_REMAINING_SEC", str(2 * 86400)))
UNDERPACE_COOLDOWN_SEC      = int(os.environ.get("CODEXBAR_UNDERPACE_COOLDOWN_SEC",      "1800"))
UNDERPACE_WINDOW            =     os.environ.get("CODEXBAR_UNDERPACE_WINDOW",            "7d").strip()

_underpace_lock = threading.Lock()
_underpace_last_fired: float = 0

USAGE_RESET_5H_REMAINING_THRESHOLD = int(os.environ.get("CODEXBAR_RESET_5H_REMAINING_THRESHOLD", "40"))
USAGE_RESET_SLEEP_START_HOUR       = int(os.environ.get("CODEXBAR_RESET_SLEEP_START_HOUR",       "23"))
USAGE_RESET_SLEEP_END_HOUR         = int(os.environ.get("CODEXBAR_RESET_SLEEP_END_HOUR",         "6"))
USAGE_RESET_LONG_MIN_DURATION_SEC  = int(os.environ.get("CODEXBAR_RESET_LONG_MIN_DURATION_SEC",  str(5 * 86400)))
USAGE_RESET_TIMER_JUMP_GRACE_SEC   = int(os.environ.get("CODEXBAR_RESET_TIMER_JUMP_GRACE_SEC",   "600"))

_usage_reset_lock = threading.Lock()
_usage_reset_windows: dict[tuple[str, str, int], dict] = {}


def _compute_pace(pct: int, resets_in_sec: int, duration_sec: int) -> float | None:
    if duration_sec <= 0 or resets_in_sec < 0:
        return None
    elapsed_frac = pace.elapsed_fraction(resets_in_sec, duration_sec)
    if elapsed_frac <= 0:
        return None
    return pace.pace_ratio(pct, elapsed_frac)


def _active_session_cwd() -> tuple[str, str]:
    """Return (cwd, project_name) of the most recently updated attention record, or ('', '')."""
    with _attention_lock:
        if not _attention_records:
            return "", ""
        rec = max(_attention_records.values(), key=lambda r: (int(r.get("updated_at", 0)), int(r.get("seq", 0))))
    return rec.get("cwd", ""), rec.get("project_name", "")


def _send_underpace_ntfy_async(resets_in_sec: int, project_name: str, pct: int = 0) -> None:
    if not NTFY_TOPIC:
        return

    def worker() -> None:
        try:
            days_left = resets_in_sec // 86400
            proj = f" in {project_name}" if project_name else ""
            message = f"7d window at {pct}%{proj}, {days_left}d remaining — good time for /smell-fix"
            req = urllib.request.Request(
                f"https://ntfy.sh/{quote(NTFY_TOPIC, safe='')}",
                data=message.encode(),
                headers={
                    "Title": "CodexBar: idle quota",
                    "Priority": "low",
                    "Tags": "hourglass_flowing_sand",
                },
                method="POST",
            )
            with urllib.request.urlopen(req, timeout=5):
                pass
        except Exception as e:
            print(f"[ntfy] underpace notification failed: {e}")

    threading.Thread(target=worker, daemon=True).start()


def _send_usage_reset_ntfy_async(
    provider: str,
    label: str,
    duration_sec: int,
    previous_remaining: int | None,
    early: bool = False,
) -> None:
    if not NTFY_USAGE_RESET_TOPIC:
        return

    def worker() -> None:
        try:
            window = label or _window_label(duration_sec)
            if early:
                message = f"{provider} {window} usage reset early surprise 🎁"
                title = "CodexBar: quota reset"
                tags = "tada,gift,partying_face"
                priority = "default"
            else:
                prev = f" You had {previous_remaining}% remaining." if previous_remaining is not None else ""
                message = f"{provider} {window} usage refreshed 🔄{prev}"
                title = "CodexBar: usage refreshed"
                tags = "arrows_counterclockwise"
                priority = "low"
            req = urllib.request.Request(
                f"https://ntfy.sh/{quote(NTFY_USAGE_RESET_TOPIC, safe='')}",
                data=message.encode(),
                headers={
                    "Title": title,
                    "Priority": priority,
                    "Tags": tags,
                },
                method="POST",
            )
            with urllib.request.urlopen(req, timeout=5):
                pass
        except Exception as e:
            print(f"[ntfy] reset notification failed: {e}")

    threading.Thread(target=worker, daemon=True).start()


def _is_sleeping_hour(now: datetime | None = None) -> bool:
    hour = (now or datetime.now()).hour
    start = USAGE_RESET_SLEEP_START_HOUR
    end = USAGE_RESET_SLEEP_END_HOUR
    if start == end:
        return True
    if start < end:
        return start <= hour < end
    return hour >= start or hour < end


def _is_five_hour_window(window: dict) -> bool:
    return window.get("duration_sec") == 18000 or window.get("label") == "5h"


def _is_long_reset_window(window: dict) -> bool:
    duration = int(window.get("duration_sec", 0) or 0)
    return duration >= USAGE_RESET_LONG_MIN_DURATION_SEC


def _window_reset_detected(previous: dict, current: dict) -> bool:
    prev_resets = int(previous.get("resets_in_sec", -1))
    cur_resets = int(current.get("resets_in_sec", -1))
    if cur_resets < 0:
        return False

    prev_pct = int(previous.get("pct", -1))
    cur_pct = int(current.get("pct", -1))
    if prev_pct < 0:
        return False
    if prev_resets >= 0 and cur_resets > prev_resets + USAGE_RESET_TIMER_JUMP_GRACE_SEC:
        return cur_pct <= prev_pct or cur_pct <= 5
    return prev_pct >= 50 and cur_pct <= 5 and cur_resets > USAGE_RESET_TIMER_JUMP_GRACE_SEC


def _is_early_reset(previous: dict) -> bool:
    prev_resets = int(previous.get("resets_in_sec", -1))
    return prev_resets > USAGE_RESET_TIMER_JUMP_GRACE_SEC


def _window_looks_fresh_after_reset(previous: dict, current: dict) -> bool:
    duration = int(current.get("duration_sec", 0) or 0)
    cur_resets = int(current.get("resets_in_sec", -1))
    if duration <= 0 or cur_resets < duration - USAGE_RESET_TIMER_JUMP_GRACE_SEC:
        return False

    prev_pct = int(previous.get("pct", -1))
    cur_pct = int(current.get("pct", -1))
    return prev_pct >= 0 and cur_pct >= 0 and (cur_pct <= prev_pct or cur_pct <= 5)


def _check_usage_resets(providers: list[dict]) -> None:
    notifications: list[tuple[str, str, int, int | None, bool]] = []
    now = datetime.now()

    with _usage_reset_lock:
        for prov in providers:
            provider_name = prov.get("name", "")
            provider_windows: list[tuple[dict, dict, bool]] = []
            for w in prov.get("windows", []):
                key = (provider_name, str(w.get("label", "")), int(w.get("duration_sec", 0) or 0))
                previous = _usage_reset_windows.get(key)
                _usage_reset_windows[key] = {
                    "pct": int(w.get("pct", -1)),
                    "resets_in_sec": int(w.get("resets_in_sec", -1)),
                }
                if previous is None:
                    continue
                provider_windows.append((previous, w, _window_reset_detected(previous, w)))

            five_hour_reset = any(_is_five_hour_window(w) and detected for _, w, detected in provider_windows)
            for previous, w, detected in provider_windows:
                simultaneous_long_reset = (
                    five_hour_reset
                    and _is_long_reset_window(w)
                    and _window_looks_fresh_after_reset(previous, w)
                )
                if not detected and not simultaneous_long_reset:
                    continue
                previous_remaining = max(0, min(100, 100 - int(previous.get("pct", 100))))
                early = _is_early_reset(previous) or simultaneous_long_reset
                if _is_long_reset_window(w):
                    notifications.append((provider_name, str(w.get("label", "")), int(w.get("duration_sec", 0) or 0), previous_remaining, early))
                elif (
                    _is_five_hour_window(w)
                    and previous_remaining < USAGE_RESET_5H_REMAINING_THRESHOLD
                    and not _is_sleeping_hour(now)
                ):
                    notifications.append((provider_name, str(w.get("label", "")), int(w.get("duration_sec", 0) or 0), previous_remaining, early))

    for provider_name, label, duration_sec, previous_remaining, early in notifications:
        _send_usage_reset_ntfy_async(provider_name, label, duration_sec, previous_remaining, early)


def _check_underpace(providers: list[dict]) -> None:
    global _underpace_last_fired
    now = time.time()

    triggered_pct: int | None = None
    triggered_resets_in_sec: int | None = None
    for prov in providers:
        if prov.get("name") != "Claude":
            continue
        for w in prov.get("windows", []):
            if w.get("label") == UNDERPACE_WINDOW:
                pct = w.get("pct", -1)
                resets_in_sec = w.get("resets_in_sec", -1)
                if 0 <= pct < UNDERPACE_PCT_THRESHOLD and resets_in_sec >= UNDERPACE_MIN_REMAINING_SEC:
                    triggered_pct = pct
                    triggered_resets_in_sec = resets_in_sec
                break

    if triggered_resets_in_sec is None:
        return

    with _underpace_lock:
        if (now - _underpace_last_fired) < UNDERPACE_COOLDOWN_SEC:
            return
        _underpace_last_fired = now
        _, project_name = _active_session_cwd()

    _send_underpace_ntfy_async(triggered_resets_in_sec, project_name, triggered_pct or 0)


def get_pace_payload(providers: list[dict] | None = None) -> dict:
    """Return current pace per window, computed from cached usage data."""
    if providers is None:
        cached = _get_cached_response(allow_stale=True)
        providers = cached.get("providers", []) if cached else []

    windows_out = []
    for prov in providers:
        if prov.get("name") != "Claude":
            continue
        for w in prov.get("windows", []):
            p = _compute_pace(
                w.get("pct", -1),
                w.get("resets_in_sec", -1),
                w.get("duration_sec", 0),
            )
            pct = w.get("pct", -1)
            resets_in_sec = w.get("resets_in_sec", -1)
            idle_trigger = (
                w.get("label") == UNDERPACE_WINDOW
                and 0 <= pct < UNDERPACE_PCT_THRESHOLD
                and resets_in_sec >= UNDERPACE_MIN_REMAINING_SEC
            )
            windows_out.append({
                "label": w.get("label"),
                "pct": pct,
                "pace": round(p, 4) if p is not None else None,
                "idle_trigger": idle_trigger,
            })

    with _underpace_lock:
        last_fired = _underpace_last_fired

    return {
        "ok": True,
        "updated_at": int(time.time()),
        "monitor_window": UNDERPACE_WINDOW,
        "pct_threshold": UNDERPACE_PCT_THRESHOLD,
        "min_remaining_sec": UNDERPACE_MIN_REMAINING_SEC,
        "last_fired": int(last_fired) if last_fired else None,
        "windows": windows_out,
    }


# ── Cache layer ───────────────────────────────────────────────────────────

_cache_lock = threading.Lock()
_usage_fetch_lock = threading.Lock()
_cached_response: dict | None = None
_cache_expires_at: float = 0


def _get_cached_response(allow_stale: bool = False) -> dict | None:
    with _cache_lock:
        if _cached_response and (allow_stale or time.time() < _cache_expires_at):
            return _cached_response
    return None


def get_usage_cached() -> dict:
    global _cached_response, _cache_expires_at

    cached = _get_cached_response()
    if cached:
        return cached

    if not _usage_fetch_lock.acquire(blocking=False):
        cached = _get_cached_response(allow_stale=True)
        if cached:
            return cached
        _usage_fetch_lock.acquire()

    try:
        cached = _get_cached_response()
        if cached:
            return cached

        # ── Claude (required) ──
        try:
            access_token, refresh_token = get_access_token()
        except Exception as e:
            return {"ok": False, "error": f"keychain: {e}", "updated_at": int(time.time()), "providers": []}

        try:
            claude_raw = fetch_claude_usage(access_token)
        except urllib.error.HTTPError as e:
            if e.code == 401 and refresh_token:
                try:
                    access_token = refresh_access_token(refresh_token)
                    claude_raw = fetch_claude_usage(access_token)
                except Exception as e2:
                    return {"ok": False, "error": f"refresh failed: {e2}", "updated_at": int(time.time()), "providers": []}
            else:
                return {"ok": False, "error": f"HTTP {e.code}", "updated_at": int(time.time()), "providers": []}
        except Exception as e:
            return {"ok": False, "error": str(e), "updated_at": int(time.time()), "providers": []}

        # ── Codex (optional, best-effort) ──
        codex_raw = None
        creds = read_codex_credentials()
        if creds:
            try:
                codex_raw = fetch_codex_usage(creds["access_token"], creds.get("account_id"))
            except urllib.error.HTTPError as e:
                if e.code in (401, 403) and creds.get("refresh_token"):
                    try:
                        new_token = refresh_codex_token(creds["refresh_token"])
                        codex_raw = fetch_codex_usage(new_token, creds.get("account_id"))
                    except Exception:
                        pass
            except Exception:
                pass

        # ── Build response ──
        providers = []
        attention = get_attention_by_provider()

        claude_windows = build_claude_windows(claude_raw)
        if claude_windows:
            providers.append({
                "name": "Claude",
                "attention": attention.get("Claude", _empty_attention()),
                "windows": claude_windows,
            })

        if codex_raw is not None:
            codex_windows = build_codex_windows(codex_raw)
            if codex_windows:
                providers.append({
                    "name": "Codex",
                    "attention": attention.get("Codex", _empty_attention()),
                    "windows": codex_windows,
                })

        shown = {prov["name"] for prov in providers}
        for name in DISPLAY_ATTENTION_PROVIDERS:
            attn = attention.get(name, _empty_attention())
            if name not in shown and attn.get("state") != "idle":
                providers.append({"name": name, "attention": attn, "windows": []})

        payload = {
            "ok": True,
            "updated_at": int(time.time()),
            "providers": providers,
        }

        with _cache_lock:
            _cached_response = payload
            _cache_expires_at = time.time() + CACHE_TTL_SEC

        _check_usage_resets(providers)
        _check_underpace(providers)
        return payload
    finally:
        _usage_fetch_lock.release()


# ── HTTP server ───────────────────────────────────────────────────────────

class Handler(BaseHTTPRequestHandler):
    def log_message(self, fmt, *args):
        ts = datetime.now().strftime("%H:%M:%S")
        print(f"[{ts}] {fmt % args}")

    def _send_json(self, payload: dict, status: int = 200):
        body = json.dumps(payload).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Access-Control-Allow-Origin", "*")
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        path = urlparse(self.path).path
        if path == "/usage":
            payload = get_usage_cached()
            self._send_json(payload)
        elif path == "/status":
            self._send_json(get_status_payload())
        elif path == "/pace":
            self._send_json(get_pace_payload())
        elif path == "/health":
            self.send_response(200)
            self.end_headers()
            self.wfile.write(b"ok")
        else:
            self.send_response(404)
            self.end_headers()

    def do_POST(self):
        path = urlparse(self.path).path
        if path != "/attention":
            self.send_response(404)
            self.end_headers()
            return

        try:
            length = int(self.headers.get("Content-Length", "0"))
            raw = self.rfile.read(length) if length > 0 else b"{}"
            payload = json.loads(raw)
            if not isinstance(payload, dict):
                raise ValueError("expected JSON object")
        except Exception as e:
            self._send_json({"ok": False, "error": f"invalid JSON: {e}"}, 400)
            return

        ok, response = ingest_attention_event(payload)
        self._send_json(response, 200 if ok else 400)


class CodexBarHTTPServer(ThreadingHTTPServer):
    allow_reuse_address = True
    daemon_threads = True


def _pids_on_port() -> list[int]:
    """Return PIDs currently listening on PORT."""
    try:
        result = subprocess.run(
            ["lsof", "-ti", f"tcp:{PORT}"],
            capture_output=True, text=True, timeout=5,
        )
        return [int(pid) for pid in result.stdout.split() if pid.strip()]
    except Exception:
        return []


def _replace_existing_if_requested() -> bool:
    """Return True if startup should continue."""
    import signal
    my_pid = os.getpid()
    pids = [pid for pid in _pids_on_port() if pid != my_pid]
    if not pids:
        return True

    replace = os.environ.get("CODEXBAR_REPLACE_EXISTING", "").strip().lower() in ("1", "true", "yes")
    if replace:
        for pid in pids:
            os.kill(pid, signal.SIGTERM)
            print(f"Killed existing process {pid} on port {PORT}")
        deadline = time.time() + 3
        while time.time() < deadline:
            remaining = [pid for pid in _pids_on_port() if pid != my_pid]
            if not remaining:
                return True
            time.sleep(0.2)
        remaining = [pid for pid in _pids_on_port() if pid != my_pid]
        print(f"Port {PORT} is still in use by PID(s): {', '.join(str(pid) for pid in remaining)}")
        print("If launchd is managing CodexBar, unload or update the LaunchAgent before replacing it manually:")
        print("  launchctl bootout gui/$(id -u) ~/Library/LaunchAgents/com.codexbar.companion.plist")
        return False

    print(f"Port {PORT} is already in use by PID(s): {', '.join(str(pid) for pid in pids)}")
    print("Another CodexBar companion may already be running, possibly via launchd.")
    print(f"Check: curl http://127.0.0.1:{PORT}/health")
    print("To replace it explicitly, run:")
    print(f"  CODEXBAR_REPLACE_EXISTING=1 python3 companion/server.py")
    return False


def main():
    import socket
    if not _replace_existing_if_requested():
        return
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.connect(("8.8.8.8", 80))
        local_ip = s.getsockname()[0]
        s.close()
    except Exception:
        local_ip = socket.gethostbyname(socket.gethostname())

    print(f"CodexBar companion proxy starting on port {PORT}")
    print(f"ESP32 PROXY_HOST = \"{local_ip}\"")
    print(f"Endpoint: http://{local_ip}:{PORT}/usage")
    print()

    try:
        server = CodexBarHTTPServer(("0.0.0.0", PORT), Handler)
    except OSError as e:
        if getattr(e, "errno", None) == 48:
            pids = [pid for pid in _pids_on_port() if pid != os.getpid()]
            suffix = f" by PID(s): {', '.join(str(pid) for pid in pids)}" if pids else ""
            print(f"Port {PORT} is already in use{suffix}.")
            print("If launchd is managing CodexBar, unload or update the LaunchAgent before replacing it manually:")
            print("  launchctl bootout gui/$(id -u) ~/Library/LaunchAgents/com.codexbar.companion.plist")
            print("Then start again, or reinstall the plist from companion/README.md.")
            return
        raise

    print("Fetching initial usage data…")
    payload = get_usage_cached()
    if payload["ok"]:
        for prov in payload.get("providers", []):
            print(f"  [{prov['name']}]")
            for w in prov["windows"]:
                print(f"    {w['label']:6s} {w['pct']:3d}%  resets in {w['resets_in_sec']}s")
    else:
        print(f"  Error: {payload.get('error')}")
    print()

    print(f"Serving on http://0.0.0.0:{PORT} — Ctrl-C to stop\n")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nStopped.")


if __name__ == "__main__":
    main()