"""Budget-Stopp (hart) und Infra-Fail-closed (Lauf-Pause statt Teilergebnis)."""

import asyncio

import agents
import ledger
import llm
import pytest
from conftest import run_anlegen, topic_anlegen


async def test_budget_stoppt_hart(monkeypatch):
    topic = topic_anlegen()
    run = run_anlegen(topic, budget=120)  # Fake-Call kostet 150 Tokens
    ctx = llm.Kontext(run, topic, "minimax")
    ctx.ebene = "test"
    with pytest.raises(ledger.BudgetErschoepft):
        await llm.call(ctx, stage="soll", template="Korpus-Soll",
                       werte={"topic": topic, "quelle": "q", "text": "x"}, erwartet=list)
    assert ledger.verbraucht(run) >= 120  # der Verbrauch bleibt sichtbar


async def test_infra_pause_statt_fail_open(monkeypatch):
    topic = topic_anlegen()
    run = run_anlegen(topic)
    ctx = llm.Kontext(run, topic, "minimax")
    ctx.ebene = "test"

    async def immer_429(key, prompt, timeout, **kw):
        return agents.AgentErgebnis(1, "", "HTTP 429: rate limited")

    monkeypatch.setattr(agents, "run_agent", immer_429)
    monkeypatch.setattr(llm, "INFRA_BACKOFF_BASE", 0.01)
    with pytest.raises(llm.LaufPause):
        await llm.call(ctx, stage="soll", template="Korpus-Soll",
                       werte={"topic": topic, "quelle": "q", "text": "x"}, erwartet=list)
    stati = [e["status"] for e in
             __import__("db").query("SELECT status FROM events WHERE run_id=?", (run,))]
    assert stati and all(s == "infra" for s in stati)  # jeder Versuch im Ledger


async def test_budget_vorab_stoppt_ohne_neuen_call(monkeypatch):
    """Erschöpftes Budget → nächster call() wirft, BEVOR ein Agent-Call startet."""
    import db
    topic = topic_anlegen()
    run = run_anlegen(topic, budget=100)
    ctx = llm.Kontext(run, topic, "minimax")
    ctx.ebene = "test"
    # Budget künstlich überschreiten (ohne Agent-Call), Event-Zahl merken
    ledger.log_call(run, ebene="test", stage="seed", status="ok",
                    tokens={"input": 200, "output": 0})
    vorher = db.one("SELECT COUNT(*) AS n FROM events WHERE run_id=?", (run,))["n"]

    async def darf_nicht_laufen(key, prompt, timeout, **kw):
        raise AssertionError("Agent-Call trotz erschöpftem Budget gestartet")

    monkeypatch.setattr(agents, "run_agent", darf_nicht_laufen)
    with pytest.raises(ledger.BudgetErschoepft):
        await llm.call(ctx, stage="soll", template="Korpus-Soll",
                       werte={"topic": topic, "quelle": "q", "text": "x"}, erwartet=list)
    nachher = db.one("SELECT COUNT(*) AS n FROM events WHERE run_id=?", (run,))["n"]
    assert nachher == vorher  # keine neue Ledger-Zeile = kein Call


async def test_hedge_gewinner_trotz_haupt_timeout(monkeypatch):
    """Haupt-Call läuft in Timeout, Zwilling (-h) liefert → Ergebnis statt Pause."""
    topic = topic_anlegen()
    run = run_anlegen(topic)
    ctx = llm.Kontext(run, topic, "minimax")
    ctx.ebene = "test"
    monkeypatch.setattr(llm, "HEDGE_NACH_S", 0.05)

    async def haupt_stallt(key, prompt, timeout, **kw):
        if key.endswith("-h"):  # Zwilling liefert gültiges JSON
            return agents.AgentErgebnis(0, "[]", "")
        await asyncio.sleep(10)  # Haupt hängt → wird nach Timeout gecancelt
        return agents.AgentErgebnis(0, "[]", "")

    monkeypatch.setattr(agents, "run_agent", haupt_stallt)
    # timeout klein halten, damit der Haupt-Task zügig als Timeout endet
    monkeypatch.setattr(llm, "timeout_fuer", lambda *a, **k: 0.2)
    res = await llm.call(ctx, stage="soll", template="Korpus-Soll",
                         werte={"topic": topic, "quelle": "q", "text": "x"}, erwartet=list)
    assert res == []  # Zwilling hat geliefert, kein LaufPause


async def test_hedge_verlierer_zaehlt_im_budget(monkeypatch):
    """Beide Hedge-Calls liefern → zwei Ledger-Zeilen, Verlierer-Tokens im Budget."""
    import db
    topic = topic_anlegen()
    run = run_anlegen(topic)
    ctx = llm.Kontext(run, topic, "minimax")
    ctx.ebene = "test"
    monkeypatch.setattr(llm, "HEDGE_NACH_S", 0.05)
    monkeypatch.setattr(llm, "timeout_fuer", lambda *a, **k: 0.4)  # schwelle=0.2

    async def haupt_leer_zwilling_ok(key, prompt, timeout, **kw):
        if key.endswith("-h"):  # Zwilling startet bei 0.2, liefert bei ~0.35
            await asyncio.sleep(0.15)
            return agents.AgentErgebnis(0, "[]", "")
        await asyncio.sleep(0.25)  # Haupt: fertig, aber leer (nicht ok) → Verlierer
        return agents.AgentErgebnis(1, "", "leer", {"input": 40, "output": 20})

    monkeypatch.setattr(agents, "run_agent", haupt_leer_zwilling_ok)
    res = await llm.call(ctx, stage="soll", template="Korpus-Soll",
                         werte={"topic": topic, "quelle": "q", "text": "x"}, erwartet=list)
    assert res == []
    zeilen = db.query("SELECT status FROM events WHERE run_id=?", (run,))
    assert any(z["status"] == "hedge" for z in zeilen)  # Verlierer geloggt
    assert ledger.verbraucht(run) >= 60  # Verlierer-Tokens (40+20) sichtbar


async def test_spawn_kill_bei_cancel(monkeypatch):
    """Task-Cancel (Hedge-Verlierer/Pause) killt den CLI-Prozess, kein Zombie."""
    import os
    monkeypatch.delenv("CREATOR_FAKE_AGENTS", raising=False)
    monkeypatch.setattr(agents, "_ram_gate", lambda key: _true())
    task = asyncio.ensure_future(
        agents._spawn("t-cancel", ["sleep", "30"], None, 30))
    for _ in range(200):  # warten bis der Prozess registriert ist
        if agents._prozesse:
            break
        await asyncio.sleep(0.01)
    proc = next(iter(agents._prozesse.values()))
    pid = proc.pid
    task.cancel()
    with pytest.raises(asyncio.CancelledError):
        await task
    for _ in range(200):  # Kill ist async
        try:
            os.kill(pid, 0)
        except ProcessLookupError:
            break
        await asyncio.sleep(0.01)
    with pytest.raises(ProcessLookupError):
        os.kill(pid, 0)  # Prozess ist tot
    assert not agents._prozesse  # aus der Registry entfernt


async def _true():
    return True


async def test_opencode_timeout_loggt_tokens(monkeypatch):
    """opencode-Timeout: Tokens der Session werden geloggt, nicht als 0 verworfen."""
    monkeypatch.delenv("CREATOR_FAKE_AGENTS", raising=False)

    async def spawn_timeout(*a, **kw):
        raise asyncio.TimeoutError

    monkeypatch.setattr(agents, "_spawn", spawn_timeout)
    monkeypatch.setattr(agents, "_opencode_tokens",
                        lambda key: {"input": 5000, "output": 3000})
    res = await agents._opencode("t-oc", "prompt", 10, "minimax/M", "none")
    assert res.err == "timeout" and not res.ok
    assert res.tokens == {"input": 5000, "output": 3000}  # nicht 0


def test_gather_nur_in_llm():
    """Rückfall-Guard: bare asyncio.gather darf NUR in llm.py stehen (sonst
    laufen Geschwister-Tasks bei LaufPause/Budget weiter — nutze llm.alle)."""
    from pathlib import Path
    backend = Path(__file__).resolve().parent.parent / "backend"
    treffer = []
    for p in backend.glob("*.py"):
        if p.name == "llm.py":
            continue
        for i, zeile in enumerate(p.read_text(encoding="utf-8").splitlines(), 1):
            if "asyncio.gather" in zeile:
                treffer.append(f"{p.name}:{i}")
    assert not treffer, f"bare asyncio.gather außerhalb llm.py: {treffer}"


async def test_inhaltsfehler_kein_laufabbruch(monkeypatch):
    topic = topic_anlegen()
    run = run_anlegen(topic)
    ctx = llm.Kontext(run, topic, "minimax")
    ctx.ebene = "test"

    async def unsinn(key, prompt, timeout, **kw):
        return agents.AgentErgebnis(0, "kein json", "")

    monkeypatch.setattr(agents, "run_agent", unsinn)
    res = await llm.call(ctx, stage="soll", template="Korpus-Soll",
                         werte={"topic": topic, "quelle": "q", "text": "x"}, erwartet=list)
    assert res is None  # begrenzte Restarts, dann None — kein Absturz, keine Pause


async def test_stop_cancelt_verwaisten_haupt_task(monkeypatch):
    """Stop cancelt den Wartenden — der ensure_future-Haupt-Task muss MIT
    sterben, sonst hält er Slot + Tokens bis zum Timeout (aak: 10 Waisen)."""
    import asyncio
    import agents
    import llm
    from conftest import run_anlegen, topic_anlegen
    topic = topic_anlegen("orphan")
    ctx = llm.Kontext(run_anlegen(topic), topic, "minimax")
    ctx.ebene = "inventar"
    gestartet = asyncio.Event()
    gecancelt = asyncio.Event()

    async def langsamer_agent(key, prompt, timeout, **kw):
        gestartet.set()
        try:
            await asyncio.sleep(60)
        except asyncio.CancelledError:
            gecancelt.set()
            raise
        return agents.AgentErgebnis(0, "nie", "")

    monkeypatch.setattr(agents, "run_agent", langsamer_agent)
    monkeypatch.setattr(llm, "HEDGE_NACH_S", 0)  # Pfad: return await haupt
    task = asyncio.ensure_future(llm._roher_call("k", "p", 10, ctx, "judge", "none"))
    await gestartet.wait()
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass
    await asyncio.wait_for(gecancelt.wait(), timeout=2)  # Waise wurde mitgecancelt


async def test_pause_drainiert_statt_zu_canceln(monkeypatch):
    """ManuellePause in llm.alle: Laufende laufen zu Ende (Ergebnis zählt),
    nichts wird gecancelt."""
    import asyncio
    import llm
    fertig = {"langsam": False}

    async def schnell_pausiert():
        raise llm.ManuellePause("manuell pausiert")

    async def langsam():
        await asyncio.sleep(0.05)
        fertig["langsam"] = True

    import pytest
    with pytest.raises(llm.ManuellePause):
        await llm.alle([schnell_pausiert(), langsam()])
    assert fertig["langsam"] is True  # NICHT gecancelt — ausgelaufen


async def test_pause_beendet_lauf_als_paused():
    """Pause-Flag → nächster Agent liefert „pausiert" → Lauf endet 'paused';
    Start setzt normal fort (Flag wird aufgehoben)."""
    import db
    import pipeline
    from conftest import topic_anlegen
    topic = topic_anlegen("pausetest")
    run_id = pipeline.lauf_starten(topic)
    pipeline.lauf_pausieren(topic)  # vor dem ersten Call — deterministisch
    await pipeline._laeufe[topic]
    run = db.one("SELECT * FROM runs WHERE id=?", (run_id,))
    assert run["status"] == "paused" and "pausiert" in run["grund"]
    # Resume: Start hebt die Pause auf, Lauf läuft durch
    pipeline.lauf_starten(topic)
    await pipeline._laeufe[topic]
    assert db.one("SELECT status FROM topics WHERE name=?", (topic,))["status"] == "fertig"


async def test_output_cap_ohne_neuversuch(monkeypatch):
    """stop=max_tokens ist deterministisch — sofort None statt 2 Neuversuche
    (der Aufrufer halbiert den Chunk)."""
    import agents
    import db
    import llm
    from conftest import run_anlegen, topic_anlegen
    topic = topic_anlegen("cap")
    ctx = llm.Kontext(run_anlegen(topic), topic, "minimax")
    ctx.ebene = "inventar"
    aufrufe = {"n": 0}

    async def cap_agent(key, prompt, timeout, **kw):
        aufrufe["n"] += 1
        return agents.AgentErgebnis(1, "", "leere Antwort (stop=max_tokens)",
                                    {"input": 100, "output": 32000})
    monkeypatch.setattr(agents, "run_agent", cap_agent)
    monkeypatch.setattr(llm, "HEDGE_NACH_S", 0)
    res = await llm.call(ctx, stage="extraktion", template="Atom-Extraktion",
                         werte={"topic": topic, "quelle": "q", "text": "x"}, erwartet=list)
    assert res is None and aufrufe["n"] == 1  # genau EIN Versuch
    ev = db.one("SELECT status, tok_out FROM events WHERE run_id=? AND stage='extraktion'",
                (ctx.run_id,))
    assert ev["status"] == "cap" and ev["tok_out"] == 32000