#!/usr/bin/env python3
"""Lighthouse: a local Codex quota monitor and lightweight heartbeat."""

from __future__ import annotations

import fcntl
import glob
import json
import os
import pathlib
import selectors
import signal
import sqlite3
import subprocess
import sys
import tempfile
import time
import uuid
from datetime import datetime, timezone
from typing import Any


CODEX_HOME = pathlib.Path(os.environ.get("CODEX_HOME", "/home/io/.codex"))
CODEX_BIN = pathlib.Path("/usr/lib/chatgpt/resources/codex")
RUNTIME_DIR = pathlib.Path("/home/io/.local/share/lighthouse")
STATE_DIR = pathlib.Path(os.environ.get("XDG_STATE_HOME", "/home/io/.local/state")) / "lighthouse"
STATE_FILE = STATE_DIR / "state.json"
LOCK_FILE = STATE_DIR / "lighthouse.lock"
LOG_FILE = STATE_DIR / "lighthouse.log"
AUDIT_FILE = STATE_DIR / "beacon-audit.jsonl"
AUDIT_TEXT_FILE = STATE_DIR / "beacon-audit.log"
AUTOCOMMAND_DIR = STATE_DIR / "autocommand"
AUTOCOMMAND_REQUESTS = AUTOCOMMAND_DIR / "requests"
AUTOCOMMAND_RESULTS = AUTOCOMMAND_DIR / "results"
AUTOCOMMAND_PRESENCE = AUTOCOMMAND_DIR / "online.json"
SERVICE = "lighthouse-refresh.timer"
MODEL = "gpt-5.6-luna"
HEARTBEAT_EFFORT = "low"
WINDOW_START_EFFORT = "max"
PING_PROMPT = "Say okay, just okay"
STATUS_CACHE_SECONDS = 60
CHECK_INTERVAL_MINUTES = 5
HEARTBEAT_INTERVAL_MINUTES = 90
HEARTBEAT_INTERVAL_SECONDS = HEARTBEAT_INTERVAL_MINUTES * 60
NEAR_FULL_MINUTES = 4 * 60 + 45
NEAR_FULL_SECONDS = NEAR_FULL_MINUTES * 60
NEAR_FULL_LATCH_SECONDS = 30 * 60
PING_TIMEOUT_SECONDS = 45
FAILED_PING_COOLDOWN_SECONDS = 15 * 60
AUDIT_MAX_EVENTS = 300
SESSION_TAIL_BYTES = 2 * 1024 * 1024


def now() -> int:
    return int(time.time())


def iso(epoch: int | None = None) -> str:
    return datetime.fromtimestamp(epoch or now(), timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")


def setup() -> None:
    RUNTIME_DIR.mkdir(parents=True, exist_ok=True)
    STATE_DIR.mkdir(parents=True, exist_ok=True)
    AUTOCOMMAND_REQUESTS.mkdir(parents=True, exist_ok=True)
    AUTOCOMMAND_RESULTS.mkdir(parents=True, exist_ok=True)
    os.chmod(STATE_DIR, 0o700)


def load_state() -> dict[str, Any]:
    try:
        value = json.loads(STATE_FILE.read_text(encoding="utf-8"))
        return value if isinstance(value, dict) else {}
    except (OSError, ValueError):
        return {}


def save_state(state: dict[str, Any]) -> None:
    fd, temporary = tempfile.mkstemp(prefix="state-", suffix=".json", dir=STATE_DIR)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as stream:
            json.dump(state, stream, separators=(",", ":"), sort_keys=True)
            stream.write("\n")
            stream.flush()
            os.fsync(stream.fileno())
        os.chmod(temporary, 0o600)
        os.replace(temporary, STATE_FILE)
    finally:
        try:
            os.unlink(temporary)
        except FileNotFoundError:
            pass


def log(message: str) -> None:
    line = f"{iso()}  {message}\n"
    try:
        with LOG_FILE.open("a", encoding="utf-8") as stream:
            stream.write(line)
        os.chmod(LOG_FILE, 0o600)
        lines = LOG_FILE.read_text(encoding="utf-8", errors="replace").splitlines()[-250:]
        LOG_FILE.write_text("\n".join(lines) + "\n", encoding="utf-8")
        os.chmod(LOG_FILE, 0o600)
    except OSError:
        pass


def audit_event(event: str, decision: str, **details: Any) -> None:
    """Append one private, bounded machine-readable and human-readable event."""
    epoch = now()
    record = {
        "at": epoch,
        "at_iso": iso(epoch),
        "event": event,
        "decision": decision,
        **{key: value for key, value in details.items() if value is not None},
    }
    try:
        with AUDIT_FILE.open("a", encoding="utf-8") as stream:
            stream.write(json.dumps(record, separators=(",", ":"), sort_keys=True) + "\n")
        records = AUDIT_FILE.read_text(encoding="utf-8", errors="replace").splitlines()[-AUDIT_MAX_EVENTS:]
        AUDIT_FILE.write_text("\n".join(records) + "\n", encoding="utf-8")
        os.chmod(AUDIT_FILE, 0o600)

        readable = [record["at_iso"], event, decision]
        for key in ("trigger", "duration_seconds", "reply", "tokens", "five_hour", "weekly"):
            if key in record:
                value = json.dumps(record[key], separators=(",", ":")) if isinstance(record[key], (dict, list)) else str(record[key])
                readable.append(f"{key}={value}")
        with AUDIT_TEXT_FILE.open("a", encoding="utf-8") as stream:
            stream.write("  |  ".join(readable) + "\n")
        lines = AUDIT_TEXT_FILE.read_text(encoding="utf-8", errors="replace").splitlines()[-AUDIT_MAX_EVENTS:]
        AUDIT_TEXT_FILE.write_text("\n".join(lines) + "\n", encoding="utf-8")
        os.chmod(AUDIT_TEXT_FILE, 0o600)
    except OSError:
        pass


def recent_audit(limit: int = 8) -> list[dict[str, Any]]:
    try:
        lines = AUDIT_FILE.read_text(encoding="utf-8", errors="replace").splitlines()[-max(1, limit):]
    except OSError:
        return []
    events = []
    for line in lines:
        try:
            value = json.loads(line)
            if isinstance(value, dict):
                events.append(value)
        except ValueError:
            continue
    return events


def open_audit() -> int:
    if not AUDIT_TEXT_FILE.exists():
        AUDIT_TEXT_FILE.write_text("Lighthouse Beacon has not recorded any audit events yet.\n", encoding="utf-8")
        os.chmod(AUDIT_TEXT_FILE, 0o600)
    try:
        subprocess.Popen(
            [
                "zenity", "--text-info", "--title=Lighthouse Beacon Audit",
                f"--filename={AUDIT_TEXT_FILE}", "--width=1100", "--height=720",
                "--font=Ubuntu Sans 16", "--ok-label=Close",
            ],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            start_new_session=True,
        )
        return 0
    except OSError as error:
        print(f"Could not open Beacon audit: {error}", file=sys.stderr)
        return 1


def timer_state() -> tuple[bool, bool]:
    def quiet(*args: str) -> bool:
        return subprocess.run(args, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False).returncode == 0

    return (
        quiet("systemctl", "--user", "is-enabled", "--quiet", SERVICE),
        quiet("systemctl", "--user", "is-active", "--quiet", SERVICE),
    )


def normalize_window(value: Any) -> dict[str, Any] | None:
    if not isinstance(value, dict):
        return None
    used = value.get("usedPercent", value.get("used_percent"))
    duration = value.get("windowDurationMins", value.get("window_minutes"))
    resets = value.get("resetsAt", value.get("resets_at"))
    try:
        used_number = max(0.0, min(100.0, float(used)))
    except (TypeError, ValueError):
        used_number = None
    try:
        duration_number = int(duration)
    except (TypeError, ValueError):
        duration_number = None
    try:
        resets_number = int(resets)
    except (TypeError, ValueError):
        resets_number = None
    return {
        "used_percent": used_number,
        "remaining_percent": None if used_number is None else max(0.0, min(100.0, 100.0 - used_number)),
        "window_minutes": duration_number,
        "resets_at": resets_number,
    }


def classify_windows(rate_limits: dict[str, Any]) -> tuple[dict[str, Any] | None, dict[str, Any] | None]:
    windows: dict[int, dict[str, Any]] = {}
    for name in ("primary", "secondary"):
        window = normalize_window(rate_limits.get(name))
        if window and window.get("window_minutes") in (300, 10080):
            windows[int(window["window_minutes"])] = window
    return windows.get(300), windows.get(10080)


def normalize_usage(rate_limits: dict[str, Any], source: str, observed: int, reset_credits: Any = None) -> dict[str, Any]:
    five_hour, weekly = classify_windows(rate_limits)
    credits = reset_credits if isinstance(reset_credits, dict) else {}
    credit_list = credits.get("credits") if isinstance(credits.get("credits"), list) else []
    available = [item for item in credit_list if isinstance(item, dict) and item.get("status") == "available"]
    available_count = credits.get("availableCount")
    if not isinstance(available_count, int):
        available_count = len(available)
    return {
        "observed_at": iso(observed),
        "observed_epoch": observed,
        "source": source,
        "five_hour": five_hour,
        "weekly": weekly,
        "plan_type": rate_limits.get("planType", rate_limits.get("plan_type")),
        "credits": rate_limits.get("credits"),
        "reset_credits": {
            "available_count": available_count,
            "title": available[0].get("title") if available else None,
        },
    }


def app_server_usage(timeout: float = 12.0) -> dict[str, Any]:
    if not CODEX_BIN.is_file():
        raise RuntimeError(f"Codex CLI not found at {CODEX_BIN}")
    process = subprocess.Popen(
        [str(CODEX_BIN), "app-server"],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.DEVNULL,
        text=True,
        bufsize=1,
        env={**os.environ, "CODEX_HOME": str(CODEX_HOME)},
    )

    def send(payload: dict[str, Any]) -> None:
        assert process.stdin is not None
        process.stdin.write(json.dumps(payload, separators=(",", ":")) + "\n")
        process.stdin.flush()

    def receive(identifier: int) -> dict[str, Any]:
        assert process.stdout is not None
        deadline = time.monotonic() + timeout
        selector = selectors.DefaultSelector()
        selector.register(process.stdout, selectors.EVENT_READ)
        while True:
            remaining = deadline - time.monotonic()
            if remaining <= 0 or not selector.select(remaining):
                break
            line = process.stdout.readline()
            if not line:
                break
            try:
                payload = json.loads(line)
            except ValueError:
                continue
            if payload.get("id") == identifier:
                return payload
        raise RuntimeError("Codex app-server did not return rate limits")

    try:
        send({"id": 0, "method": "initialize", "params": {"clientInfo": {"name": "lighthouse", "version": "2.0"}}})
        initialized = receive(0)
        if "error" in initialized:
            raise RuntimeError(str(initialized["error"]))
        send({"method": "initialized", "params": {}})
        send({"id": 6, "method": "account/rateLimits/read", "params": {}})
        response = receive(6)
        if "error" in response:
            raise RuntimeError(str(response["error"]))
        result = response.get("result") or {}
        by_id = result.get("rateLimitsByLimitId") or {}
        rate_limits = by_id.get("codex") if isinstance(by_id, dict) else None
        if not isinstance(rate_limits, dict):
            rate_limits = result.get("rateLimits")
        if not isinstance(rate_limits, dict):
            raise RuntimeError("Codex app-server returned no Codex rate limit record")
        usage = normalize_usage(rate_limits, "app-server", now(), result.get("rateLimitResetCredits"))
        if not usage.get("five_hour") and not usage.get("weekly"):
            raise RuntimeError("Codex app-server returned no recognized 5-hour or weekly window")
        return usage
    finally:
        try:
            process.terminate()
            process.wait(timeout=2)
        except (ProcessLookupError, subprocess.TimeoutExpired):
            process.kill()


def parse_timestamp(value: Any, fallback: int) -> int:
    if not isinstance(value, str):
        return fallback
    try:
        return int(datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp())
    except ValueError:
        return fallback


def session_fallback() -> dict[str, Any] | None:
    files = glob.glob(str(CODEX_HOME / "sessions" / "**" / "*.jsonl"), recursive=True)
    files.sort(key=lambda path: os.path.getmtime(path), reverse=True)
    newest: dict[int, tuple[int, dict[str, Any], dict[str, Any]]] = {}
    for path in files[:24]:
        try:
            fallback_epoch = int(os.path.getmtime(path))
            with open(path, "rb") as stream:
                size = stream.seek(0, os.SEEK_END)
                stream.seek(max(0, size - SESSION_TAIL_BYTES))
                if size > SESSION_TAIL_BYTES:
                    stream.readline()
                lines = stream.read().decode("utf-8", errors="replace").splitlines()
        except OSError:
            continue
        for line in lines:
            try:
                record = json.loads(line)
            except ValueError:
                continue
            payload = record.get("payload") if isinstance(record, dict) else None
            rate_limits = payload.get("rate_limits") if isinstance(payload, dict) else None
            if not isinstance(rate_limits, dict):
                continue
            observed = parse_timestamp(record.get("timestamp"), fallback_epoch)
            for name in ("primary", "secondary"):
                window = normalize_window(rate_limits.get(name))
                duration = window.get("window_minutes") if window else None
                if duration in (300, 10080) and (duration not in newest or observed >= newest[duration][0]):
                    newest[duration] = (observed, window, rate_limits)
    if not newest:
        return None
    most_recent = max(item[0] for item in newest.values())
    metadata = max(newest.values(), key=lambda item: item[0])[2]
    return {
        "observed_at": iso(most_recent),
        "observed_epoch": most_recent,
        "source": "codex-session-fallback",
        "five_hour": newest.get(300, (0, None, {}))[1],
        "weekly": newest.get(10080, (0, None, {}))[1],
        "plan_type": metadata.get("plan_type", metadata.get("planType")),
        "credits": metadata.get("credits"),
        "reset_credits": {"available_count": 0, "title": None},
    }


def refresh_usage(state: dict[str, Any], force: bool = False) -> tuple[dict[str, Any] | None, str | None]:
    cached = state.get("usage") if isinstance(state.get("usage"), dict) else None
    age = now() - int(cached.get("observed_epoch", 0)) if cached else 10**9
    if cached and not force and age <= STATUS_CACHE_SECONDS:
        return cached, state.get("status_error")
    try:
        usage = app_server_usage()
        state["usage"] = usage
        state["status_error"] = None
        state["last_status_at"] = now()
        save_state(state)
        return usage, None
    except Exception as error:  # A stale-but-safe display is better than a crashed panel.
        message = str(error)[:500]
        fallback = session_fallback()
        if fallback and (not cached or fallback.get("observed_epoch", 0) > cached.get("observed_epoch", 0)):
            cached = fallback
            state["usage"] = fallback
        state["status_error"] = message
        state["last_status_at"] = now()
        save_state(state)
        return cached, message


def status_payload(force: bool = False) -> dict[str, Any]:
    state = load_state()
    usage, error = refresh_usage(state, force=force)
    enabled, active = timer_state()
    last_success = int(state.get("last_successful_ping_at") or 0)
    if not last_success and state.get("last_ping_status") == "ok":
        last_success = int(state.get("last_ping_at") or 0)
    beacon = {
        "enabled": enabled,
        "active": active,
        "mode": "heartbeat",
        "check_interval_minutes": CHECK_INTERVAL_MINUTES,
        "heartbeat_interval_minutes": HEARTBEAT_INTERVAL_MINUTES,
        "near_full_minutes": NEAR_FULL_MINUTES,
        "model": MODEL,
        "reasoning_effort": HEARTBEAT_EFFORT,
        "window_start_effort": WINDOW_START_EFFORT,
        "last_check_at": state.get("last_check_at"),
        "last_decision": state.get("last_decision", "waiting"),
        "last_ping_at": state.get("last_ping_at"),
        "last_ping_status": state.get("last_ping_status", "never"),
        "last_reply": state.get("last_reply"),
        "last_error": state.get("last_ping_error"),
        "last_ping_manual": state.get("last_ping_manual"),
        "last_token_usage": state.get("last_token_usage"),
        "last_ping_duration_seconds": state.get("last_ping_duration_seconds"),
        "last_ping_reason": state.get("last_ping_reason"),
        "window_verification": state.get("pending_window_verification"),
        "last_window_verified_at": state.get("last_window_verified_at"),
        "next_heartbeat_at": last_success + HEARTBEAT_INTERVAL_SECONDS if last_success else now(),
        "recent_audit": recent_audit(8),
    }
    return {
        "ok": usage is not None,
        "now": now(),
        "usage": usage,
        "status_error": error,
        "beacon": beacon,
    }


def extract_token_usage(path: pathlib.Path) -> dict[str, Any] | None:
    latest = None
    try:
        for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
            try:
                value = json.loads(line)
            except ValueError:
                continue
            candidates = [value]
            while candidates:
                item = candidates.pop()
                if isinstance(item, dict):
                    if any(key in item for key in ("input_tokens", "output_tokens", "total_tokens")):
                        latest = {key: item.get(key) for key in ("input_tokens", "cached_input_tokens", "output_tokens", "total_tokens") if key in item}
                    candidates.extend(item.values())
                elif isinstance(item, list):
                    candidates.extend(item)
    except OSError:
        return None
    return latest


def extract_session_id(path: pathlib.Path) -> str | None:
    """Find the generated Codex thread/session id in an exec JSON event stream."""
    try:
        for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
            try:
                value = json.loads(line)
            except ValueError:
                continue
            candidates = [value]
            while candidates:
                item = candidates.pop()
                if isinstance(item, dict):
                    for key in ("thread_id", "threadId", "session_id", "sessionId"):
                        identifier = item.get(key)
                        if isinstance(identifier, str) and len(identifier) >= 16:
                            return identifier
                    candidates.extend(item.values())
                elif isinstance(item, list):
                    candidates.extend(item)
    except OSError:
        pass
    return None


def cleanup_ping_session(events_path: pathlib.Path, sessions_before: set[str]) -> None:
    """Remove only the local session generated by Beacon's non-ephemeral request."""
    session_id = extract_session_id(events_path)
    current = set(glob.glob(str(CODEX_HOME / "sessions" / "**" / "*.jsonl"), recursive=True))
    new_files = current - sessions_before
    targets: set[str] = set()
    if session_id:
        targets.update(path for path in current if session_id in pathlib.Path(path).name)
    for path in new_files:
        try:
            with open(path, "r", encoding="utf-8", errors="replace") as stream:
                if PING_PROMPT in stream.read(SESSION_TAIL_BYTES):
                    targets.add(path)
        except OSError:
            continue
    for path in targets:
        try:
            pathlib.Path(path).unlink()
        except OSError as error:
            log(f"could not clean Beacon session {path}: {error}")

    history = CODEX_HOME / "history.jsonl"
    if session_id and history.is_file():
        try:
            lines = history.read_text(encoding="utf-8", errors="replace").splitlines()
            kept = [line for line in lines if session_id not in line]
            if len(kept) != len(lines):
                fd, temporary = tempfile.mkstemp(prefix="history-", suffix=".jsonl", dir=CODEX_HOME)
                with os.fdopen(fd, "w", encoding="utf-8") as stream:
                    stream.write("\n".join(kept) + ("\n" if kept else ""))
                    stream.flush()
                    os.fsync(stream.fileno())
                os.chmod(temporary, 0o600)
                os.replace(temporary, history)
        except OSError as error:
            log(f"could not clean Beacon history entry: {error}")

    if session_id:
        for database in CODEX_HOME.glob("state_*.sqlite"):
            try:
                connection = sqlite3.connect(database, timeout=5)
                connection.execute("PRAGMA foreign_keys=ON")
                row = connection.execute(
                    "SELECT source, title, first_user_message FROM threads WHERE id=?",
                    (session_id,),
                ).fetchone()
                if row and row[0] == "exec" and PING_PROMPT in (row[1], row[2]):
                    connection.execute(
                        "DELETE FROM thread_spawn_edges WHERE parent_thread_id=? OR child_thread_id=?",
                        (session_id, session_id),
                    )
                    connection.execute("DELETE FROM threads WHERE id=?", (session_id,))
                    connection.commit()
                connection.close()
            except (OSError, sqlite3.Error) as error:
                log(f"could not clean Beacon thread index: {error}")


def run_ping(state: dict[str, Any], manual: bool, reason: str, ensure_window: bool = False) -> bool:
    ping_epoch = now()
    started = time.monotonic()
    effort = WINDOW_START_EFFORT if ensure_window else HEARTBEAT_EFFORT
    request_id = str(uuid.uuid4())
    request_path = AUTOCOMMAND_REQUESTS / f"{request_id}.json"
    result_path = AUTOCOMMAND_RESULTS / f"{request_id}.json"
    audit_event(
        "PING SENT",
        f"handing request to visible Watery Autocommand · Luna-{effort}",
        trigger=reason,
        model=MODEL,
        effort=effort,
        timeout_seconds=PING_TIMEOUT_SECONDS,
    )
    try:
        try:
            online = json.loads(AUTOCOMMAND_PRESENCE.read_text(encoding="utf-8"))
            online_at = int(online.get("at") or 0)
        except (OSError, ValueError):
            online_at = 0
        if now() - online_at > 10:
            raise RuntimeError("Watery Autocommand window is not online")

        payload = {
            "id": request_id,
            "created_at": ping_epoch,
            "prompt": PING_PROMPT,
            "reason": reason,
            "manual": manual,
            "model": MODEL,
            "effort": effort,
            "ensure_window": ensure_window,
        }
        fd, temporary = tempfile.mkstemp(prefix="request-", suffix=".json", dir=AUTOCOMMAND_REQUESTS)
        with os.fdopen(fd, "w", encoding="utf-8") as stream:
            json.dump(payload, stream, separators=(",", ":"), sort_keys=True)
            stream.write("\n")
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(temporary, request_path)

        deadline = time.monotonic() + PING_TIMEOUT_SECONDS + 5
        while time.monotonic() < deadline and not result_path.exists():
            time.sleep(0.25)
        if not result_path.exists():
            raise TimeoutError("Watery Autocommand did not return a result")
        result = json.loads(result_path.read_text(encoding="utf-8"))
        result_path.unlink(missing_ok=True)
        reply = str(result.get("reply") or "").strip()[:100]
        token_usage = result.get("tokens") if isinstance(result.get("tokens"), dict) else None
        duration = float(result.get("duration_seconds") or round(time.monotonic() - started, 2))
        returncode = int(result.get("returncode") or 0)
        success = bool(result.get("success")) and bool(reply)
        state.update({
            "last_ping_at": ping_epoch,
            "last_ping_iso": iso(ping_epoch),
            "last_ping_status": "ok" if success else "error",
            "last_reply": reply,
            "last_ping_manual": manual,
            "last_token_usage": token_usage,
            "last_ping_duration_seconds": duration,
            "last_ping_reason": reason,
            "last_ping_error": None if success else str(result.get("error") or f"Codex exited {returncode}; reply={reply!r}")[:500],
        })
        if success:
            state["last_successful_ping_at"] = ping_epoch
        save_state(state)
        audit_event(
            "ANSWER RECEIVED" if success else "FAILED",
            "non-empty reply received" if success else "invocation returned no usable reply",
            trigger=reason,
            duration_seconds=duration,
            reply=reply or "(empty)",
            tokens=token_usage,
            returncode=returncode,
        )
        log(f"{'manual' if manual else 'automatic'} Autocommand {'ok' if success else 'failed'} model={MODEL} effort={effort}")
        return success
    except Exception as error:
        duration = round(time.monotonic() - started, 2)
        state.update({
            "last_ping_at": ping_epoch,
            "last_ping_iso": iso(ping_epoch),
            "last_ping_status": "error",
            "last_ping_manual": manual,
            "last_ping_duration_seconds": duration,
            "last_ping_reason": reason,
            "last_ping_error": str(error)[:500],
        })
        save_state(state)
        audit_event("FAILED", "Watery Autocommand request failed", trigger=reason, duration_seconds=duration, error=str(error)[:200])
        log(f"{'manual' if manual else 'automatic'} Autocommand error: {str(error)[:160]}")
        return False
    finally:
        request_path.unlink(missing_ok=True)


def beacon() -> int:
    state = load_state()
    usage, error = refresh_usage(state, force=True)
    check_epoch = now()
    state["last_check_at"] = check_epoch
    if error or not usage or usage.get("source") != "app-server":
        state["last_decision"] = "status unavailable; no ping"
        save_state(state)
        audit_event("CHECK ONLY", "reliable app-server status unavailable; no ping", error=error)
        log("beacon skipped: reliable app-server status unavailable")
        return 0
    five_hour = usage.get("five_hour") or {}
    weekly = usage.get("weekly") or {}
    five_reset = five_hour.get("resets_at")
    weekly_reset = weekly.get("resets_at")
    five_active = isinstance(five_reset, int) and five_reset > check_epoch
    weekly_active = isinstance(weekly_reset, int) and weekly_reset > check_epoch
    five_snapshot = {
        "left_percent": five_hour.get("remaining_percent"),
        "minutes_to_reset": max(0, five_reset - check_epoch) // 60 if isinstance(five_reset, int) else None,
        "resets_at": five_reset,
    }
    weekly_snapshot = {
        "left_percent": weekly.get("remaining_percent"),
        "resets_at": weekly_reset,
    }
    if weekly and weekly_active and float(weekly.get("remaining_percent") or 0) <= 0.01:
        state["last_decision"] = "weekly quota exhausted; no ping"
        save_state(state)
        audit_event("CHECK ONLY", state["last_decision"], five_hour=five_snapshot, weekly=weekly_snapshot)
        return 0

    pending = state.get("pending_window_verification")
    if isinstance(pending, dict):
        requested_at = int(pending.get("requested_at") or 0)
        expected_reset = pending.get("reset_at")
        verification_age = check_epoch - requested_at
        if verification_age < 120:
            state["last_decision"] = "request delivered; waiting to verify server countdown"
            save_state(state)
            audit_event(
                "VERIFYING",
                state["last_decision"],
                trigger=pending.get("reason"),
                five_hour=five_snapshot,
                weekly=weekly_snapshot,
            )
            return 0

        reset_is_fixed = (
            isinstance(expected_reset, int)
            and isinstance(five_reset, int)
            and abs(five_reset - expected_reset) <= 30
            and five_reset > check_epoch
        )
        state.pop("pending_window_verification", None)
        if reset_is_fixed:
            state["last_window_verified_at"] = check_epoch
            state["last_near_full_ping_at"] = requested_at
            state["last_ping_status"] = "verified"
            state["last_decision"] = "server countdown verified; 5-hour window is running"
            save_state(state)
            audit_event(
                "WINDOW VERIFIED",
                state["last_decision"],
                trigger=pending.get("reason"),
                five_hour=five_snapshot,
                weekly=weekly_snapshot,
            )
            return 0

        state["last_ping_status"] = "window-not-started"
        state["last_ping_error"] = "Codex answered, but the 5-hour reset continued sliding"
        state["last_decision"] = "server countdown did not start; request rejected as heartbeat"
        save_state(state)
        audit_event(
            "WINDOW NOT STARTED",
            state["last_decision"],
            trigger=pending.get("reason"),
            expected_reset=expected_reset,
            observed_reset=five_reset,
            five_hour=five_snapshot,
            weekly=weekly_snapshot,
        )

    five_seconds = five_reset - check_epoch if isinstance(five_reset, int) else None
    near_full = five_seconds is not None and five_seconds >= NEAR_FULL_SECONDS
    window_needs_start = not five_active or near_full

    last_attempt = int(state.get("last_ping_at") or 0)
    last_success = int(state.get("last_successful_ping_at") or 0)
    if not last_success and state.get("last_ping_status") == "ok":
        last_success = last_attempt
    heartbeat_due = check_epoch - last_success >= HEARTBEAT_INTERVAL_SECONDS
    last_near_full = int(state.get("last_near_full_ping_at") or 0)
    near_full_due = window_needs_start and check_epoch - last_near_full >= NEAR_FULL_LATCH_SECONDS

    reasons = []
    if near_full_due:
        reasons.append("5-hour timer is near full or dormant")
    if heartbeat_due:
        reasons.append("90-minute heartbeat due")
    if not reasons:
        remaining = "unknown" if five_seconds is None else f"{max(0, five_seconds) // 60}m"
        latch = " · near-full latch active" if window_needs_start and not near_full_due else ""
        state["last_decision"] = f"heartbeat fresh; 5-hour timer {remaining}; no ping{latch}"
        save_state(state)
        audit_event("CHECK ONLY", state["last_decision"], five_hour=five_snapshot, weekly=weekly_snapshot)
        return 0
    if last_attempt and check_epoch - last_attempt < FAILED_PING_COOLDOWN_SECONDS:
        state["last_decision"] = "ping due; waiting for 15-minute retry cooldown"
        save_state(state)
        audit_event("CHECK ONLY", state["last_decision"], trigger=" + ".join(reasons), five_hour=five_snapshot, weekly=weekly_snapshot)
        return 0

    reason = " + ".join(reasons)
    state["last_ping_reason"] = reason
    state["last_decision"] = f"{reason}; saying okay"
    save_state(state)
    audit_event("CHECK → PING", state["last_decision"], trigger=reason, five_hour=five_snapshot, weekly=weekly_snapshot)
    success = run_ping(state, manual=False, reason=reason, ensure_window=near_full_due)
    state = load_state()
    if success:
        after, after_error = refresh_usage(state, force=True)
        after_five = (after or {}).get("five_hour") or {}
        after_weekly = (after or {}).get("weekly") or {}
        after_reset = after_five.get("resets_at")
        if near_full_due and isinstance(after_reset, int):
            state["pending_window_verification"] = {
                "requested_at": check_epoch,
                "reset_at": after_reset,
                "reason": reason,
            }
            state["last_ping_status"] = "awaiting-verification"
            state["last_decision"] = f"{reason}; answer received, server countdown not yet verified"
        else:
            state["last_decision"] = f"{reason}; automatic okay succeeded"
        save_state(state)
        audit_event(
            "QUOTA AFTER",
            (
                "request delivered; countdown verification pending"
                if near_full_due and isinstance(after_reset, int)
                else "post-heartbeat status refreshed"
            ) if not after_error else "post-heartbeat status refresh failed",
            trigger=reason,
            five_hour={
                "before_left": five_hour.get("remaining_percent"),
                "after_left": after_five.get("remaining_percent"),
                "resets_at": after_five.get("resets_at"),
            },
            weekly={
                "before_left": weekly.get("remaining_percent"),
                "after_left": after_weekly.get("remaining_percent"),
                "resets_at": after_weekly.get("resets_at"),
            },
            error=after_error,
        )
    else:
        state["last_decision"] = f"{reason}; automatic okay failed; retry in 15m"
        save_state(state)
    return 0 if success else 1


def manual_ping() -> int:
    state = load_state()
    before = state.get("usage") if isinstance(state.get("usage"), dict) else {}
    before_reset = ((before.get("five_hour") or {}).get("resets_at"))
    ensure_window = isinstance(before_reset, int) and before_reset - now() >= NEAR_FULL_SECONDS
    state["last_decision"] = "manual ping requested"
    save_state(state)
    audit_event("CHECK → PING", "manual ping requested", trigger="manual button")
    success = run_ping(state, manual=True, reason="manual button", ensure_window=ensure_window)
    if success:
        refreshed_state = load_state()
        after, after_error = refresh_usage(refreshed_state, force=True)
        after_five = ((after or {}).get("five_hour") or {})
        after_reset = after_five.get("resets_at")
        check_epoch = now()
        if isinstance(after_reset, int) and after_reset - check_epoch >= NEAR_FULL_SECONDS:
            refreshed_state["pending_window_verification"] = {
                "requested_at": check_epoch,
                "reset_at": after_reset,
                "reason": "manual button",
            }
            refreshed_state["last_ping_status"] = "awaiting-verification"
            refreshed_state["last_decision"] = "manual request delivered; server countdown not yet verified"
        else:
            refreshed_state["last_decision"] = "manual request delivered inside an active window"
        save_state(refreshed_state)
        audit_event(
            "QUOTA AFTER",
            "post-manual-ping status refreshed" if not after_error else "post-manual-ping status refresh failed",
            trigger="manual button",
            five_hour={
                "before_left": (before.get("five_hour") or {}).get("remaining_percent"),
                "after_left": ((after or {}).get("five_hour") or {}).get("remaining_percent"),
            },
            weekly={
                "before_left": (before.get("weekly") or {}).get("remaining_percent"),
                "after_left": ((after or {}).get("weekly") or {}).get("remaining_percent"),
            },
            error=after_error,
        )
    print("Lighthouse ping: reply received" if success else "Lighthouse ping failed", file=sys.stdout if success else sys.stderr)
    return 0 if success else 1


def set_timer(enabled: bool) -> int:
    subprocess.run(["systemctl", "--user", "daemon-reload"], check=False)
    command = ["systemctl", "--user", "enable" if enabled else "disable", "--now", SERVICE]
    return subprocess.run(command, check=False).returncode


def self_test() -> int:
    sample = {
        "primary": {"usedPercent": 17, "windowDurationMins": 300, "resetsAt": 2000000000},
        "secondary": {"usedPercent": 9, "windowDurationMins": 10080, "resetsAt": 2000600000},
        "planType": "plus",
    }
    normalized = normalize_usage(sample, "self-test", now())
    assertions = [
        normalized["five_hour"]["remaining_percent"] == 83,
        normalized["weekly"]["remaining_percent"] == 91,
        normalized["five_hour"]["window_minutes"] == 300,
        normalized["weekly"]["window_minutes"] == 10080,
    ]
    if not all(assertions):
        print("Lighthouse self-test failed", file=sys.stderr)
        return 1
    print("Lighthouse self-test: ok")
    return 0


def main() -> int:
    setup()
    command = sys.argv[1] if len(sys.argv) > 1 else "status-json"
    with LOCK_FILE.open("a+") as lock:
        fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
        if command == "status-json":
            print(json.dumps(status_payload(force=False), separators=(",", ":")))
            return 0
        if command in ("refresh-status", "refresh"):
            print(json.dumps(status_payload(force=True), separators=(",", ":")))
            return 0
        if command == "beacon":
            return beacon()
        if command == "ping":
            return manual_ping()
        if command == "enable":
            return set_timer(True)
        if command == "disable":
            return set_timer(False)
        if command == "log":
            try:
                print("\n".join(LOG_FILE.read_text(encoding="utf-8", errors="replace").splitlines()[-40:]))
            except FileNotFoundError:
                pass
            return 0
        if command == "audit-json":
            print(json.dumps(recent_audit(30), separators=(",", ":")))
            return 0
        if command == "open-audit":
            return open_audit()
        if command == "self-test":
            return self_test()
    print("Usage: lighthouse {status-json|refresh-status|beacon|ping|enable|disable|log|audit-json|open-audit|self-test}", file=sys.stderr)
    return 2


if __name__ == "__main__":
    raise SystemExit(main())
