#!/usr/bin/env python3
"""КАССА ЭКСП-22: потолки фаз + потолок пака + механический фриз-гейт. $0.

Три урока эксп-21, каждый стоил денег, и все три чинятся ЗДЕСЬ, а не дисциплиной:
  1. новый тег покупок не попадал в глоб кассы, и фаза «стоила» ноль — трижды, последний раз
     гейт ЭТО СЕРТИФИЦИРОВАЛ (§17.4). ⇒ теги объявлены в одном месте, и `selftest` проверяет,
     что КАЖДЫЙ тег виден своей кассой и кассой пака;
  2. потолки фаз промта пробивались, потому что касса кода охраняла другие границы. ⇒ каждая
     покупка проходит ДВА гарда: фазовый и пакетный, оба проекционные;
  3. $2.57 из $2.62 потрачены кодом вне зафризенного дерева. ⇒ покупка отказывает, если файл
     вызывающего не отслеживается git или отличается от закоммиченного: «фриз ДО покупок фазы»
     становится механизмом, а не обещанием.

Обход — только `TM_ALLOW_UNFROZEN=1` в окружении: громко, осознанно и видно в истории.
"""
from __future__ import annotations

import json
import os
import subprocess
import time
import sys
from functools import lru_cache
from pathlib import Path

REPO = Path("/home/ubuntu/projects/textmachine")
ZONE = Path(__file__).resolve().parent
sys.path.insert(0, str(REPO / "eval" / "role_topology"))

import buy as BUY                                              # noqa: E402

import roster as ROSTER                                        # noqa: E402

OUT = Path.home() / "books" / "tenant-panel"
OUT.mkdir(parents=True, exist_ok=True)
BUY.OUT = OUT                                                  # сырьё пака — своё, не эксп-21
# ⚠ ПРАЙС-ТАБЛИЦА ТОЖЕ СВОЯ. Поймано исполнением на середине скрина: `buy.priced` звал
# `role_topology/prices.cost`, где нет ни одной новой модели ростера — прогон упал на `kimi-k2.6`
# c KeyError. Хуже тихого падения было бы его отсутствие: у пересекающихся моделей таблица
# эксп-21 ставит кэш-цену Gemini равной входной, то есть цена считалась бы по устаревшему прайсу
# и расходилась с объявленным в пре-реге. Подмена делается ЗДЕСЬ, один раз и явно.
BUY.cost = ROSTER.cost

# Теги покупок по фазам. ⚠ Заводишь новый тег — вписываешь СЮДА тем же коммитом: ровно этого
# не сделали трижды в эксп-21, и расход фазы печатался заниженным.
TAGS = {
    "Ф0": ("tp0-cont-*.json", "tp0-bank-*.json"),
    "Ф1": ("tp1-scr-*.json",),
    "Ф2": ("tp2-draft-*.json", "tp2-edit-*.json", "tp2-floor-*.json"),
    "Ф3": ("tp3-ed-*.json",),
}
# ⚠ ПОТОЛОК Ф1 ПОДНЯТ 08.08 С $1.00 ДО $1.05 СЛОВОМ ВЛАДЕЛЬЦА, объявлено ДО траты. Причина:
# проекционный гард остановил скрин перед `gemini-3.1-pro-preview` — ОБЯЗАТЕЛЬНЫМ кандидатом Ф3
# по промту. Его цена — единственная неизвестная в смете Ф3 (оценка $0.80 из $1.60), и без замера
# половина фазы стояла бы на прайс-странице вместо провода. Пакетный потолок $5.00 не двигался.
# ⚠ ПОТОЛКИ Ф3 И ПАКА ПОДНЯТЫ 08.08 СЛОВОМ ВЛАДЕЛЬЦА, объявлено ДО покупок Ф3: $1.60 → $3.00 и
# $5.00 → $6.50. Причина — замер, а не желание: обязательный кандидат промта
# `gemini-3.1-pro-preview` тарифицирует 11 144–12 010 выходных токенов на ответ в 2 508 (скрытое
# размышление видно ТОЛЬКО в `total_tokens`), то есть $0.152 за редакторскую единицу и $2.44 за
# 16 единиц — больше всего прежнего потолка фазы. Альтернатива «подмножество единиц» владельцем
# отклонена в пользу полной мощности панели.
CEILINGS = {"Ф0": 0.40, "Ф1": 1.05, "Ф2": 2.00, "Ф3": 3.00}
PACK_CEILING = 6.50
GLOB_ALL = "tp*.json"          # по какому шаблону касса видит СВОИ покупки


def configure(**kw) -> None:
    """Пере-настроить кассу под ДРУГОЙ пак, не копируя её.

    Касса — проверенный механизм: фриз-гейт, атомарный замок, проекционный сторож, два пути счёта.
    Копировать её ради нового пака значит копировать и будущие расхождения; поэтому пак-специфика
    (зона кода, каталог сырья, теги, потолки, прайс-таблица) вынесена в параметры. Дефолты
    оставлены эксп-22 побайтно — вызова `configure` в его коде нет, и поведение не меняется.
    Тот же приём уже ратифицирован в судейском риге (`absjudge.configure`).
    """
    g = globals()
    for k, v in kw.items():
        if k not in g:
            raise SystemExit(f"касса: неизвестный параметр configure: {k}")
        g[k] = v
    OUT.mkdir(parents=True, exist_ok=True)
    BUY.OUT = OUT                       # сырьё пака
    BUY.cost = ROSTER.cost              # прайс-таблица пака


class Pack:
    """Пакетный потолок поверх фазового. Дублирует утиный интерфейс `buy.Ledger`."""

    def __init__(self) -> None:
        self.name, self.ceiling = "ПАК", PACK_CEILING
        self.prefixes = tuple(p for ps in TAGS.values() for p in ps)
        self._led = BUY.Ledger("ПАК", PACK_CEILING, self.prefixes, default_expect=0.05)

    def spent(self) -> float:
        return self._led.spent()

    def expect(self) -> float:
        self._led.spent()
        return self._led.expect()

    def afford(self) -> tuple[bool, str]:
        return self._led.afford()


def model_expect(model: str) -> float:
    """Ожидаемая цена следующего вызова ИМЕННО ЭТОЙ модели.

    ⚠ Заведено после того, как потолок Ф1 был пробит ($1.087758 против $1.05). Механизм пробоя:
    проекционный гард брал p95 УЖЕ КУПЛЕННЫХ вызовов — то есть прошлое, в котором новой модели
    ещё не было. `gemini-3.1-pro-preview` оказался в семь раз дороже любого предшественника, и
    оценка «как раньше» его не удержала. Здесь ожидание берётся по СВОИМ записям этой модели,
    а при их отсутствии считается из прайса по медианным токенам пака — консервативно, с
    множителем на скрытое размышление у провайдеров класса `additive_total`.
    """
    mine = [json.loads(f.read_text(encoding="utf-8")).get("cost_usd", 0.0)
            for f in OUT.glob(GLOB_ALL)
            if f.name.count("SELFTEST") == 0]
    prices = [c for c, m in ((json.loads(f.read_text(encoding="utf-8")).get("cost_usd", 0.0),
                              json.loads(f.read_text(encoding="utf-8")).get("model"))
                             for f in OUT.glob(GLOB_ALL)) if m == model and c > 0]
    if len(prices) >= 2:
        return max(prices)
    if model not in ROSTER.CANDIDATES:
        return max(mine or [0.02])
    _, _, pin, _pc, pout, _t, _s = ROSTER.CANDIDATES[model]
    hidden = 5.0 if ROSTER.REASONING_BILLING.get(model) in ("additive_total", "additive") else 1.5
    return 4500 / 1e6 * pin + 3000 * hidden / 1e6 * pout


class Guarded:
    """Фазовая касса, которая не пускает покупку, пробивающую ЛЮБОЙ из двух потолков."""

    def __init__(self, phase: str, default_expect: float = 0.02) -> None:
        self.phase = phase
        self.name = f"{phase} эксп-22"
        self.ceiling = CEILINGS[phase]
        self._led = BUY.Ledger(self.name, self.ceiling, TAGS[phase], default_expect=default_expect)
        self._pack = Pack()
        self.next_model: str | None = None

    def spent(self) -> float:
        return self._led.spent()

    def expect(self) -> float:
        self._led.spent()
        base = self._led.expect()
        return max(base, model_expect(self.next_model)) if self.next_model else base

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


@lru_cache(maxsize=64)
def _git_clean(path: str) -> tuple[bool, str]:
    """Файл отслеживается git и совпадает с закоммиченным? Это и есть проверка фриза."""
    rel = str(Path(path).resolve().relative_to(REPO))
    r = subprocess.run(["git", "-C", str(REPO), "ls-files", "--error-unmatch", rel],  # noqa: S603, S607
                       capture_output=True, text=True)
    if r.returncode:
        return False, f"файл не в git: {rel}"
    d = subprocess.run(["git", "-C", str(REPO), "diff", "--quiet", "HEAD", "--", rel])  # noqa: S603, S607
    if d.returncode:
        return False, f"файл отличается от закоммиченного: {rel}"
    return True, ""


def _refuse_unfrozen() -> None:
    if os.environ.get("TM_ALLOW_UNFROZEN") == "1":
        print("⚠ ФРИЗ-ГЕЙТ ОТКЛЮЧЁН (TM_ALLOW_UNFROZEN=1) — покупка вне фриза")
        return
    import inspect                                             # noqa: PLC0415
    for fr in inspect.stack()[1:]:
        if fr.filename.startswith("<") or not Path(fr.filename).is_file():
            raise SystemExit(f"⛔ покупка из НЕ-ФАЙЛА ({fr.filename}) запрещена")
        p = Path(fr.filename).resolve()
        # кадры САМОЙ кассы пропускаются по файлу, а не по каталогу: настроенная под другой пак
        # касса лежит в зоне эксп-22, а покупает код из зоны того пака, и сверка по каталогу
        # отвергала бы собственный же модуль (поймано до первой покупки нового пака)
        if p == Path(__file__).resolve() or (p.parent == ZONE and p.name == "money.py"):
            continue
        if p.parent != ZONE:
            raise SystemExit(f"⛔ покупка из кода вне зоны эксп-22 запрещена: {p}")
        ok, why = _git_clean(str(p))
        if not ok:
            raise SystemExit(f"⛔ ФРИЗ НЕ ВЗЯТ: {why}\n"
                             "   зафризь фазу коммитом ДО её покупок (промт, §ЗАКОН)")
        return
    raise SystemExit("⛔ вызывающий файл не определён")


# Транзиентные отказы провода, которые лечатся ожиданием, а не диагнозом. Поймано исполнением
# 08.08: `gemini-3.1-pro-preview` отдал HTTP 503 «This model is currently experiencing high
# demand» на ЧЕТЫРЁХ вызовах подряд ($0, клетки потеряны), а через минуту тот же слаг ответил с
# первой попытки; страница статуса Google при этом инцидентов не показывала. Без отступа
# обязательный кандидат Ф3 выпал бы из замера по причине, которая замером не является.
_TRANSIENT = ("503", "502", "504", "overloaded", "high demand", "timeout", "Timeout",
              "RemoteProtocolError", "APIConnectionError")


def _claim(tag: str) -> bool:
    """Атомарно занять клетку. False — её уже покупает другой процесс.

    ⚠ Заведено исполнением: панель покупалась НЕСКОЛЬКИМИ процессами, и проверка «файла ещё нет»
    у `buy.purchase` не атомарна — два воркера могли одновременно пройти её и купить одну клетку
    ДВАЖДЫ. На диске остаётся одна запись, поэтому вторая оплата в леджер не попадает вовсе:
    деньги потрачены, а касса их не видит. Это тот же класс, что «расход фазы мимо глоба», только
    хуже — незаметный по построению.
    """
    try:
        fd = os.open(str(OUT / f"{tag}.lock"), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
    except FileExistsError:
        return False
    os.close(fd)
    return True


def purchase(led: Guarded, tag: str, model: str, client, kw: dict, **extra) -> dict:
    """Один платный вызов. Кэш → фриз-гейт → проекционный гард → покупка (+ отступ на 5xx)."""
    f = OUT / f"{tag}.json"
    if f.exists():
        return json.loads(f.read_text(encoding="utf-8"))
    if not any(tag.startswith(p.split("*")[0]) for p in TAGS[led.phase]):
        raise SystemExit(f"⛔ тег {tag!r} не покрыт глобом кассы {led.phase} — впиши в TAGS")
    _refuse_unfrozen()
    if not _claim(tag):                                        # клетку уже покупает другой процесс
        f = OUT / f"{tag}.json"
        for _ in range(240):
            if f.exists():
                return json.loads(f.read_text(encoding="utf-8"))
            time.sleep(5)
        print(f"    ⚠ замок {tag} протух — покупаю сам")
        (OUT / f"{tag}.lock").unlink(missing_ok=True)
        _claim(tag)
    led.next_model = model                                     # гард считает ожидание по ЭТОЙ модели
    rec = {}
    try:
        for k in range(4):
            rec = BUY.purchase(led, tag, model, client, kw, **extra)
            err = rec.get("error", "") if rec.get("finish") == "error" else ""
            if not err or not any(t in err for t in _TRANSIENT):
                return rec
            wait = 15 * (k + 1)
            print(f"    транзиентный отказ ({err[:60]}) — отступ {wait} с")
            time.sleep(wait)
        return rec
    finally:
        (OUT / f"{tag}.lock").unlink(missing_ok=True)


def report() -> None:
    pack = Pack()
    print(f"{'фаза':6s}{'потрачено':>12s}{'потолок':>10s}  теги")
    for ph in ("Ф0", "Ф1", "Ф2", "Ф3"):
        g = Guarded(ph)
        print(f"{ph:6s}{g.spent():12.6f}{g.ceiling:10.2f}  {' '.join(TAGS[ph])}")
    print(f"{'ПАК':6s}{pack.spent():12.6f}{PACK_CEILING:10.2f}")


def selftest() -> int:
    """Проверка кассы НА СВОИХ ТЕГАХ до покупок: каждый тег виден и фазой, и паком."""
    bad = 0
    # ⚠ ПО ОДНОМУ ТЕГУ ЗА РАЗ. Если писать все пробы сразу, `spent() > 0` у последней кассы
    # доказывается ЧУЖИМ тегом — ровно та слепота глоба, которая в эксп-21 трижды дала
    # заниженный расход фазы.
    base_pack = Pack().spent()
    for ph, globs in TAGS.items():
        base_ph = Guarded(ph).spent()
        for g in globs:
            stem = g.split("*")[0] + "SELFTEST"
            f = OUT / f"{stem}.json"
            f.write_text(json.dumps(dict(tag=stem, cost_usd=0.011)), encoding="utf-8")
            try:
                d_ph = Guarded(ph).spent() - base_ph
                d_pack = Pack().spent() - base_pack
                ok = abs(d_ph - 0.011) < 1e-9 and abs(d_pack - 0.011) < 1e-9
                bad += not ok
                print(f"[{'OK ' if ok else 'ПРОВАЛ'}] тег {stem} даёт +{d_ph:.3f} кассе {ph} "
                      f"и +{d_pack:.3f} кассе пака")
            finally:
                f.unlink()
    # Проекционный гард обязан ОТКАЗЫВАТЬ, когда ожидание пробивает потолок.
    g = Guarded("Ф0")
    g.ceiling = 0.0
    ok, _why = g.afford()
    bad += ok
    print(f"[{'OK ' if not ok else 'ПРОВАЛ'}] проекционный гард отказывает при нулевом потолке")
    # ⚠ РЕГРЕССИЯ НА ПРОБОЙ ПОТОЛКА Ф1: ожидание обязано считаться по ЭТОЙ модели, а не по
    # p95 прошлого. Дорогая новая модель не должна проходить гард только потому, что раньше
    # таких дорогих вызовов не было.
    g2 = Guarded("Ф0")
    g2.ceiling = g2.spent() + 0.05
    g2.next_model = "gemini-3.1-pro-preview"
    ok2, _ = g2.afford()
    bad += ok2
    print(f"[{'OK ' if not ok2 else 'ПРОВАЛ'}] гард видит цену дорогой НОВОЙ модели "
          f"(${model_expect('gemini-3.1-pro-preview'):.4f}), а не p95 прошлого")
    # Прайс: КАЖДАЯ модель ростера обязана считаться кассой, иначе покупка падает на середине.
    for m in ROSTER.CANDIDATES:
        try:
            BUY.cost(m, 1000, 0, 1000, 0, 2000)
        except Exception as e:                                 # noqa: BLE001
            bad += 1
            print(f"[ПРОВАЛ] {m}: касса не умеет считать цену ({type(e).__name__})")
    print(f"[OK ] цена считается для всех {len(ROSTER.CANDIDATES)} моделей ростера"
          if not bad else "")
    # Фриз-гейт: НЕ-файл (встроенный питон) обязан быть отвергнут.
    try:
        exec(compile("import money; money._refuse_unfrozen()", "<string>", "exec"),  # noqa: S102
             {"__name__": "x"})
        print("[ПРОВАЛ] фриз-гейт пропустил покупку из НЕ-файла")
        bad += 1
    except SystemExit:
        print("[OK ] фриз-гейт отвергает покупку из НЕ-файла")
    print(f"\n{'КАССА ГОДНА' if not bad else f'ПРОВАЛОВ: {bad}'}")
    return bad


if __name__ == "__main__":
    a = sys.argv[1:] or ["--report"]
    if a[0] == "--selftest":
        sys.exit(1 if selftest() else 0)
    report()