# -*- coding: utf-8 -*-
"""t-261015000631-13aef0: атомарный клейм — регресс гонки двойного claim.

Инцидент 2026-10-15 00:00: L-задача t-261014221908-d20347 клеймнута ДВАЖДЫ
(TaskClaimed id=2767823 hour 00:00:00 и id=2767828 day 00:00:01) — оба уровня
читали read model до клейма друг друга, а старый цикл писал TaskClaimed по
снапшоту без перепроверки → двойное исполнение = двойной burn 9router.

Проверяет:
  1) es.claim_task_if_ready: ready → успех; повторно/не-ready/несуществующая
     → False; после requeue (TaskPromoted) → снова успех (обратимость);
  2) атомарность СТОРА: 8 потоков параллельно клеймят одну ready-задачу —
     ровно один truthy-результат, ровно одно событие TaskClaimed;
  3) интеграция — точный сценарий прод-инцидента: hour- и day-тик с ОДНИМ
     и тем же устаревшим снапшотом модели (оба видят задачу ready) →
     ровно один TaskClaimed и один TaskCompleted, тик-проигравший
     отрабатывает вхолостую без падения;
  4) свёртка статуса в сторе согласована с проекциями (done не ре-клеймится).

Запуск: ./venv/bin/python tests/test_task_claim_race.py
"""
import os
import sys
import threading
from concurrent.futures import ThreadPoolExecutor

_DB = "/tmp/simbiot-test-claim-race.db"
for _p in (_DB, _DB + "-wal", _DB + "-shm"):
    if os.path.exists(_p):
        os.unlink(_p)
os.environ["SIMBIOT_DB"] = _DB
sys.path.insert(0, "/root/simbiot-harness")

from harness import es
from harness import commands

es.init_db()

fails = []


def check(name, cond, extra=""):
    print(("PASS" if cond else "FAIL"), name, ("" if cond else " | " + str(extra)))
    if not cond:
        fails.append(name)


def events_of(tid, etype=None):
    db = es._conn()
    try:
        if etype:
            rows = db.execute(
                "SELECT * FROM events WHERE aggregate='task' AND aggregate_id=? "
                "AND event_type=?", (tid, etype)).fetchall()
        else:
            rows = db.execute(
                "SELECT * FROM events WHERE aggregate='task' AND aggregate_id=? "
                "ORDER BY version", (tid,)).fetchall()
        return [dict(r) for r in rows]
    finally:
        db.close()


def status_of(tid):
    t = (es.run_projections().get("tasks") or {}).get(tid)
    return (t or {}).get("status")


# ── 1. Семантика claim_task_if_ready ──────────────────────────────────────
tid, _ = commands.submit_task(title="гонка клейма: атомарный claim",
                              minutes=60, details="фикс t-261015000631")
check("1.1: ready → клейм успешен", bool(es.claim_task_if_ready(
    tid, {"level": "hour", "session_id": "s-hour-1"})))
check("1.2: running → клейм ОТКАЗАН (второй уровень не клеймнет)",
      es.claim_task_if_ready(tid, {"level": "day", "session_id": "s-day-1"}) is False)
check("1.3: событий TaskClaimed ровно одно",
      len(events_of(tid, "TaskClaimed")) == 1, str(len(events_of(tid, "TaskClaimed"))))
check("1.4: несуществующая задача → False",
      es.claim_task_if_ready("t-nope-000000", {}) is False)

# закрыли → клейм невозможен; requeue вернул в очередь → снова возможен
es.append_event("task", tid, "TaskCompleted", {"answer": "done"})
check("1.5: done → клейм ОТКАЗАН",
      es.claim_task_if_ready(tid, {"level": "hour", "session_id": "s-hour-2"}) is False)
commands.requeue_task(tid, reason="тест обратимости")
check("1.6: после requeue (TaskPromoted → ready) клейм снова успешен",
      bool(es.claim_task_if_ready(tid, {"level": "min", "session_id": "s-min-1"})))
check("1.7: свёртка стора == проекции (running)",
      status_of(tid) == "running", status_of(tid))

# ── 2. Атомарность стора: 8 потоков, одна ready-задача ────────────────────
tid2, _ = commands.submit_task(title="гонка клейма: 8 потоков",
                               minutes=60, details="один победитель")
barrier = threading.Barrier(8)


def _claim(_i):
    barrier.wait()  # максимальная конкуренция на одной ready-задаче
    return es.claim_task_if_ready(tid2, {"level": "race", "session_id": f"s-{_i}"})


with ThreadPoolExecutor(max_workers=8) as ex:
    results = list(ex.map(_claim, range(8)))
winners = [r for r in results if r]
losers = [r for r in results if r is False]
check("2.1: ровно ОДИН победитель из 8 параллельных", len(winners) == 1, str(results))
check("2.2: остальные 7 — чистый False (не None-дубль)",
      len(losers) == 7, str(results))
check("2.3: TaskClaimed в сторе ровно один",
      len(events_of(tid2, "TaskClaimed")) == 1,
      str(len(events_of(tid2, "TaskClaimed"))))
check("2.4: проекция видит running", status_of(tid2) == "running", status_of(tid2))

# ── 3. Интеграция: прод-сценарий hour+day со stale-снапшотом ───────────────
from harness import scheduler
from harness.levels import LEVELS
import harness.isolated_runner as _ir

tid3, _ = commands.submit_task(title="гонка клейма: hour vs day",
                               minutes=60, details="сценарий 2026-10-15 00:00")
# ОДИН снапшот модели для обоих тиков — как в проде: оба уровня читали модель
# ДО клейма друг друга. Снапшот устаревает сразу после первого клейма —
# именно это раньше и рождало второй TaskClaimed.
stale_model = es.run_projections()
check("3.1: подготовка — в снапшоте задача ready",
      (stale_model["tasks"].get(tid3) or {}).get("status") == "ready")

_orig_run_llm_tick = _ir.run_llm_tick


def _fake_run_llm_tick(level_cfg, prompt, session_id=None, trace_prefix=None, timeout=None):
    return (f"дайджест {level_cfg.name}: отработана пачка",
            {"model_calls": 1}, f"trace-{level_cfg.name}")


ticks_out = {}
tick_err = {}


def _run_tick(level_name, seq):
    try:
        # fresh-снапшот имитировать НЕ будем: оба тика получают ОДИН stale —
        # худший случай. Внутри тика claim пойдёт в стор атомарно.
        ticks_out[level_name] = scheduler.run_level_tick(
            LEVELS[level_name], seq, stale_model)
    except Exception as e:  # noqa: BLE001
        tick_err[level_name] = repr(e)


try:
    _ir.run_llm_tick = staticmethod(_fake_run_llm_tick)
    barrier2 = threading.Barrier(2)

    def _run(level_name, seq):
        barrier2.wait()
        _run_tick(level_name, seq)

    th_hour = threading.Thread(target=_run, args=("hour", 4242))
    th_day = threading.Thread(target=_run, args=("day", 4243))
    th_hour.start(), th_day.start()
    th_hour.join(), th_day.join()
finally:
    _ir.run_llm_tick = _orig_run_llm_tick

check("3.2: оба тика завершились без исключений", not tick_err, str(tick_err))
claims3 = events_of(tid3, "TaskClaimed")
comps3 = events_of(tid3, "TaskCompleted")
check("3.3: TaskClaimed ровно ОДИН (регресс: было два)",
      len(claims3) == 1, f"n={len(claims3)}")
check("3.4: TaskCompleted ровно ОДИН (исполнение не задвоилось)",
      len(comps3) == 1, f"n={len(comps3)}")
winner_level = (claims3[0]["data"] if claims3 else "").find('"level"') >= 0
check("3.5: клейм содержит level победителя", bool(winner_level),
      claims3[0]["data"][:120] if claims3 else "нет событий")
loser = [lv for lv in ("hour", "day")
         if lv != __import__("json").loads(claims3[0]["data"])["level"]][0] \
    if claims3 else "?"
check("3.6: проигравший тик отработал вхолостую (пустой батч, без падения)",
      loser in ticks_out and ticks_out[loser][1] in ([], None),
      str({k: v[1] for k, v in ticks_out.items()}))
check("3.7: финальный статус задачи done", status_of(tid3) == "done", status_of(tid3))

print()
print("FAILS:", len(fails))
sys.exit(1 if fails else 0)