# -*- coding: utf-8 -*-
"""t-261002205233-7b65b1: индикация взятия входящих TG в работу, hermetic.

Запуск: ./venv/bin/python tests/test_seen_indication.py

Проверяет:
- set_reaction шлёт setMessageReaction с корректным reaction (JSON-массив);
- mark_seen: реакция + MessageSeen (проекция seen=True), повтор — skip
  (нет ни реакции, ни второго события), TG-сбой — False без события;
- receive_message сохраняет message_id (нативный, не хэш-фолбэк);
- typing_keepalive: держит sendChatAction typing и останавливается по Event;
- run_level_tick: поднимает keepalive по TG-чатам инбокса ВЕСЬ тик, без
  привязки к pending/seen (t-261010092931), исключает портал-чаты, при
  пустом инбоксе — фолбэк на чат харнеса; гасит поток в finally (любой исход).
"""
import json
import os
import sys
import threading
import time

_DST = "/tmp/simbiot-test-seen-indication.db"
for p in (_DST, _DST + "-wal", _DST + "-shm"):
    if os.path.exists(p):
        os.unlink(p)
os.environ["SIMBIOT_DB"] = _DST

sys.path.insert(0, "/root/simbiot-harness")
from harness import es, tg_inbox
from harness.scheduler import run_level_tick
from harness.levels import LEVELS

fails = []


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


es.init_db()

# ── 1. set_reaction: setMessageReaction с JSON reaction (массив) ──
calls = []
tg_inbox._api = lambda method, params, timeout=35: calls.append((method, params)) or {"ok": True}
tg_inbox.set_reaction(42, 777, emoji="👀")
check("set_reaction: метод setMessageReaction", calls and calls[0][0] == "setMessageReaction")
# Bot API: reaction — массив ReactionType (было reaction_type одиночным
# объектом: Telegram видел пустую реакцию, 👀 не появлялся — аудит 2026-10-17)
rt = json.loads(calls[0][1]["reaction"])
check("set_reaction: reaction = [emoji 👀]",
      rt == [{"type": "emoji", "emoji": "👀"}] and calls[0][1]["message_id"] == 777, str(calls))

# ── 2. receive_message хранит message_id ──
tg_inbox.receive_message({"update_id": 516173900, "message": {
    "message_id": 777, "date": 1780000000, "chat": {"id": 42},
    "from": {"id": 1, "username": "alex", "first_name": "Alex"}, "text": "привет"}})
m = es.run_projections()["inbox"]["m-516173900"]
check("receive_message: message_id в проекции", m.get("message_id") == 777, str(m.get("message_id")))
check("receive_message: seen по умолчанию False", m.get("seen") is False)

# ── 3. mark_seen: реакция + MessageSeen, повтор — skip ──
calls.clear()
before = es.count_events()
ok = tg_inbox.mark_seen("m-516173900")
check("mark_seen: True при постановке", ok is True)
check("mark_seen: setMessageReaction отправлен",
      any(c[0] == "setMessageReaction" for c in calls), str([c[0] for c in calls]))
check("mark_seen: MessageSeen записан",
      es.count_events() - before == 1, str(es.count_events() - before))
check("mark_seen: проекция seen=True",
      es.run_projections()["inbox"]["m-516173900"].get("seen") is True)
check("mark_seen: seen_emoji 👀",
      es.run_projections()["inbox"]["m-516173900"].get("seen_emoji") == "👀")
# повтор: флаг seen → ни реакции, ни события (идемпотентность)
calls.clear()
before = es.count_events()
ok2 = tg_inbox.mark_seen("m-516173900")
check("mark_seen: повтор skip (реакции нет)", ok2 is False and calls == [], str(calls))
check("mark_seen: повтор без нового события", es.count_events() == before)

# ── 4. TG-сбой: False, БЕЗ MessageSeen (флаг не подаём) ──
tg_inbox.receive_message({"update_id": 516173901, "message": {
    "message_id": 778, "date": 1780000000, "chat": {"id": 42},
    "from": {"id": 1, "username": "alex"}, "text": "второе"}})

def _boom(method, params, timeout=35):
    raise ConnectionError("tg down")
tg_inbox._api = _boom
before = es.count_events()
ok = tg_inbox.mark_seen("m-516173901")
check("mark_seen: TG-сбой → False", ok is False)
check("mark_seen: TG-сбой без MessageSeen (retry позже)",
      es.count_events() == before and
      es.run_projections()["inbox"]["m-516173901"].get("seen") is not True)

# ── 5. typing_keepalive: держит typing, останавливается по Event ──
typing_calls, reaction_calls = [], []
def _api_counter(method, params, timeout=35):
    (typing_calls if method == "sendChatAction" else reaction_calls).append(params)
    return {"ok": True}
tg_inbox._api = _api_counter
stop = threading.Event()
th = threading.Thread(target=tg_inbox.typing_keepalive, args=([42], stop),
                      kwargs={"interval": 0.5}, daemon=True)
th.start()
time.sleep(1.6)
stop.set()
th.join(timeout=3)
check("typing_keepalive: sendChatAction держится (>2 за 1.6с)",
      len(typing_calls) >= 2, str(len(typing_calls)))
check("typing_keepalive: только typing, реакций нет",
      reaction_calls == [] and all(p.get("action") == "typing" for p in typing_calls))
check("typing_keepalive: останов по Event (join < 1с)",
      not th.is_alive(), "поток жив после stop.set()")

# ── 6. run_level_tick: keepalive по TG-чатам, портал исключён, stop в finally ──
import harness.isolated_runner as _ir_module

from harness import scheduler
_orig_es = scheduler.es
_orig_tg_typing_backup = tg_inbox.typing_keepalive
ka_seen = {}
def _fake_keepalive(chat_ids, stop, interval=2.0):
    ka_seen["ids"] = list(chat_ids)
    ka_seen["stop"] = stop
tg_inbox.typing_keepalive = _fake_keepalive

written = []
class _FakeES:
    def __getattr__(self, name):
        def _stub(*a, **k):
            return None
        return _stub
    def append_event(self, aggregate, aid, evtype, data={}):
        written.append((evtype, aid))
        return "x"

    # t-261015000631-13aef0: условный клейм в стабе всегда успешен
    def claim_task_if_ready(self, task_id, claim):
        written.append(("TaskClaimed", task_id))
        return ("x", 1)

class _FakeRunner:
    @staticmethod
    def run_llm_tick(level_cfg, prompt, session_id=None, trace_prefix=None, timeout=None,
                     circle=None):
        return "дайджест", {"model_calls": 1}, "trace-x"

model = {
    "tasks": {"t-x": {"id": "t-x", "title": "задача", "size": "S", "status": "ready",
                      "created": "2026-10-02T20:00:00", "details": ""}},
    "digests": {},
    "inbox": {
        "m-1": {"update_id": 1, "chat_id": 42, "text": "вопрос", "kind": "text",
                "replied": False, "ignored": False, "ts": "2026-10-02T20:01:00+00:00"},
        "m-2": {"update_id": 2, "chat_id": "portal:web", "text": "портал", "kind": "text",
                "replied": False, "ignored": False, "ts": "2026-10-02T20:02:00+00:00"},
    },
    "escalations": {}, "policy": {},
}
try:
    scheduler.es = _FakeES()
    _ir_module.run_llm_tick = staticmethod(_FakeRunner.run_llm_tick)
    run_level_tick(LEVELS["hour"], 4242, model)
    check("tick: keepalive поднят ровно по TG-чату (портал исключён)",
          ka_seen.get("ids") == [42], str(ka_seen))
    check("tick: finally погасил keepalive (stop set)",
          ka_seen.get("stop") is not None and ka_seen.get("stop").is_set())

    # worker умер → finally всё равно гасит поток
    ka_seen.clear()
    class _DeadRunner:
        @staticmethod
        def run_llm_tick(*a, **k):
            raise RuntimeError("worker dead")
    _ir_module.run_llm_tick = staticmethod(_DeadRunner.run_llm_tick)
    try:
        scheduler.run_level_tick(LEVELS["hour"], 4243, model)
    except RuntimeError:
        pass
    check("tick: worker упал → finally всё равно гасит keepalive",
          ka_seen.get("stop") is not None and ka_seen.get("stop").is_set())

    # t-261010092931: всё seen → «печатает…» всё равно горит весь тик
    # (индикатор больше не зависит от pending: Alex видит «печатает…»
    # всё время, пока LLM-тик работает)
    ka_seen.clear()
    model2 = dict(model)
    model2["inbox"] = {
        "m-1": {**model["inbox"]["m-1"], "seen": True},
        "m-2": {**model["inbox"]["m-2"], "seen": True},
    }
    _ir_module.run_llm_tick = staticmethod(_FakeRunner.run_llm_tick)
    run_level_tick(LEVELS["hour"], 4244, model2)
    check("tick: всё seen → keepalive всё равно поднят по TG-чату",
          ka_seen.get("ids") == [42], str(ka_seen))
    check("tick: всё seen → finally погасил keepalive",
          ka_seen.get("stop") is not None and ka_seen.get("stop").is_set())

    # t-261010092931: пустой инбокс → фолбэк на последний живой TG-чат
    # (_default_chat_id, личка Alex): индикатор горит и без входящих
    ka_seen.clear()
    model3 = dict(model)
    model3["inbox"] = {}
    _orig_default_chat = tg_inbox._default_chat_id
    tg_inbox._default_chat_id = lambda: 42
    try:
        _ir_module.run_llm_tick = staticmethod(_FakeRunner.run_llm_tick)
        run_level_tick(LEVELS["hour"], 4245, model3)
    finally:
        tg_inbox._default_chat_id = _orig_default_chat
    check("tick: пустой инбокс → keepalive по чату харнеса (фолбэк)",
          ka_seen.get("ids") == [42], str(ka_seen))
    check("tick: пустой инбокс → finally погасил keepalive",
          ka_seen.get("stop") is not None and ka_seen.get("stop").is_set())

    # инбокс без TG-чатов (только портал/tui) → тоже фолбэк на чат харнеса
    ka_seen.clear()
    model4 = dict(model)
    model4["inbox"] = {"m-2": model["inbox"]["m-2"]}
    tg_inbox._default_chat_id = lambda: 42
    try:
        _ir_module.run_llm_tick = staticmethod(_FakeRunner.run_llm_tick)
        run_level_tick(LEVELS["hour"], 4246, model4)
    finally:
        tg_inbox._default_chat_id = _orig_default_chat
    check("tick: только портал-чаты → keepalive по фолбэку, портал исключён",
          ka_seen.get("ids") == [42], str(ka_seen))

    # фолбэк недоступен (нет известного чата) → keepalive не поднимается
    ka_seen.clear()
    tg_inbox._default_chat_id = lambda: None
    try:
        _ir_module.run_llm_tick = staticmethod(_FakeRunner.run_llm_tick)
        run_level_tick(LEVELS["hour"], 4247, model3)
    finally:
        tg_inbox._default_chat_id = _orig_default_chat
    check("tick: чат неизвестен → keepalive не поднимается", ka_seen == {}, str(ka_seen))
finally:
    scheduler.es = _orig_es
    tg_inbox.typing_keepalive = _orig_tg_typing_backup

print()
print("FAILURES:", fails if fails else "нет — все проверки пройдены")
sys.exit(1 if fails else 0)