"""Simbiot Harness — дашборд задач: веб-страница + JSON-API.

Задача t-260929164722-60dae2 [M]: страница со всеми задачами харнеса
(статусы/размеры/уровни), автообновление. Плюс лестница уровней,
эскалации и свежие дайджесты — состояние харнеса одним взглядом.

Дизайн: stdlib-only (http.server + ThreadingHTTPServer), никаких
зависимостей и лишних процессов. Читает событие-стор напрямую через
run_projections() (та же проекция, что у scheduler/CLI — консистентно).

Порты: 8765 (dashboard) / 8766 (healthcheck, без авторизации),
SIMBIOT_DASH_PORT/SIMBIOT_DASH_ADDR/SIMBIOT_DASH_TOKEN — из env.
Открытый мир — токен SIMBIOT_DASH_TOKEN (query ?token= или cookie):
страница и API без токена отдают 401 (кроме порта 8766 — там только
"ok", без данных; healthcheck для systemd не должен падать на auth).
"""
import json
import os
import sys
import threading
import time
from http import cookies
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from harness import commands, es  # noqa: E402

_HERE = os.path.dirname(os.path.abspath(__file__))
_INDEX = os.path.join(_HERE, "dashboard.html")

_ADDR = os.environ.get("SIMBIOT_DASH_ADDR", "0.0.0.0")
_PORT = int(os.environ.get("SIMBIOT_DASH_PORT", "8765"))
_HEALTH_PORT = int(os.environ.get("SIMBIOT_DASH_HEALTH_PORT", "8766"))
_TOKEN = os.environ.get("SIMBIOT_DASH_TOKEN", "")
_REFRESH = int(os.environ.get("SIMBIOT_DASH_REFRESH", "5"))

_STATUS_ORDER = {"ready": 0, "running": 1, "failed": 2, "done": 3}
_SIZE_COLOR = {"S": "#1d7a3e", "M": "#a3690f", "L": "#FF4F00", "XL": "#b3261e"}

# ── чат Simbiot прямо в дашборде (t-261001100740-99d55e) ──────────────────
# Канал один для всех UI: БД портала agent_portal_chat (/root/coop/coop.db),
# вход — sender='user' (пишет портал/дашборд), ответ агента — sender='agent'
# (HTTP-push из сек-рефлекса portal_channel.push). Реплаи из стора харнеса
# (ReplySent) — дубль текста ответа, берём из проекции инбокса (канал «portal»,
# у master-хода там же ходим TG-реплаи — их chat_id не 'portal:*', не мешаем).
import sqlite3 as _sqlite3

_COOP_DB = "/root/coop/coop.db"
_COOP_AGENT_CACHE = {"id": None, "name": None, "ts": 0.0}
_CHAT_MAX = 8000  # ровно кап coop-портала в chat/send


def _coop_agent(ttl: float = 60.0):
    """Агент Simbiot в БД портала (id нужен для адресации agent_portal_chat)."""
    now = time.time()
    if _COOP_AGENT_CACHE["id"] and now - _COOP_AGENT_CACHE["ts"] < ttl:
        return _COOP_AGENT_CACHE["id"], _COOP_AGENT_CACHE["name"]
    try:
        db = _sqlite3.connect(_COOP_DB)
        db.row_factory = _sqlite3.Row
        row = db.execute(
            "SELECT id, agent_name FROM agents WHERE harness_type='simbiot' "
            "ORDER BY (agent_name='Simbiot') DESC, id ASC LIMIT 1").fetchone()
        db.close()
    except Exception:
        return None, None
    if not row:
        return None, None
    _COOP_AGENT_CACHE.update({"id": int(row["id"]), "name": row["agent_name"],
                              "ts": now})
    return _COOP_AGENT_CACHE["id"], _COOP_AGENT_CACHE["name"]


def _ts_to_ms(ts) -> int:
    """ISO-строка стора или int-ms → int-ms (единая сортировка чата)."""
    if isinstance(ts, (int, float)):
        return int(ts)
    from datetime import datetime

    try:
        return int(datetime.fromisoformat(str(ts).replace("Z", "+00:00")).timestamp() * 1000)
    except Exception:
        return 0


def build_chat(limit: int = 80) -> dict:
    """Лента чата: БД портала (primary) + реплаи стора (полнота текста батчей).

    Дедуп: вход пользователя в обоих источниках под одним ключом
    ('m-'+msg_id) — версию стора предпочитаем (там replied/reply_text).
    """
    agent_id, agent_name = _coop_agent()
    if not agent_id:
        return {"ok": False, "error": "no_agent", "messages": []}

    db = _sqlite3.connect(_COOP_DB)
    db.row_factory = _sqlite3.Row
    rows = db.execute(
        "SELECT msg_id, sender, text, ts, consumed FROM agent_portal_chat "
        "WHERE agent_id=? ORDER BY ts DESC LIMIT ?", (agent_id, limit)).fetchall()
    db.close()

    store = {}
    try:
        model = es.run_projections()
        for m in (model.get("inbox") or {}).values():
            if isinstance(m, dict) and str(m.get("chat_id") or "").startswith("portal"):
                m = dict(m)
                m["store_id"] = "m-" + str(m.get("update_id") or "")
                m["ts_ms"] = _ts_to_ms(m.get("ts") or m.get("date"))
                store[m["store_id"]] = m
    except Exception:
        pass  # стор недоступен: БД портала всё равно покажет ленту

    msgs = []
    # анкеры: msg_id agent-реплаев вида "reply-<ms>-<update_id>" — покрывает
    # push-строку в БД портала; реплай стора дублирует её — не показываем
    anchors = set()
    for r in rows:
        if r["sender"] == "agent" and str(r["msg_id"]).startswith("reply-"):
            anchors.add("-".join(str(r["msg_id"]).split("-", 2)[2:]))
    for r in reversed(rows):  # старые → новые
        # служебные команды канала (/set-tg-bot с токеном, cmd-settg-*) — не
        # чата: токен секрет, в ленту UI не показываем (t-261001100740)
        if (r["sender"] == "user" and (str(r["msg_id"]).startswith("cmd-settg-")
                                       or str(r["text"]).startswith("/set-tg-bot"))):
            continue
        if r["sender"] == "agent" and str(r["text"]).lstrip().startswith("/set-tg"):
            continue
        key = "m-" + str(r["msg_id"])
        st = store.get(key) if r["sender"] == "user" else None
        if st is not None and st.get("replied") and st.get("reply_text") \
                and str(st.get("update_id") or "") not in anchors:
            # реплай стора НЕ отражён в БД портала — показываем ПАРУ:
            # вопрос пользователя + ответ агента из стора
            msgs.append({
                "id": str(r["msg_id"]), "from": "user",
                "text": st.get("text") or r["text"],
                "ts": int(r["ts"] or 0), "channel": "portal", "pending": False,
            })
            msgs.append({
                "id": str(r["msg_id"]) + "#reply", "from": "agent",
                "text": st.get("reply_text"),
                "ts": _ts_to_ms(st.get("ts") or r["ts"]) or int(r["ts"] or 0),
                "channel": "portal", "pending": False,
            })
            continue
        if st is not None:
            text = st.get("text") or r["text"]
            pending = not (st.get("replied") or r["consumed"])
        else:
            text = r["text"]
            pending = bool(r["sender"] == "user" and not r["consumed"] and
                           time.time() * 1000 - int(r["ts"] or 0) < 600_000)
        msgs.append({
            "id": str(r["msg_id"]),
            "from": r["sender"],
            "text": text,
            "ts": int(r["ts"] or 0),
            "channel": "portal",
            "pending": bool(pending),
        })
    # последние реплаи стора, которых ещё нет в БД (push идёт напрямую);
    # их нет как строк БД — показываем только если ретрай не доехал до портала
    seen_db_ids = {str(r["msg_id"]) for r in rows}
    for st in store.values():
        if (str(st.get("update_id") or "") not in seen_db_ids
                and st.get("replied") and st.get("reply_text")):
            msgs.append({"id": st["store_id"], "from": "agent",
                         "text": st.get("reply_text"), "ts": st.get("ts_ms") or 0,
                         "channel": "portal", "pending": False})
    msgs.sort(key=lambda m: m["ts"])
    # схлопнуть подряд идущие ОДИНАКОВЫЕ agent-пуши (ретраи ретраят с новым
    # msg_id — settg-диагностика светится как шум ×3): keep-last + счётчик
    collapsed = []
    for m in msgs:
        if (m["from"] == "agent" and collapsed
                and collapsed[-1]["from"] == "agent"
                and collapsed[-1].get("text") == m.get("text")
                and int(m["ts"]) - int(collapsed[-1]["ts"]) < 600_000):
            collapsed[-1]["repeat"] = collapsed[-1].get("repeat", 1) + 1
            collapsed[-1]["ts"] = m["ts"]
            continue
        collapsed.append(m)
    return {"ok": True, "agent": agent_name, "messages": collapsed[-limit:]}


def chat_send(text: str) -> dict:
    """Сообщение владельца агенту: прямая запись в agent_portal_chat.

    HTTP-роут портала требует юзер-сессию, дашборд авторизован своим токеном —
    пишем напрямую в ту же таблицу (sender='user', consumed=0): сек-рефлекс
    харнеса поднимет её ≤20 с и мин-LLM ответит в тот же канал.
    """
    text = (text or "").strip()
    if not text:
        return {"ok": False, "error": "empty"}
    agent_id, _ = _coop_agent()
    if not agent_id:
        return {"ok": False, "error": "no_agent"}
    mid = f"dash-{agent_id}-{int(time.time() * 1000)}-{os.urandom(3).hex()}"
    db = _sqlite3.connect(_COOP_DB, timeout=10)
    try:
        db.execute(
            "INSERT INTO agent_portal_chat (msg_id, agent_id, sender, text, ts) "
            "VALUES (?,?,?,?,?)", (mid, agent_id, "user", text[:_CHAT_MAX],
                                   int(time.time() * 1000)))
        db.commit()
    finally:
        db.close()
    return {"ok": True, "msg_id": mid}


def _utcnow() -> str:
    return time.strftime("%Y-%m-%dT%H:%M:%S+00:00", time.gmtime())


def build_state() -> dict:
    """Срез харнеса для API: задачи, лестница, эскалации, дайджесты, спенд.

    Полный run_projections() на каждый запрос — безопасно: инкрементальный
    чекпоинт в БД, работа пропорциональна дельте событий, не истории.
    """
    model = es.run_projections()

    # отчёт тика, закрывшего задачу: digest с summary_ref == result_ref
    digest_by_ref = {}
    for d in (model.get("digests") or {}).values():
        if d.get("summary_ref"):
            digest_by_ref[d["summary_ref"]] = d

    tasks = []
    for t in model.get("tasks", {}).values():
        report = digest_by_ref.get(t.get("result_ref")) or {}
        tasks.append({
            "id": t.get("id"), "title": t.get("title", ""),
            "size": t.get("size", "?"), "status": t.get("status", "?"),
            "level": t.get("level"), "created": t.get("created"),
            "finished": t.get("finished"), "error": t.get("error"),
            # детали для раскрытия по клику (т-260929220004)
            "details": (t.get("details") or "")[:2000],
            "promotions": t.get("promotions"), "promoted_to": t.get("promoted_to"),
            "session_id": t.get("session_id"),
            # t-261001082111: время взятия в работу (TaskClaimed) — для
            # пометки «в работе» и таймера elapsed на карточке running
            "claimed_at": t.get("claimed_at"),
            "report_level": report.get("level"),
            "report_ts": report.get("ts"),
            "report": (report.get("text") or "")[:1500],
        })
    # два стабильных прохода: новые сверху внутри статуса,
    # статусы по порядку ready→running→failed→done
    tasks.sort(key=lambda x: x.get("created") or "", reverse=True)
    tasks.sort(key=lambda x: _STATUS_ORDER.get(x["status"], 9))

    ladder = []
    for name, cs in sorted(model.get("cycle_state", {}).items()):
        ladder.append({
            "level": name, "busy": bool(cs.get("busy")),
            "paused": bool(cs.get("paused")),
            "ticks": cs.get("ticks", 0), "absorbed": cs.get("absorbed", 0),
            "errors": cs.get("errors", 0), "last_tick": cs.get("last_tick"),
            "last_error": cs.get("last_error"),
        })

    escalations = [e for e in (model.get("escalations") or {}).values()
                   if e.get("status") == "open"]
    escalations.sort(key=lambda e: e.get("ts", ""), reverse=True)

    digests = []
    for d in model.get("digests", {}).values():
        digests.append({
            "level": d.get("level"), "period": d.get("period"),
            "text": d.get("text", ""), "ts": d.get("ts"),
            "summary_ref": d.get("summary_ref"),
        })
    digests.sort(key=lambda d: d.get("ts") or "", reverse=True)

    spend = model.get("spend", [])[-12:]

    counts = {}
    for t in tasks:
        key = f"{t['status']}:{t['size']}"
        counts[key] = counts.get(key, 0) + 1

    return {
        "generated_at": _utcnow(),
        "counts": {
            "tasks_total": len(tasks),
            "tasks_by_status": {
                s: sum(1 for t in tasks if t["status"] == s)
                for s in ("ready", "running", "failed", "done")
            },
            "tasks_by_status_size": counts,
            "escalations_open": len(escalations),
            "events_total": es.count_events(),
            "digests_total": len(model.get("digests", {})),
        },
        "ladder": ladder,
        "tasks": tasks,
        "escalations": escalations,
        "digests": digests[:12],
        "spend": spend,
        "meta": {"refresh_seconds": _REFRESH, "token_required": bool(_TOKEN)},
    }


class _Handler(BaseHTTPRequestHandler):
    server_version = "SimbiotDash/1.0"

    def log_message(self, fmt, *args):  # тише в journalctl: одна строка
        print(f"[dash] {self.address_string()} {fmt % args}", flush=True)

    # ── auth ──
    def _authorized(self) -> bool:
        if not _TOKEN:
            return True
        supplied = None
        qs = self.path.partition("?")[2]
        from urllib.parse import parse_qs
        for v in parse_qs(qs).get("token", []):
            supplied = v
        if supplied is None:
            c = cookies.SimpleCookie(self.headers.get("Cookie", ""))
            if "simbiot_token" in c:
                supplied = c["simbiot_token"].value
        return supplied == _TOKEN

    def _forbidden(self):
        body = b'{"error": "unauthorized: pass ?token=SIMBIOT_DASH_TOKEN"}\n'
        self.send_response(401)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("WWW-Authenticate", 'Bearer realm="simbiot-dash"')
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _json(self, obj, code=200):
        body = json.dumps(obj, ensure_ascii=False, default=str).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Cache-Control", "no-store")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _html(self, text: str):
        body = text.encode()
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Cache-Control", "no-store")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    # ── routes ──
    def do_GET(self):
        route = self.path.partition("?")[0]
        if route == "/api/state":
            if not self._authorized():
                return self._forbidden()
            try:
                return self._json(build_state())
            except Exception as e:  # стор бит/занят — 503, фронт покажет retry
                return self._json({"error": str(e)[:500]}, code=503)
        if route == "/api/chat":
            # чат Simbiot в дашборде (t-261001100740): read — авторизуем так же
            if not self._authorized():
                return self._forbidden()
            try:
                return self._json(build_chat())
            except Exception as e:
                return self._json({"ok": False, "error": str(e)[:500],
                                   "messages": []}, code=503)
        if route == "/api/health":
            return self._json({"ok": True, "ts": _utcnow()})
        if route in ("/", "/index.html"):
            if not self._authorized():
                return self._forbidden()
            try:
                with open(_INDEX, encoding="utf-8") as f:
                    return self._html(f.read())
            except OSError as e:
                return self._json({"error": f"dashboard.html not found: {e}"}, code=500)
        return self._json({"error": "not found"}, code=404)

    def do_POST(self):
        route = self.path.partition("?")[0]
        if route == "/api/chat/send":
            # send чата (t-261001100740): та же авторизация, что у read-API
            if not self._authorized():
                return self._forbidden()
            try:
                n = int(self.headers.get("Content-Length") or 0)
            except ValueError:
                n = 0
            if n <= 0 or n > 64_000:
                return self._json({"ok": False, "error": "bad length"}, code=400)
            try:
                data = json.loads(self.rfile.read(n).decode("utf-8"))
            except Exception:
                return self._json({"ok": False, "error": "bad json"}, code=400)
            try:
                return self._json(chat_send(str(data.get("text") or "")))
            except Exception as e:
                return self._json({"ok": False, "error": str(e)[:500]}, code=503)
        return self._json({"error": "not found"}, code=404)


def _health_server():
    """Отдельный мини-порт для healthcheck: без auth, только {ok:true}."""
    class H(BaseHTTPRequestHandler):
        def log_message(self, fmt, *args):
            pass

        def do_GET(self):
            body = b'{"ok":true}'
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)

    srv = ThreadingHTTPServer(("127.0.0.1", _HEALTH_PORT), H)
    srv.serve_forever()


def main():
    es.init_db()  # идемпотентно
    if not os.path.exists(_INDEX):
        print(f"[dash] FATAL: {_INDEX} не найден", file=sys.stderr)
        sys.exit(1)
    threading.Thread(target=_health_server, daemon=True).start()
    print(f"[dash] http://{_ADDR}:{_PORT}/  (health 127.0.0.1:{_HEALTH_PORT}/api/health) "
          f"refresh={_REFRESH}s token={'on' if _TOKEN else 'off'}", flush=True)
    ThreadingHTTPServer((_ADDR, _PORT), _Handler).serve_forever()


if __name__ == "__main__":
    main()