"""Simbiot Harness — scheduler: тактовая лестница + cycle runners.

Правила:
- тик уровня идемпотентен (UNIQUE level:seq);
- один исполнитель на уровень → сериализация;
- занятый уровень поглощает тик (TickAbsorbed) и пишет
  детерминированную дайджест-заглушку — без LLM;
- запущенная задача не прерывается.
"""
import json
import os
import time
import traceback
from datetime import datetime, timezone, timedelta

from . import es
from .levels import LEVELS
from . import runtime
from . import commands
from .commands import _ORDER

# интервал демона
POLL = 1.0
# уровни, включённые в v0.1 (порядок для лога)
ACTIVE = ["sec", "min", "hour", "day"]

# состояние исполнителей уровня: level → {"busy": bool, "task": id|None}
_runners = {}
# когда уровень последний раз слал TG-уведомление о занятости (анти-спам)
_NOTIFY_TS = {}


def _now():
    return datetime.now(timezone.utc)


def _log(msg: str):
    print(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}", flush=True)


def tick_number(level_cfg, ts: int):
    return ts // level_cfg.every


def emit_tick(level_cfg, seq: int):
    """CycleTicked идемпотентно. Возвращает True если это новый тик."""
    r = es.append_tick(level_cfg.name, seq)
    return r is not None


def deterministic_digest_stub(level_cfg, seq: int, reason: str):
    """Заглушка дайджеста для поглощённого тика — детерминированно, без LLM."""
    period = datetime.fromtimestamp(seq * level_cfg.every, tz=timezone.utc)
    es.append_event("cycle", f"stub:{level_cfg.name}:{seq}", "DigestWritten", {
        "level": level_cfg.name,
        "period": period.isoformat(),
        "summary_ref": None,
        "text": f"[absorbed:{reason}] уровень {level_cfg.name} был занят/пропущен; такт {seq}",
        "stub": True,
    })


def find_task(level_cfg, model: dict):
    """Ищет ready-задачу доступного классу уровня. Возвращает task_id или None."""
    classes = level_cfg.task_classes
    for tid, t in model["tasks"].items():
        if t["status"] == "ready" and t.get("size") in classes:
            return tid
    return None


def _lvl_rank(name: str) -> int:
    """Ранг уровня в лестнице; неизвестный → ниже всех (эскалации ему не адресованы)."""
    return _ORDER.index(name) if name in _ORDER else len(_ORDER)


def build_context(level_cfg, model: dict, task=None):
    """Сборка контекста тика: мандат + память уровня + выжимки нижних + задача."""
    parts = [level_cfg.mandate]

    # идентичность/цели v0.1 — статично из policy (если есть)
    policy = model.get("policy")
    if policy and policy.get("identity"):
        parts.append("ИДЕНТИЧНОСТЬ: " + policy["identity"])

    # выжимки нижних уровней: последние дайджесты
    order = ["min", "hour", "day"]
    idx = order.index(level_cfg.name) if level_cfg.name in order else 0
    lower = order[:idx]
    digests = []
    for lvl in lower:
        for key, d in model["digests"].items():
            if d["level"] == lvl and not d.get("stub"):
                digests.append(d)
    digests.sort(key=lambda d: d["ts"], reverse=True)
    if digests:
        parts.append("ВЫЖИМКИ НИЖНИХ УРОВНЕЙ (последние, сжатые):")
        for d in digests[:6]:
            parts.append(f"- [{d['level']}] {d['text'][:400]}")

    # эскалации: открытые, адресованные этому уровню и выше — в приоритете
    escs = []
    for e in (model.get("escalations") or {}).values():
        if not isinstance(e, dict) or e.get("status") != "open" or e.get("stub"):
            continue
        to_level = e.get("to_level", "day")
        if to_level not in _ORDER or _lvl_rank(level_cfg.name) < _lvl_rank(to_level):
            continue
        escs.append(e)
    if escs:
        escs.sort(key=lambda e: ({'critical': 0, 'warn': 1, 'info': 2}.get(e.get("severity"), 3),
                                 e.get("ts", "")), reverse=False)
        lines = []
        for e in escs[:10]:
            who = f"от {e.get('from_level')}" if e.get("from_level") else ""
            lines.append(f"- [{e['id']}] ({e.get('severity','info')}) {who} → "
                         f"{e.get('to_level')}: {e.get('reason','')[:300]}"
                         + (f"\n    контекст: {e.get('context','')[:400]}" if e.get("context") else ""))
        parts.append("ЭСКАЛАЦИИ (важное с нижних уровней; отработай и закрой: "
                     "./venv/bin/python cli.py escalation-ack <id>):\n" + "\n".join(lines))

    # инбокс: входящие TG-сообщения, ждущие ответа (в приоритете).
    # Голосовое с пустой расшифровкой (тишина/отказ) тоже показываем —
    # минутный уровень должен знать, что голосовое пришло, но текста нет,
    # и ответить «не расслышал», а не делать вид, что сообщения не было.
    # Команды бота (/лестница и др.) — НЕ здесь: их мгновенно закрывает
    # сек-рефлекс (t-260929210755), мин-LLM на них не тратится. Фильтруем
    # даже без ответа — иначе гонка на границе тика (команда пришла,
    # сек ещё не отвечал) или лежащий TG заставляли минуту отвечать LLM.
    from . import tg_commands
    inbox = [m for m in (model.get("inbox") or {}).values()
             if not m.get("replied") and not m.get("ignored")
             and not tg_commands.parse(m.get("text") or "")
             and (m.get("text") or m.get("kind") in ("voice", "audio", "video_note"))]
    if inbox:
        lines = []
        for m in sorted(inbox, key=lambda x: x.get("ts") or "")[-10:]:
            m_id = "m-" + str(m.get("update_id"))
            who = m.get("username") or m.get("first_name") or m.get("user_id")
            tag = ""
            if m.get("kind") == "voice":
                if m.get("transcribed") and m.get("text"):
                    tag = " [голосовое, расшифровка]"
                else:
                    tag = (" [голосовое, расшифровка не удалась — попроси прислать текстом]"
                           if m.get("transcribed") else
                           " [голосовое, ещё не расшифровано]")
            lines.append(f"- [{m_id}] от {who}{tag}: {m.get('text','')[:300]}")
        parts.append("ВХОДЯЩИЕ СООБЩЕНИЯ (ответь через: ./venv/bin/python cli.py tg-reply <id> \"текст\"):\n" + "\n".join(lines))

    # доска задач
    tasks = [t for t in model["tasks"].values() if t["status"] in ("ready", "running")]
    if tasks:
        lines = [f"- {t['id']}: {t['title'][:80]} [{t['size']}] {t['status']}" for t in tasks[:15]]
        parts.append("ДОСКА ЗАДАЧ:\n" + "\n".join(lines))

    if task:
        parts.append(f"ТЕКУЩАЯ ЗАДАЧА: {task['id']}: {task['title']}\n"
                     + (f"ПОСТАНОВКА: {task['details']}\n" if task.get("details") else "")
                     + "Выполни её и выдай результат.")

    return "\n\n".join(parts)


def _notify_busy(level_cfg, task):
    """TG-visibility: уровни сами сообщают о занятости.

    Анти-спам: по одному уведомлению на уровень не чаще раза в 15 минут —
    повторные взятия задач внутри часа не заспамят чат.
    """
    global _NOTIFY_TS
    now = time.time()
    if now - _NOTIFY_TS.get(level_cfg.name, 0) < 900:  # 15 минут
        return
    _NOTIFY_TS[level_cfg.name] = now
    from . import tg_inbox
    tg_inbox.notify_chat(
        f"🔧 Уровень {level_cfg.name} взял задачу [{task.get('size', '?')}] "
        f"«{task.get('title', '')[:70]}»\n"
        f"Срез лестницы по запросу: /лестница"
    )


def run_level_tick(level_cfg, seq: int, model: dict):
    """Один тик уровня с LLM: дайджест + (опц.) задача. Возвращает текст.

    LLM-тик гоняется в изолированном сабпроцессе (isolated_runner):
    у Strands глобальный asyncio-рантайм, второй create_harness()
    в том же процессе валит его («cannot schedule new futures»).
    """
    from . import isolated_runner

    task_id = find_task(level_cfg, model)
    task = model["tasks"].get(task_id) if task_id else None
    if task:
        es.append_event("task", task_id, "TaskClaimed", {
            "level": level_cfg.name, "session_id": f"s-{level_cfg.name}-{seq}",
        })
        # TG-visibility: взяли задачу — сообщаем в чат (анти-спам: не чаще 15м на уровень)
        try:
            from . import tg_inbox
            _notify_busy(level_cfg, task)
        except Exception:
            pass  # уведомление — best effort, не валить тик
    prompt = build_context(level_cfg, model, task)
    text, usage, trace_id = isolated_runner.run_llm_tick(
        level_cfg, prompt,
        session_id=f"s-{level_cfg.name}-{seq}",
        trace_prefix=f"{level_cfg.name}-{seq}",
    )
    if text is None:
        # воркер упал — задача не выполнена, откатываем claim
        if task:
            es.append_event("task", task_id, "TaskFailed", {
                "error": str(usage.get("error", "llm worker failed"))[:500],
            })
        raise RuntimeError(str(usage.get("error", "llm worker failed")))
    # дайджест тика
    period = datetime.fromtimestamp(seq * level_cfg.every, tz=timezone.utc)
    es.append_event("cycle", f"digest:{level_cfg.name}:{seq}", "DigestWritten", {
        "level": level_cfg.name, "period": period.isoformat(),
        "summary_ref": trace_id, "text": (text or "")[:2000],
    })
    # BudgetSpent воркер уже записал сам — инкрементально, после каждого
    # модельного вызова (устраняет потерю spend при таймауте воркера).
    # Если usage в результате пуст (воркер не смог разобрать историю —
    # старый воркер / нетекстовая структура), дублируем итог, чтобы тик
    # не терялся в spend вовсе: дедуп по trace_id в read model.
    if usage and not usage.get("model_calls"):
        es.append_event("harness", "budget", "BudgetSpent", {
            "level": level_cfg.name, "trace_id": trace_id, "source": "runner",
            **{k: v for k, v in usage.items() if k != "error"},
        })
    if task:
        es.append_event("task", task_id, "TaskCompleted", {
            "result_ref": trace_id, "answer": (text or "")[:2000],
        })
        # автодоставка (t-260929205927): задачу заказали из TG (await_delivery)
        # — результат уходит в чат сразу после завершения, без повторного запроса.
        # Единственный путь доставки — commands.deliver_result (идемпотентен:
        # без флага / уже sent — молча пропустит). Не отправилось здесь —
        # дослает delivery-демон (deliver_pending) на следующем проходе.
        try:
            from . import commands as _cmd
            if _cmd.deliver_result(task_id, text or ""):
                _log(f"delivery: результат {task_id} доставлен в TG")
        except Exception as e:
            # доставка — best effort: не валить тик; DeliverySent допишет
            # delivery-демон (см. deliver_pending) — задача не потеряется
            _log(f"delivery: отправка {task_id} не удалась ({e}) — доставит delivery-демон")
    return text, task_id


def process_level(level_cfg, model: dict):
    """Тик одного уровня: занят? → absorb; иначе работа."""
    st = _runners.setdefault(level_cfg.name, {"busy": False, "task": None})
    if st["busy"]:
        seq = tick_number(level_cfg, int(time.time()))
        deterministic_digest_stub(level_cfg, seq, "busy")
        es.append_event("cycle", f"absorbed:{level_cfg.name}:{seq}", "TickAbsorbed",
                        {"level": level_cfg.name, "reason": "busy"})
        return "absorbed"

    seq = tick_number(level_cfg, int(time.time()))
    if not emit_tick(level_cfg, seq):
        return "already"  # тик этого такта уже был (идемпотентность)

    st["busy"] = True
    try:
        if not level_cfg.uses_llm:
            # секундный уровень: детерминированный рефлекс без LLM
            text, task_id = _reflex_tick(level_cfg, seq, model)
            return f"reflex ok {task_id or ''}".strip()
        text, task_id = run_level_tick(level_cfg, seq, model)
        return f"ran (task={task_id or '-'})"
    except Exception as e:
        es.append_event("harness", "errors", "CycleError", {
            "level": level_cfg.name, "error": str(e)[:1000],
            "trace": traceback.format_exc()[-2000:],
        })
        # воркер LLM-тика умер — наверх, чтобы верхний уровень узнал сразу:
        # нижний сам себя не чинит (контекст мал), фикс идёт сверху.
        try:
            import secrets as _secrets
            esc_id = f"e-{datetime.now(timezone.utc).strftime('%y%m%d%H%M%S')}-{_secrets.token_hex(3)}"
            es.append_event("harness", esc_id, "EscalationRaised", {
                "id": esc_id, "from_level": level_cfg.name,
                "to_level": commands.level_above(level_cfg.name) or "day",
                "reason": f"тик {level_cfg.name} упал: {str(e)[:200]}",
                "context": traceback.format_exc()[-1500:],
                "severity": "warn", "task_id": None, "source": "scheduler",
            })
        except Exception:
            pass  # эскалация — best effort: не маскировать исходную ошибку
        return "error: " + str(e)[:120]
    finally:
        st["busy"] = False


def _reflex_tick(level_cfg, seq: int, model: dict):
    """Секундный рефлекс: телеметрия + алерты + без-LLM команды, без LLM.

    Команды бота (/лестница, /статус, /задачи, /помощь) — мгновенные
    ответы из read model (t-260929210755): мин-уровень на них не
    тратится, ReplySent закрывает входящее сразу после отправки.
    """
    import os
    load = None
    try:
        load = os.getloadavg()[0]
    except Exception:
        pass
    period = datetime.fromtimestamp(seq * level_cfg.every, tz=timezone.utc)
    payload = {"level": "sec", "seq": seq, "load1": load,
               "ts": period.isoformat()}
    # алерт-порог из policy (по умолчанию 8 — как simbiotd)
    thr = (model.get("policy") or {}).get("cpu_alert_threshold", 8)
    if load is not None and load >= thr:
        payload["alert"] = f"load {load:.2f} >= {thr}"
    # без-LLM команды: отвечаем на входящие-команды без ответа
    try:
        from . import tg_commands
        handled = tg_commands.answer_pending(model)
        if handled:
            for mid, cmd in handled:
                _log(f"tg-cmd: {mid} /{cmd} → отвечено (сек-рефлекс)")
            payload["commands"] = [c for _, c in handled]
    except Exception as e:
        # команды — часть рефлекса, но не должны валить телеметрию
        print(f"[tg-cmd] ответ на команды не удался: {e}", flush=True)
    es.append_event("cycle", f"reflex:{seq}", "ReflexObserved", payload)
    return payload, None


# ────────────── АВТОДОСТАВКА (t-260929205927) + СИРОТЫ-РЕСТАРТЫ ──────────────

DELIVERY_POLL = 15  # сек: проход delivery-демона (push уже сделал большую часть)


def orphan_requeue():
    """Сироты рестартов: задачи в running без живого воркера → очередь.

    Раньше: тик-владелец задачи умирал (systemctl restart демона при деплое
    убивает и воркеров), TaskFailed/TaskCompleted не пишется — и задача
    висела в running навсегда (эскалация e-260929234911: cad7af завис дважды
    за два часа; «повторный таймаут часового воркера» был на самом деле
    убийством живого воркера рестартом). Мин-уровень вычищал руками.

    Теперь демон при старте и каждые DELIVERY_POLL секунд:
    - running-задача с мёртвым воркером (нет свежего следа сессии) →
      TaskPromoted(ready) с reason 'orphan-requeue'; флаг await_delivery
      сохраняется, доработка после requeue тоже уйдёт в чат.
    """
    model = es.run_projections()
    from datetime import timedelta
    now = _now()
    requeued = []
    for tid, t in (model.get("tasks") or {}).items():
        if t.get("status") != "running":
            continue
        sess = t.get("session_id") or ""
        level = t.get("level") or "hour"
        cfg = LEVELS.get(level)
        # Порог молчания = таймаут убийства воркера (isolated_runner:
        # max(tick_timeout, every*0.8)) + запас. Пока воркер жив, он пишет
        # след (BudgetSpent/Tool*) каждые секунды-минуты; тишина дольше
        # его же таймаута означает: воркер уже убит раннером (TimeoutExpired
        # НЕ пишет TaskFailed — задача и висела в running) или умер с демоном.
        timeout = (cfg.tick_timeout if cfg else 300) or 300
        if cfg and cfg.uses_llm:
            timeout = max(timeout, int(cfg.every * 0.8))
        timeout += 60
        # свежесть следа: трейс сессии (aggregate_id 'hour-497423:96')
        # или BudgetSpent с её trace_id. session 's-hour-497423' →
        # префикс 'hour-497423' безопасен: след. символ ':' или '"'.
        prefix = sess[2:] if sess.startswith("s-") else sess
        db = es._conn()
        try:
            row = db.execute(
                "SELECT MAX(timestamp) AS ts FROM events "
                "WHERE (aggregate='trace' AND aggregate_id LIKE ?) "
                "   OR (aggregate='harness' AND event_type='BudgetSpent' AND data LIKE ?)",
                (prefix + "%", "%" + prefix + "%"),
            ).fetchone()
            last = row["ts"] if row else None
        finally:
            db.close()
        if last:
            try:
                last_dt = datetime.fromisoformat(last)
                if (now - last_dt) < timedelta(seconds=timeout + 60):
                    continue  # воркер жив/недавно работал — не трогаем
            except ValueError:
                pass
        es.append_event("task", tid, "TaskPromoted",
                        {"to": t.get("size") or "M", "from": t.get("size") or "M",
                         "reason": "orphan-requeue: воркер тика-владельца мёртв "
                                   "(рестарт/таймаут), задача возвращена в очередь"})
        requeued.append(tid)
        _log(f"orphan-requeue: {tid} (сессия {sess}) → ready")
    return requeued


def deliver_pending():
    """Доставка готовых задач с await_delivery, не получивших push.

    Push (в run_level_tick) закрывает 99% кейсов немедленно; этот pull —
    страховка: тик умер после TaskCompleted, но до DeliverySent; флаг
    поставлен позже завершения (старый тик) — досылаем по факту.
    Идемпотентно: delivery_status=='sent' не отправляет повторно.
    """
    from . import commands as _cmd
    model = es.run_projections()
    out = []
    for tid, t in (model.get("tasks") or {}).items():
        if not (t.get("await_delivery") and t.get("status") == "done"
                and t.get("delivery_status") != "sent"):
            continue
        # ответ задачи: TaskCompleted.answer (проекция не хранит — берём из стора)
        db = es._conn()
        try:
            row = db.execute(
                "SELECT data FROM events WHERE aggregate='task' AND aggregate_id=? "
                "AND event_type='TaskCompleted' ORDER BY id DESC LIMIT 1",
                (tid,),
            ).fetchone()
        finally:
            db.close()
        if not row:
            continue
        answer = (json.loads(row["data"]).get("answer") or "").strip()
        try:
            if _cmd.deliver_result(tid, answer):
                out.append(tid)
                _log(f"delivery-pull: {tid} доставлен")
        except Exception as e:
            _log(f"delivery-pull: {tid} ошибка {e} — повторим на след. проходе")
    return out


def _delivery_and_orphans_loop():
    """Фоновый поток демона: доставка + сироты, не мешает тикам уровней."""
    import threading
    _log("delivery-демон запущен (orphan-requeue + автодоставка)")
    # первый проход сразу: чистим сирот рестарта, доим недоставленное
    while True:
        try:
            orphan_requeue()
            deliver_pending()
        except Exception:
            pass  # демон не должен умирать: следующий проход попробует снова
        time.sleep(DELIVERY_POLL)


def main_loop():
    """Демон: уровни тикают независимо, каждый в своём потоке.
    LLM-тик одного уровня больше не блокирует остальные
    (раньше минутный тик замораживал и сек-рефлекс, и час)."""
    import threading

    es.init_db()
    _log("Simbiot Harness запущен; уровни: " + ", ".join(ACTIVE))
    # delivery-демон: сироты рестартов + автодоставка TG (t-260929205927).
    # Первый проход сразу — чистка сирот до первого тика уровней.
    dt = threading.Thread(target=_delivery_and_orphans_loop, daemon=True,
                          name="delivery")
    dt.start()
    last_seq = {}
    threads = {}
    while True:
        for name in ACTIVE:
            cfg = LEVELS[name]
            seq = tick_number(cfg, int(time.time()))
            if last_seq.get(name) == seq:
                continue  # этого такта ещё не было — ждём границу
            last_seq[name] = seq
            if cfg.uses_llm and _runners.get(name, {}).get("busy"):
                # уровень ещё работает в предыдущем такте — тик поглощается,
                # но остальные уровни продолжают жить
                continue
            t = threads.get(name)
            if t is not None and t.is_alive():
                continue
            def _run(cfg=cfg, seq=seq):
                model = es.run_projections()
                state = process_level(cfg, model)
                if state not in ("already",):
                    _log(f"{cfg.name} #{seq}: {state}")
            t = threading.Thread(target=_run, daemon=True, name=f"tick-{name}")
            threads[name] = t
            t.start()
        time.sleep(POLL)


if __name__ == "__main__":
    main_loop()