"""Единая касса пака: леджер с диска, ПРОЕКЦИОННЫЙ гард, однородная запись вызова.

⚠ Заведён 07.08 по адверсариальному ревью, которое нашло три разных дефекта одного класса:
  * `repair_arm.CEILING_USD` и `route_arm.CEILING_USD` были ОБЪЯВЛЕНЫ и не использованы нигде —
    потолок Ф2б $3.00 был декоративным, а счётчик трат локальным и перезапуск не переживал;
  * гард `screen.call` сравнивал УЖЕ потраченное с порогом и ничего не знал о цене следующего
    вызова: запас $0.03 при наблюдённом максимуме одного вызова $0.090 — потолок держался
    случаем, а не механизмом;
  * `bakeoff` считал потраченное по глобу `bo-` и не видел $1.02, ушедших на судейство.

Отсюда три правила, и все три живут ЗДЕСЬ, а не в каждом скрипте по-своему:
 1. Потраченное читается С ДИСКА по всем префиксам фазы. Память процесса не источник истины.
 2. Гард ПРОЕКЦИОННЫЙ: `spent + ожидаемая цена` против потолка, а не `spent` против порога.
    Ожидание берётся из уже купленных вызовов той же роли (p95), при их отсутствии — из сметы.
 3. Запись несёт `total_tokens`. Без него биллинг Gemini (`additive_total`, quirks 00 D22.3)
    не пере-считывается даже постфактум — ровно это и случилось с прошлым прогоном.
"""
from __future__ import annotations

import json
import time
import os
from pathlib import Path

from prices import cost

OUT = Path.home() / "books" / "role-topology"
OUT.mkdir(parents=True, exist_ok=True)


class Ledger:
    """Касса одной фазы. `prefixes` — глобы артефактов, из которых складывается потраченное."""

    def __init__(self, name: str, ceiling: float, prefixes: tuple[str, ...],
                 default_expect: float = 0.05) -> None:
        self.name, self.ceiling, self.prefixes = name, ceiling, prefixes
        self.default_expect = default_expect
        self._prices: list[float] = []

    def spent(self) -> float:
        tot, self._prices = 0.0, []
        for pat in self.prefixes:
            for f in OUT.glob(pat):
                try:
                    r = json.loads(f.read_text(encoding="utf-8"))
                except Exception:                            # noqa: BLE001, S112
                    continue
                c = float(r.get("cost_usd") or 0.0)
                tot += c
                if c > 0:
                    self._prices.append(c)
        return tot

    def expect(self) -> float:
        """Ожидаемая цена следующего вызова: p95 уже купленного, иначе смета."""
        if len(self._prices) < 3:
            return self.default_expect
        s = sorted(self._prices)
        return s[min(len(s) - 1, int(round(0.95 * (len(s) - 1))))]

    def afford(self) -> tuple[bool, str]:
        sp = self.spent()
        proj = sp + self.expect()
        if proj > self.ceiling:
            return False, (f"[СТОП {self.name}] потрачено ${sp:.6f} + ожидание ${self.expect():.6f} "
                           f"= ${proj:.6f} > потолок ${self.ceiling:.2f}")
        return True, ""


def usage_of(u, ch) -> dict:
    """Токены вызова в однородном виде. `total_tokens` персистится ВСЕГДА (см. шапку).

    `ch` может быть None — это ответ без сообщения (fail-closed фильтра); токены при этом
    провайдер обычно всё равно тарифицирует, поэтому usage читается, а не обнуляется.
    """
    cached = getattr(getattr(u, "prompt_tokens_details", None), "cached_tokens", 0) or 0
    reasoning = getattr(getattr(u, "completion_tokens_details", None), "reasoning_tokens", 0) or 0
    msg = getattr(ch, "message", None) if ch is not None else None
    return dict(prompt_tokens=u.prompt_tokens or 0, cached_tokens=cached,
                completion_tokens=u.completion_tokens or 0, reasoning_tokens=reasoning,
                total_tokens=getattr(u, "total_tokens", 0) or 0,
                reasoning_chars=len(getattr(msg, "reasoning_content", None) or ""),
                # ⚠ САМ ТРЕЙС РАЗМЫШЛЕНИЯ — ТОЛЬКО ПО ЯВНОМУ ФЛАГУ, и это не осторожность, а вес:
                # у dspro он бывает 30+ тыс. знаков на клетку, то есть удваивает файл сырья, а
                # нужен он редко. Провайдер отдаёт его в ответе БЕСПЛАТНО, но `usage_of` до 20.08
                # сохранял одну ДЛИНУ — и когда замер промта дал ×28 размышления, прочитать,
                # на что оно потрачено, оказалось нечем: пришлось бы пере-покупать клетки.
                # Флаг `TM_KEEP_TRACE=1` включает сохранение для абляционных прогонов.
                **({"reasoning_text": getattr(msg, "reasoning_content", None) or ""}
                   if os.environ.get("TM_KEEP_TRACE") == "1" else {}))


def priced(model: str, us: dict) -> float:
    return round(cost(model, us["prompt_tokens"], us["cached_tokens"], us["completion_tokens"],
                      us["reasoning_tokens"], us["total_tokens"]), 6)


ZONE = Path(__file__).resolve().parent
# ⚠ Расширено эксп-22 (08.08): гард сравнивал каталог с `role_topology` и тем самым запрещал
# покупки ЛЮБОМУ другому паку полигона, хотя его смысл — «код в репозитории, а не временный
# скрипт». Зона полигона по CLAUDE.md — весь `eval/`, поэтому граница проведена здесь.
# Проверка «НЕ-файл» (встроенный питон из командной строки) — та, из-за которой гард заведён, —
# не тронута; эксп-22 поверх этого требует ещё и git-чистоты вызывающего файла (`tenant_panel/money.py`).
POLYGON = ZONE.parent


def _refuse_unfrozen_caller() -> None:
    """Отказать в ПОКУПКЕ, если вызов пришёл из кода вне зоны полигона.

    ⚠ Заведено 07.08 по собственной находке, повторённой ТРИЖДЫ: покупки делались скриптами из
    временного каталога сессии и встроенным питоном из командной строки. Такие траты не
    воспроизводимы из репозитория и не попадают под пре-рег-фриз — это ровно тот дефект, за
    который снята первая редакция («$2.57 из $2.62 потрачены незафиксированным кодом»).
    Дисциплина словами не держится: за один пак я нарушил её три раза. Теперь держится кассой.

    Проверяется ФАЙЛ ВЫЗЫВАЮЩЕГО, а не текущий: кэш и своды из любого места читать можно,
    отказ касается только денег. Обход намеренно оставлен явным и громким — TM_ALLOW_UNFROZEN=1
    в окружении, чтобы разовая отладка требовала осознанного действия и была видна в истории.
    """
    import inspect                                            # noqa: PLC0415
    import os                                                 # noqa: PLC0415
    if os.environ.get("TM_ALLOW_UNFROZEN") == "1":
        return
    for fr in inspect.stack()[1:]:
        # ⚠ НЕ-ФАЙЛ проверяется ПЕРВЫМ, до сравнения каталога. Первая версия сравнивала каталог
        # раньше, и `Path("<string>").resolve()` при запуске ИЗ каталога зоны давала parent == ZONE,
        # то есть `python -c` из самого каталога пака проходил гард насквозь — именно тот способ
        # запуска, ради которого гард и заведён. Поймано адверсариальным ревью исполнением 07.08.
        if fr.filename.startswith("<") or not Path(fr.filename).is_file():
            raise SystemExit(f"⛔ покупка из НЕ-ФАЙЛА ({fr.filename}) запрещена: "
                             "встроенный питон из командной строки не воспроизводим из репозитория")
        p = Path(fr.filename).resolve()
        if p.suffix != ".py":
            raise SystemExit(f"⛔ покупка из НЕ-ФАЙЛА ({fr.filename}) запрещена")
        if POLYGON in p.parents:                              # код полигона в цепочке — пропускаем
            continue
        raise SystemExit(f"⛔ покупка из кода вне зоны полигона запрещена: {p}\n"
                         f"   перенеси скрипт в {POLYGON} и закоммить пре-рег-фризом")
    return


def purchase(led: Ledger, tag: str, model: str, client, kw: dict, **extra) -> dict:
    """Один платный вызов под кассой. Кэш читается ПЕРВЫМ, гард — перед покупкой.

    Порядок важен и выведен из двух прошлых ошибок: гард перед чтением кэша объявлял уже
    купленную клетку незамеренной, а гард после покупки не защищал ничего.
    """
    f = OUT / f"{tag}.json"
    if f.exists():
        return json.loads(f.read_text(encoding="utf-8"))
    _refuse_unfrozen_caller()
    ok, why = led.afford()
    if not ok:
        print(why)
        return dict(tag=tag, model=model, skipped=True, cost_usd=0.0, content="", finish="skipped")
    t0 = time.time()
    try:
        r = client.chat.completions.create(**kw)
    except Exception as e:                                   # noqa: BLE001
        # Провал ОДНОГО вызова — данные, а не повод ронять прогон. ⚠ Но запись отказа кладётся
        # ОТДЕЛЬНЫМ тегом: прошлая редакция писала её под тем же именем, и транзиентный таймаут
        # навсегда отравлял клетку — пере-запуск читал ошибку из кэша и вызов не повторял.
        rec = dict(tag=tag, model=model, finish="error", error=f"{type(e).__name__}: {str(e)[:200]}",
                   content="", cost_usd=0.0, latency_s=round(time.time() - t0, 1), **extra)
        (OUT / f"{tag}.ERROR.json").write_text(json.dumps(rec, ensure_ascii=False, indent=1),
                                               encoding="utf-8")
        return rec
    ch = r.choices[0] if r.choices else None
    # ⚠ ОТВЕТ БЕЗ СООБЩЕНИЯ. Поймано исполнением 07.08: `ch.message` пришёл None, и прогон упал
    # на 130 неоплаченных клетках. Это не «пустой content», а отсутствие самого объекта сообщения —
    # так выглядит fail-closed фильтра (quirks 00: у Gemini `finish_reason` приходит СОСТАВНОЙ
    # строкой `'content_filter: PROHIBITED_CONTENT'`, точные матчеры на неё мертвы).
    # Клетка записывается как ДАННЫЕ (отказ провода с причиной), а не роняет фазу: провал одного
    # вызова — это факт о модели на этом материале, и он нужен в отчёте.
    if ch is None or getattr(ch, "message", None) is None:
        rec = dict(tag=tag, model=model, model_returned=getattr(r, "model", ""),
                   finish=str(getattr(ch, "finish_reason", "no-choice")), content="",
                   no_message=True, latency_s=round(time.time() - t0, 1),
                   cost_usd=priced(model, usage_of(r.usage, None)) if r.usage else 0.0,
                   **(usage_of(r.usage, None) if r.usage else {}), **extra)
        (OUT / f"{tag}.NOMSG.json").write_text(json.dumps(rec, ensure_ascii=False, indent=1),
                                               encoding="utf-8")
        print(f"    ⚠ ответ без сообщения: {tag} finish={rec['finish']}")
        return rec
    us = usage_of(r.usage, ch)
    rec = dict(tag=tag, model=model, model_returned=r.model, finish=ch.finish_reason,
               content=ch.message.content or "", latency_s=round(time.time() - t0, 1),
               cost_usd=priced(model, us), **us, **extra)
    f.write_text(json.dumps(rec, ensure_ascii=False, indent=1), encoding="utf-8")
    return rec