#!/usr/bin/python3
# ruff: noqa: E501
"""Small, dependency-free status collector, web surface, and chat card."""

from __future__ import annotations

import argparse
import html
import json
import math
import os
import re
import subprocess
import tempfile
import threading
import time
import urllib.error
import urllib.request
from collections import deque
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

INTERVAL = 30
HISTORY_POINTS = 24 * 60 * 60 // INTERVAL
SAMPLE = re.compile(r'^([a-zA-Z_:][a-zA-Z0-9_:]*)(?:\{([^}]*)\})?\s+([^\s]+)')
LABEL = re.compile(r'([a-zA-Z_][a-zA-Z0-9_]*)="((?:\\.|[^"\\])*)"')


def parse_metrics(text):
    rows = []
    for line in text.splitlines():
        match = SAMPLE.match(line)
        if not match or line.startswith("#"):
            continue
        try:
            value = float(match.group(3))
        except ValueError:
            continue
        if not math.isfinite(value):
            continue
        labels = {
            key: bytes(value, "utf-8").decode("unicode_escape")
            for key, value in LABEL.findall(match.group(2) or "")
        }
        rows.append((match.group(1), labels, value))
    return rows


def values(rows, *names):
    wanted = set(names)
    return [(labels, value) for name, labels, value in rows if name in wanted]


def total(rows, *names):
    found = values(rows, *names)
    return sum(value for _labels, value in found) if found else None


def _fetch(url, headers=None, timeout=2):
    request = urllib.request.Request(url, headers=headers or {})
    with urllib.request.urlopen(request, timeout=timeout) as response:
        return response.read().decode()


def _supervisor_states():
    try:
        result = subprocess.run(
            ["supervisorctl", "-c", "/opt/agent/supervisord.conf", "status"],
            capture_output=True,
            text=True,
            timeout=3,
        )
    except (OSError, subprocess.TimeoutExpired):
        return None
    if result.returncode:
        return None
    states = {}
    for line in result.stdout.splitlines():
        fields = line.split()
        if len(fields) >= 2:
            states[fields[0]] = fields[1]
    return states or None


def _duty_state(path=None, now=None):
    path = Path(path or os.environ.get("A2Y_DUTIES_STATE", "/home/agent/.hermes/cron/jobs.json"))
    now = now or time.time()
    if not path.is_file():
        return None
    try:
        payload = json.loads(path.read_text())
    except (OSError, ValueError):
        return {"state": "unreadable"}
    jobs = payload if isinstance(payload, list) else payload.get("jobs", [])
    managed = [j for j in jobs if (j.get("origin") or {}).get("platform") == "a2y-duty"]
    if not managed:
        return {"state": "not configured", "count": 0}
    stamps = [j.get("last_run_at") for j in managed if j.get("last_run_at")]
    stale = []
    for job in managed:
        stamp = job.get("last_run_at")
        if not stamp:
            continue
        try:
            last = datetime.fromisoformat(str(stamp).replace("Z", "+00:00"))
            fields = str(job.get("schedule") or "").split()
            if fields and fields[0].startswith("*/"):
                max_age = int(fields[0][2:]) * 2 * 60 + 300
            elif len(fields) > 1 and fields[1].startswith("*/"):
                max_age = int(fields[1][2:]) * 2 * 3600 + 3600
            elif len(fields) > 4 and fields[4] != "*":
                max_age = 15 * 86400
            else:
                max_age = 49 * 3600
            if now - last.astimezone(timezone.utc).timestamp() > max_age:
                stale.append(str(job.get("name") or "unnamed"))
        except (ValueError, TypeError, IndexError):
            stale.append(str(job.get("name") or "unnamed"))
    return {
        "state": "stale" if stale else ("healthy" if stamps else "waiting for first run"),
        "count": len(managed),
        "last_run_at": max(stamps) if stamps else None,
        "stale": stale,
    }


class Collector:
    def __init__(self, interval=INTERVAL):
        self.interval = interval
        self.agent = os.environ.get("AGENT_NAME", "agent")
        self.metrics_enabled = os.environ.get("A2Y_STATUS_METRICS_ENABLED", "0") == "1"
        metrics_port = os.environ.get("ACP2API_METRICS_PORT")
        if not metrics_port and os.environ.get("ACP2API_PORT"):
            metrics_port = str(int(os.environ["ACP2API_PORT"]) + 8)
        metrics_addr = os.environ.get("ACP2API_METRICS_ADDR", "")
        if metrics_addr and metrics_addr != "off":
            metrics_addr = metrics_addr.replace("0.0.0.0:", "127.0.0.1:", 1)
        default_acp_url = (
            f"http://{metrics_addr}/metrics"
            if metrics_addr
            else (f"http://127.0.0.1:{metrics_port}/metrics" if metrics_port else "")
        )
        self.acp_url = os.environ.get("A2Y_STATUS_ACP_METRICS_URL", default_acp_url)
        lite = os.environ.get("LITELLM_PORT", "4000")
        self.lite_url = os.environ.get("A2Y_STATUS_LITELLM_METRICS_URL", f"http://127.0.0.1:{lite}/metrics")
        self.history = deque(maxlen=HISTORY_POINTS)
        self.current = None
        self.lock = threading.Lock()
        self.last_error = None

    def collect(self):
        now = time.time()
        rows, scrape_errors = [], []
        if self.metrics_enabled:
            if self.acp_url:
                try:
                    rows += parse_metrics(_fetch(self.acp_url))
                except (OSError, urllib.error.URLError) as exc:
                    scrape_errors.append(f"acp2api: {exc}")
            try:
                key = os.environ.get("LITELLM_MASTER_KEY", "")
                headers = {"Authorization": f"Bearer {key}"} if key else {}
                rows += parse_metrics(_fetch(self.lite_url, headers))
            except (OSError, urllib.error.URLError) as exc:
                scrape_errors.append(f"litellm: {exc}")
        tokens = total(rows, "acp2api_tokens_total")
        if tokens is None:
            tokens = total(rows, "litellm_input_tokens_metric_total", "litellm_output_tokens_metric_total")
        turns = total(rows, "acp2api_turns_total")
        duration_sum = total(rows, "acp2api_turn_duration_seconds_sum", "turn_duration_seconds_sum")
        duration_count = total(rows, "acp2api_turn_duration_seconds_count", "turn_duration_seconds_count")
        latency = duration_sum / duration_count if duration_sum is not None and duration_count else None
        context_values = values(rows, "acp2api_context_fill_ratio")
        context = max((v for _l, v in context_values), default=None)
        sessions = total(rows, "acp2api_sessions_live", "sessions_live")
        resets_rows = values(rows, "acp2api_sessions_retired_total")
        reset_reasons = {}
        for labels, value in resets_rows:
            reason = labels.get("reason", "unknown")
            reset_reasons[reason] = reset_reasons.get(reason, 0) + value
        resets = None if not resets_rows else {
            "total": sum(reset_reasons.values()),
            "by_reason": reset_reasons,
        }
        costs = values(rows, "acp2api_cost_total", "acp2api_cost")
        cost = None if not costs else [
            {"amount": value, "currency": labels.get("currency", "reported"), "account": labels.get("account")}
            for labels, value in costs[:20]
        ]
        outcomes = values(rows, "acp2api_turns_total")
        limit_hits = None if not outcomes else sum(
            value for labels, value in outcomes
            if any(word in str(labels.get("outcome", "")).lower() for word in ("429", "quota", "rate", "401", "auth"))
        )
        if limit_hits is not None:
            previous = self.history[-1].get("limit_hits") if self.history else None
            if (previous is None and limit_hits > 0) or (
                previous is not None and limit_hits > previous
            ):
                stamp = datetime.now(timezone.utc).isoformat()
                self.last_error = f"limit/auth outcome observed at {stamp}"
        processes = _supervisor_states()
        process_health = None if processes is None else all(state == "RUNNING" for state in processes.values())
        health = "healthy"
        if process_health is False or (self.metrics_enabled and scrape_errors):
            health = "degraded"
        tokens_per_hour = None
        if tokens is not None and self.history and self.history[-1].get("tokens") is not None:
            elapsed = now - self.history[-1]["at"]
            delta = tokens - self.history[-1]["tokens"]
            if elapsed > 0 and delta >= 0:
                tokens_per_hour = delta * 3600 / elapsed
        first = self.history[0] if self.history else None
        tokens_window = (
            tokens - first["tokens"]
            if first
            and tokens is not None
            and first.get("tokens") is not None
            and tokens >= first["tokens"]
            else None
        )
        window_hours = (now - first["at"]) / 3600 if first else 0.0
        point = {
            "at": now,
            "tokens": tokens,
            "turns": turns,
            "latency_seconds": latency,
            "context_ratio": context,
            "limit_hits": limit_hits,
            "sessions_live": sessions,
            "context_resets": resets,
            "tokens_per_hour": tokens_per_hour,
        }
        self.history.append(point)
        snapshot = {
            "agent": self.agent,
            "health": health,
            "collected_at": now,
            "collector_interval_seconds": self.interval,
            "metrics": "enabled" if self.metrics_enabled else "disabled",
            "scrape_errors": scrape_errors,
            "tokens": tokens,
            "tokens_per_hour": tokens_per_hour,
            "tokens_window": tokens_window,
            "window_hours": window_hours,
            "turns": turns,
            "latency_seconds": latency,
            "context_ratio": context,
            "limit_hits": limit_hits,
            "sessions_live": sessions,
            "context_resets": resets,
            "cost": cost,
            "processes": processes,
            "duties": _duty_state(),
            "last_error": self.last_error,
            "history": list(self.history),
        }
        with self.lock:
            self.current = snapshot
        return snapshot

    def snapshot(self):
        with self.lock:
            current = self.current
        data = dict(current or self.collect())
        data["collector_age_seconds"] = max(0, time.time() - data["collected_at"])
        return data

    def run(self):
        while True:
            time.sleep(self.interval)
            self.collect()


def _show(value, suffix=""):
    if value is None:
        return "no data"
    if isinstance(value, float) and not value.is_integer():
        return f"{value:.2f}{suffix}"
    return f"{int(value):,}{suffix}"


def status_text(data):
    if data["metrics"] == "disabled":
        metrics = "metrics disabled"
    else:
        costs = data.get("cost") or []
        reported = ", ".join(
            f"{item['amount']:.4g} {item['currency']}" for item in costs
        ) or "no cost data"
        metrics = f"tokens {_show(data['tokens'])}, turns {_show(data['turns'])}; reported cost {reported}"
    reset = data.get("context_resets")
    resets = "no data" if reset is None else _show(reset["total"])
    duties = data.get("duties") or {}
    errors = data.get("last_error") or "; ".join(data.get("scrape_errors") or []) or "none"
    window = "no data"
    if data.get("tokens_window") is not None:
        coverage = min(24.0, float(data.get("window_hours") or 0))
        window = f"{_show(data['tokens_window'])} tokens / {coverage:.1f}h collected (partial until 24h)"
    return (
        f"{data['agent']}: {data['health']}\n"
        f"Usage: {metrics}\n"
        f"Rolling window: {window}\n"
        f"Context: {_show(None if data.get('context_ratio') is None else data['context_ratio'] * 100, '%')}; resets: {resets}\n"
        f"Duties: {duties.get('state', 'no data')}; last error: {errors}\n"
    )


def _spark(data, key, width=180, height=34):
    series = [p.get(key) for p in data.get("history", []) if p.get(key) is not None]
    if len(series) < 2:
        return ""
    low, high = min(series), max(series)
    span = high - low or 1
    points = []
    for i, value in enumerate(series):
        x = i * width / max(1, len(series) - 1)
        y = height - ((value - low) / span * (height - 4) + 2)
        points.append(f"{x:.1f},{y:.1f}")
    return " ".join(points)


def status_svg(data):
    color = "#31c48d" if data["health"] == "healthy" else "#f59e0b"
    reset = data.get("context_resets")
    resets = None if reset is None else reset["total"]
    fields = [
        ("spend / hour", _show(data.get("tokens_per_hour"), " tok")),
        ("turns", _show(data.get("turns"))),
        ("latency", _show(data.get("latency_seconds"), "s")),
        ("context", _show(None if data.get("context_ratio") is None else data["context_ratio"] * 100, "%")),
        ("limit hits", _show(data.get("limit_hits"))),
        ("context resets", _show(resets)),
    ]
    tiles = []
    for i, (name, value) in enumerate(fields):
        x, y = 24 + (i % 3) * 238, 108 + (i // 3) * 84
        tiles.append(f'<rect x="{x}" y="{y}" width="218" height="64" rx="8" fill="#172033"/>')
        tiles.append(f'<text x="{x+14}" y="{y+24}" class="label">{html.escape(name)}</text>')
        tiles.append(f'<text x="{x+14}" y="{y+49}" class="value">{html.escape(value)}</text>')
    sparks = []
    for i, (key, label) in enumerate((("tokens_per_hour", "spend / hour"), ("turns", "turns"), ("context_ratio", "context"))):
        x = 24 + i * 238
        points = _spark(data, key)
        sparks.append(f'<text x="{x}" y="310" class="label">{label} · 24h</text>')
        if points:
            shifted = " ".join(f"{float(p.split(',')[0])+x:.1f},{float(p.split(',')[1])+324:.1f}" for p in points.split())
            sparks.append(f'<polyline points="{shifted}" fill="none" stroke="#7dd3fc" stroke-width="2"/>')
        else:
            sparks.append(f'<text x="{x}" y="344" class="muted">no data</text>')
    return f'''<svg xmlns="http://www.w3.org/2000/svg" width="760" height="380" viewBox="0 0 760 380">
<rect width="760" height="380" rx="18" fill="#0b1020"/>
<style>.title{{font:700 28px Liberation Sans,DejaVu Sans,sans-serif;fill:#f8fafc}}.state{{font:700 18px Liberation Sans,DejaVu Sans,sans-serif;fill:{color}}}.label{{font:14px Liberation Sans,DejaVu Sans,sans-serif;fill:#94a3b8}}.value{{font:700 21px Liberation Sans,Deja Vu Sans,sans-serif;fill:#f8fafc}}.muted{{font:13px Liberation Sans,DejaVu Sans,sans-serif;fill:#64748b}}</style>
<circle cx="36" cy="38" r="9" fill="{color}"/><text x="56" y="47" class="title">{html.escape(data['agent'])}</text>
<text x="24" y="82" class="state">{html.escape(data['health'])}</text><text x="150" y="82" class="label">{html.escape('metrics disabled' if data['metrics'] == 'disabled' else 'local metrics')}</text>
{''.join(tiles)}{''.join(sparks)}</svg>'''


def status_html(data):
    return f'''<!doctype html><html><head><meta charset="utf-8"><meta name="viewport" content="width=device-width"><title>{html.escape(data['agent'])} status</title><style>body{{margin:2rem;background:#050816;color:#e2e8f0;font:16px system-ui}}main{{max-width:800px;margin:auto}}svg{{width:100%;height:auto}}pre{{white-space:pre-wrap;background:#111827;padding:1rem;border-radius:12px}}</style></head><body><main><h1>Agent status</h1>{status_svg(data)}<pre>{html.escape(status_text(data))}</pre></main></body></html>'''


def _mvg_text(value):
    return str(value).replace("\\", "\\\\").replace("'", "\\'").replace("\n", " ")


def status_mvg(data):
    """Equivalent card instructions for ImageMagick's built-in MVG coder."""
    color = "#31c48d" if data["health"] == "healthy" else "#f59e0b"
    reset = data.get("context_resets")
    fields = [
        ("spend / hour", _show(data.get("tokens_per_hour"), " tok")),
        ("turns", _show(data.get("turns"))),
        ("latency", _show(data.get("latency_seconds"), "s")),
        ("context", _show(None if data.get("context_ratio") is None else data["context_ratio"] * 100, "%")),
        ("limit hits", _show(data.get("limit_hits"))),
        ("context resets", _show(None if reset is None else reset["total"])),
    ]
    lines = [
        "viewbox 0 0 760 380", "fill '#0b1020'", "rectangle 0,0 760,380",
        f"fill '{color}'", "circle 36,38 45,38", "font 'DejaVu-Sans'",
        "font-size 28", "font-weight 700", "fill '#f8fafc'",
        f"text 56,47 '{_mvg_text(data['agent'])}'", "font-size 18", f"fill '{color}'",
        f"text 24,82 '{_mvg_text(data['health'])}'", "font-size 14", "font-weight 400",
        "fill '#94a3b8'", f"text 150,82 '{'metrics disabled' if data['metrics'] == 'disabled' else 'local metrics'}'",
    ]
    for i, (name, value) in enumerate(fields):
        x, y = 24 + (i % 3) * 238, 108 + (i // 3) * 84
        lines += [
            "fill '#172033'", f"roundrectangle {x},{y} {x+218},{y+64} 8,8",
            "fill '#94a3b8'", "font-size 14", "font-weight 400",
            f"text {x+14},{y+24} '{_mvg_text(name)}'", "fill '#f8fafc'",
            "font-size 21", "font-weight 700", f"text {x+14},{y+49} '{_mvg_text(value)}'",
        ]
    for i, (key, label) in enumerate((("tokens_per_hour", "spend / hour"), ("turns", "turns"), ("context_ratio", "context"))):
        x = 24 + i * 238
        lines += ["fill '#94a3b8'", "font-size 14", "font-weight 400", f"text {x},310 '{label} · 24h'"]
        points = _spark(data, key)
        if points:
            shifted = " ".join(f"{float(p.split(',')[0])+x:.1f},{float(p.split(',')[1])+324:.1f}" for p in points.split())
            lines += ["fill none", "stroke '#7dd3fc'", "stroke-width 2", f"polyline {shifted}", "stroke none"]
        else:
            lines += ["fill '#64748b'", "font-size 13", f"text {x},344 'no data'"]
    return "\n".join(lines) + "\n"


def write_card(data, out):
    out = Path(out)
    out.mkdir(parents=True, exist_ok=True)
    svg = out / "status.svg"
    mvg = out / "status.mvg"
    png = out / "status.png"
    txt = out / "status.txt"
    svg.write_text(status_svg(data))
    mvg.write_text(status_mvg(data))
    txt.write_text(status_text(data))
    # The real image has the built-in MVG coder, but no SVG delegate. Keep SVG
    # as the browser surface and rasterize equivalent primitives through MVG.
    result = subprocess.run(
        ["convert", "-size", "760x380", f"mvg:{mvg}", str(png)],
        capture_output=True,
        text=True,
    )
    if result.returncode or not png.is_file():
        raise RuntimeError((result.stderr or result.stdout).strip() or "ImageMagick did not create PNG")
    return png, txt


class StatusHandler(BaseHTTPRequestHandler):
    collector = None

    def do_GET(self):
        data = self.collector.snapshot()
        content_type = "text/plain; charset=utf-8"
        if self.path == "/":
            body, content_type = status_html(data).encode(), "text/html; charset=utf-8"
        elif self.path == "/status.json":
            body, content_type = json.dumps(data, ensure_ascii=False).encode(), "application/json"
        elif self.path == "/status.txt":
            body = status_text(data).encode()
        elif self.path == "/status.svg":
            body, content_type = status_svg(data).encode(), "image/svg+xml"
        elif self.path == "/status.png":
            with tempfile.TemporaryDirectory() as directory:
                body = write_card(data, directory)[0].read_bytes()
            content_type = "image/png"
        else:
            self.send_error(404)
            return
        self.send_response(200)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, fmt, *args):
        print(f"status http: {fmt % args}")


def main(argv=None):
    parser = argparse.ArgumentParser(prog="a2y-status")
    sub = parser.add_subparsers(dest="command", required=True)
    serve = sub.add_parser("serve")
    serve.add_argument("--host", default=os.environ.get("A2Y_STATUS_HOST", "127.0.0.1"))
    serve.add_argument("--port", type=int, default=int(os.environ.get("A2Y_STATUS_PORT", "10025")))
    card = sub.add_parser("card")
    card.add_argument("--out", default=".")
    args = parser.parse_args(argv)
    collector = Collector()
    collector.collect()
    if args.command == "card":
        png, txt = write_card(collector.snapshot(), args.out)
        print(png)
        print(txt)
        return 0
    StatusHandler.collector = collector
    thread = threading.Thread(target=collector.run, daemon=True)
    thread.start()
    server = ThreadingHTTPServer((args.host, args.port), StatusHandler)
    server.serve_forever()


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