"""Provider-Layer: LLM-Calls über Claude-CLI, OpenCode-CLI oder direkte MiniMax-API.

Port aus creator/backend/agents.py mit zwei bewussten Abweichungen:
- KEIN Provider-Fallback: resolve_role(role) kommt vollständig aus der .env; ist eine
  Rolle unkonfiguriert oder der Provider nicht nutzbar, wird der Lauf pausiert
  (AgentInfraError) — nie still gewechselt (Leitprinzip 2).
- Leere-Ausgabe-Triage (Phase-0-Erkenntnis 13): leerer Text bei rc=0 ist ein Fehler.
  OpenCode-Pfad: keine Session in der OpenCode-DB → Netz/Provider → AgentInfraError
  (Pause); Session vorhanden → Retry-Fehler (Thinking-Overrun/Stall).

Der Event-Sink `on_event` wird von außen injiziert (CLI setzt database.add_event) —
dieses Modul bleibt DB-frei.
"""

import asyncio
import heapq
import logging
import os
import re
import shutil
import signal
import sqlite3
import tempfile
import time
from contextlib import asynccontextmanager
from pathlib import Path

import httpx

from config import (MAX_CONCURRENT_AGENTS, MAX_CONCURRENT_AGENTS_PER_TOPIC,
                    MAX_CONCURRENT_API_AGENTS, OPENCODE_CONFIG, PROVIDERS,
                    RAM_MIN_FREE_PCT, fake_agents_aktiv, resolve_role)

log = logging.getLogger("planer.agents")


class AgentInfraError(Exception):
    """Infrastruktur nicht nutzbar (Netz/Provider down, Rolle unkonfiguriert). KEIN
    inhaltliches Urteil — der Aufrufer pausiert den Lauf, nie fail-open."""


_active_processes: dict[str, asyncio.subprocess.Process] = {}
_active_api: set[str] = set()            # laufende direkte API-Calls (kein Prozess zu killen)
_active_started: dict[str, float] = {}   # agent_key → Wandzeit-Start (Live-Anzeige)
_active_labels: dict[str, str] = {}


def active_agents(scope_prefix: str | None = None) -> list[dict]:
    """Laufende Agenten mit Laufzeit, längste zuerst. → [{key, label, runtime}]"""
    now = time.time()
    out = [{"key": k, "label": _active_labels.get(k, ""), "runtime": round(now - t, 1)}
           for k, t in list(_active_started.items())
           if (k in _active_processes or k in _active_api)
           and (not scope_prefix or k.startswith(scope_prefix))]
    return sorted(out, key=lambda a: -a["runtime"])


# --- Abbruch (Prefix-basiert) --------------------------------------------------------
_cancelled_prefixes: set[str] = set()


def cancel_scope(prefix: str) -> None:
    _cancelled_prefixes.add(prefix)


def clear_scope(prefix: str) -> None:
    _cancelled_prefixes.discard(prefix)


def _scope_cancelled(agent_key: str) -> bool:
    return any(agent_key.startswith(p) for p in _cancelled_prefixes)


class _PrioritySemaphore:
    """Semaphore-Variante: bei knappen Slots gewinnt die NIEDRIGSTE Prioritätszahl
    (FIFO innerhalb gleicher Priorität) — frühe Board-Spalten vor späten."""

    def __init__(self, value: int):
        self._value = value
        self._waiters: list = []  # Heap aus [priority, seq, future]
        self._seq = 0

    async def acquire(self, priority: int = 100):
        if self._value > 0:
            self._value -= 1
            return
        fut = asyncio.get_event_loop().create_future()
        entry = [priority, self._seq, fut]
        self._seq += 1
        heapq.heappush(self._waiters, entry)
        try:
            await fut  # release() reicht den Slot direkt weiter (kein value-Increment)
        except BaseException:
            entry[2] = None  # Tombstone: release() überspringt tote Waiter
            if fut.done() and not fut.cancelled():
                self.release()  # Slot kam exakt vor dem Cancel → weitergeben
            raise

    def release(self):
        while self._waiters:
            entry = heapq.heappop(self._waiters)
            if entry[2] is not None and not entry[2].done():
                entry[2].set_result(None)
                return
        self._value += 1


_batch_sem = _PrioritySemaphore(MAX_CONCURRENT_AGENTS)          # Prozess-Tier (~310 MB RSS)
_batch_sem_api = _PrioritySemaphore(MAX_CONCURRENT_API_AGENTS)  # API-Tier (~0 RAM)
_topic_sems: dict[str, _PrioritySemaphore] = {}

# Kleinerer Index = höhere Priorität: Karten fertig machen statt neues WIP öffnen.
_STAGE_PRIORITY = ("sichten", "judge", "nachfass", "schneiden")


def _agent_priority(key: str) -> int:
    for i, tag in enumerate(_STAGE_PRIORITY):
        if f"-{tag}-" in key or key.endswith(f"-{tag}"):
            return i
    return len(_STAGE_PRIORITY)


@asynccontextmanager
async def _batch_gate(scope: str | None, priority: int, api: bool = False):
    """Erst der Per-Topic-Slot (fair), dann der GLOBALE Slot nach Priorität. Ein Waiter
    hält beim globalen Anstehen nur seinen Topic-Slot. Topic-Cap gilt über beide Tiers."""
    topic_sem = _topic_sems.setdefault(scope, _PrioritySemaphore(MAX_CONCURRENT_AGENTS_PER_TOPIC)) if scope else None
    global_sem = _batch_sem_api if api else _batch_sem
    if topic_sem is not None:
        await topic_sem.acquire(priority)
    await global_sem.acquire(priority)
    try:
        yield
    finally:
        global_sem.release()
        if topic_sem is not None:
            topic_sem.release()


# Stagger für OpenCode-Starts: gleichzeitige Kaltstarts kollidieren auf der internen
# Session-DB (Phase-0-Erkenntnis 13a: alle hängen, keine Session entsteht). Token-Bucket:
# der Lock vergibt nur den Start-Slot, geschlafen wird außerhalb.
_opencode_start_lock = asyncio.Lock()
_OPENCODE_START_DELAY = float(os.getenv("OPENCODE_START_DELAY", "0.5"))
_opencode_next_start = 0.0


async def _opencode_slot() -> None:
    global _opencode_next_start
    loop = asyncio.get_running_loop()
    async with _opencode_start_lock:
        now = loop.time()
        start_at = max(now, _opencode_next_start)
        _opencode_next_start = start_at + _OPENCODE_START_DELAY
    await asyncio.sleep(max(0.0, start_at - now))


# --- RAM-Gate: unter RAM_MIN_FREE_PCT freiem Speicher warten neue CLI-Prozesse ---------
_RAM_GATE_FLOOR = 2                # unter so vielen Läufen: immer zulassen (Deadlock-Schutz)
_RAM_PER_PROC_KB = 350 * 1024
_RAM_COMMIT_WINDOW_S = 10.0        # frische Zulassungen zählen als schon verbrauchtes RAM
_RAM_POLL_S = 2.0
_cli_running = 0
_cli_recent_starts: list[tuple[float, int]] = []


def _meminfo() -> tuple[int, int] | None:
    try:
        text = Path("/proc/meminfo").read_text()
    except OSError:
        return None
    m = {k: v for k, v in re.findall(r"^(MemTotal|MemAvailable):\s+(\d+)", text, re.MULTILINE)}
    if "MemTotal" not in m or "MemAvailable" not in m:
        return None
    return int(m["MemAvailable"]), int(m["MemTotal"])


async def _ram_gate(agent_key: str, est_kb: int) -> bool:
    """True = Start zugelassen (Commit registriert), False = Scope während des Wartens
    abgebrochen. Check und Commit-Append im selben synchronen Block (kein await dazwischen)."""
    if RAM_MIN_FREE_PCT <= 0:
        return True
    waited = False
    while True:
        if _scope_cancelled(agent_key):
            return False
        mem = _meminfo()
        if mem is None or _cli_running < _RAM_GATE_FLOOR:
            break
        avail_kb, total_kb = mem
        now = time.monotonic()
        _cli_recent_starts[:] = [(t, kb) for t, kb in _cli_recent_starts
                                 if now - t < _RAM_COMMIT_WINDOW_S]
        committed_kb = sum(kb for _, kb in _cli_recent_starts)
        if avail_kb - committed_kb >= total_kb * RAM_MIN_FREE_PCT / 100:
            break
        if not waited:
            log.info("agent %s: RAM-Gate wartet (%.0f%% frei)", agent_key, avail_kb * 100 / total_kb)
            waited = True
        await asyncio.sleep(_RAM_POLL_S)
    _cli_recent_starts.append((time.monotonic(), est_kb))
    return True


# Capability → Claude --allowedTools / OpenCode-Agent
_CLAUDE_TOOLS = {"full": "Write,Bash,Read,WebSearch,WebFetch", "files": "Read,Bash,Write",
                 "read": "Read", "none": None}
_OPENCODE_AGENTS = {"full": "full", "files": "files", "read": "readonly", "none": "text"}


def _use_text_api(provider: str, model: str, capabilities: str, on_line) -> bool:
    """Direkter API-Pfad nur für tool-lose, nicht-streamende MiniMax-Calls — spart den
    OpenCode-Prozess (und dessen Stall-/Kaltstart-Probleme). PLANER_TEXT_API=0 = aus."""
    return (capabilities == "none" and on_line is None
            and PROVIDERS[provider]["cli"] == "opencode"
            and model.split("/", 1)[0] in ("minimax", "minimax-kalt")
            and bool(os.environ.get("MINIMAX_API_KEY"))
            and os.getenv("PLANER_TEXT_API", "1") != "0")


def provider_available(provider: str) -> bool:
    cfg = PROVIDERS.get(provider)
    if not cfg:
        return False
    if shutil.which(cfg["cli"]) is None:
        return False
    env_key = cfg.get("env_key")
    if env_key and not os.environ.get(env_key):
        return False
    return True


def _kill(process) -> None:
    """Agent + Kindprozesse über die Prozessgruppe killen (Kinder halten sonst die
    Pipes offen und blockieren communicate())."""
    try:
        os.killpg(os.getpgid(process.pid), signal.SIGKILL)
    except (ProcessLookupError, PermissionError):
        try:
            process.kill()
        except ProcessLookupError:
            pass


def kill_process(agent_key_prefix: str) -> None:
    for key, process in list(_active_processes.items()):
        if process.returncode is not None:
            _active_processes.pop(key, None)
            _active_started.pop(key, None)
            continue
        if key.startswith(agent_key_prefix):
            log.debug("kill agent %s", key)
            _kill(process)


# Event-Sink (CLI injiziert database.add_event). Fire-and-forget je fertigem Agenten.
on_event = None


async def run_agent(
    agent_key: str,
    prompt: str,
    timeout: int,
    role: str,
    capabilities: str = "none",
    scope: str | None = None,
    on_line=None,
    label: str = "",
) -> tuple[int, str, str]:
    """Ein LLM-Call der Rolle `role`. → (rc, stdout, stderr). Wirft AgentInfraError,
    wenn die Infrastruktur nicht nutzbar ist (Rolle unkonfiguriert, Provider down,
    leere Antwort ohne Session) — der Lauf soll dann pausieren, nicht raten."""
    if fake_agents_aktiv():  # Sekunden-Smoke: deterministisch statt LLM
        import fake_agents
        start = time.monotonic()
        res = await fake_agents.respond(agent_key, prompt, capabilities)
        if on_event is not None and scope is not None:  # Telemetrie auch im Fake-Pfad testbar
            await on_event(topic=scope, kind="agent", key=agent_key, label=label,
                           status="ok" if res[0] == 0 else "error",
                           dur_ms=int((time.monotonic() - start) * 1000), wait_ms=0,
                           meta={"provider": "fake", "model": "fake", "role": role, "rc": res[0]})
        return res
    if _scope_cancelled(agent_key):
        return 1, "", "cancelled"
    try:
        provider, model = resolve_role(role)  # NUR aus .env — kein Default, kein Fallback
    except RuntimeError as e:
        raise AgentInfraError(str(e)) from e
    use_api = _use_text_api(provider, model, capabilities, on_line)
    if not use_api and shutil.which(PROVIDERS[provider]["cli"]) is None:
        raise AgentInfraError(f"CLI '{PROVIDERS[provider]['cli']}' nicht installiert (Provider {provider})")
    queued = time.monotonic()
    async with _batch_gate(scope, _agent_priority(agent_key), api=use_api):
        if _scope_cancelled(agent_key):
            return 1, "", "cancelled"
        wait_ms = int((time.monotonic() - queued) * 1000)
        start = time.monotonic()
        status = "error"
        rc = None
        err_tail = ""
        api_tokens = None
        try:
            log.info("agent %s: %s %s (Rolle %s)", agent_key, provider, model, role)
            if use_api:
                rc_, out_, err_, api_tokens = await _run_text_api(agent_key, prompt, timeout, model, label=label)
                res = (rc_, out_, err_)
            elif PROVIDERS[provider]["cli"] == "opencode":
                res = await _run_opencode(agent_key, prompt, timeout, model, capabilities,
                                          on_line=on_line, label=label)
                res = await _leere_ausgabe_triage(agent_key, res, on_line)
            else:
                res = await _run_claude_cli(agent_key, prompt, timeout, model, capabilities, label=label)
            rc = res[0]
            status = "ok" if rc == 0 else ("killed" if rc is not None and rc < 0 else "error")
            if rc not in (0, None) and rc >= 0:
                err_tail = (res[2] or res[1] or "").strip()[-300:]
            return res
        except asyncio.TimeoutError:
            status = "timeout"
            raise
        except asyncio.CancelledError:
            status = "cancelled"
            raise
        except AgentInfraError:
            status = "infra"
            raise
        finally:
            if on_event is not None and scope is not None:  # nie fatal
                try:
                    meta = {"provider": provider, "model": model, "role": role, "rc": rc}
                    if err_tail:
                        meta["stderr"] = err_tail
                    if api_tokens:
                        meta["tokens"] = api_tokens
                    elif PROVIDERS[provider]["cli"] == "opencode":
                        if (tok := await asyncio.to_thread(_session_tokens, agent_key)):
                            meta["tokens"] = tok
                    await on_event(topic=scope, kind="agent", key=agent_key, label=label,
                                   status=status, dur_ms=int((time.monotonic() - start) * 1000),
                                   wait_ms=wait_ms, meta=meta)
                except Exception:
                    log.debug("on_event failed", exc_info=True)


async def _leere_ausgabe_triage(agent_key: str, res: tuple[int, str, str], on_line) -> tuple[int, str, str]:
    """Erkenntnis 13: leerer Text bei rc=0 ist nie ok. Keine OpenCode-Session zum Key →
    der Call kam nie beim Provider an (Netz) → AgentInfraError (Lauf pausiert).
    Session vorhanden → Thinking-Overrun/Stall → normaler Fehler (Engine-Retry)."""
    rc, out, err = res
    if rc != 0 or out.strip() or on_line is not None:
        return res
    tok = await asyncio.to_thread(_session_tokens, agent_key)
    if tok is None:
        raise AgentInfraError(f"agent {agent_key}: leere Ausgabe ohne OpenCode-Session — Provider/Netz nicht erreichbar")
    return 1, "", f"leere Ausgabe (Session vorhanden, output-tokens={tok.get('output')}) — Overrun/Stall"


async def _communicate(agent_key: str, cmd: list[str], stdin_data: bytes | None, timeout: int,
                       stagger: bool = False, on_line=None, label: str = "",
                       env: dict | None = None, est_kb: int = _RAM_PER_PROC_KB) -> tuple[int, str, str]:
    start = time.monotonic()

    async def spawn():
        return await asyncio.create_subprocess_exec(
            *cmd,
            stdin=asyncio.subprocess.PIPE if stdin_data is not None else asyncio.subprocess.DEVNULL,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            start_new_session=True,  # eigene Prozessgruppe → killpg trifft auch Kinder
            env=env,
        )

    global _cli_running
    if stagger:
        if not await _ram_gate(agent_key, est_kb):
            return 1, "", "cancelled"
        await _opencode_slot()  # Gate VOR dem Start-Slot: Zulassungswelle wird trotzdem gespaced
    process = await spawn()
    if stagger:
        _cli_running += 1
    # Kollisionssicheres Tracking: identische Keys bekommen ~n-Suffix (Prefix-Kill matcht weiter)
    track_key = agent_key
    n = 2
    while track_key in _active_processes:
        track_key = f"{agent_key}~{n}"
        n += 1
    _active_processes[track_key] = process
    _active_started[track_key] = time.time()
    _active_labels[track_key] = label
    try:
        try:
            if on_line is not None:
                out_chunks: list[str] = []

                async def _pump():
                    async for raw in process.stdout:
                        s = raw.decode("utf-8", errors="replace")
                        out_chunks.append(s)
                        try:
                            on_line(s)
                        except Exception:
                            log.debug("on_line callback failed", exc_info=True)
                await asyncio.wait_for(_pump(), timeout=timeout)
                await process.wait()
                stderr_b = await process.stderr.read()
                stdout, stderr = "".join(out_chunks).encode("utf-8"), stderr_b
            else:
                stdout, stderr = await asyncio.wait_for(
                    process.communicate(input=stdin_data),
                    timeout=timeout,
                )
        except asyncio.TimeoutError:
            _kill(process)
            try:
                await asyncio.wait_for(process.wait(), timeout=5)
            except asyncio.TimeoutError:
                pass
            log.info("agent %s: timeout after %ds", agent_key, timeout)
            raise
        log.info("agent %s: exit %s after %.1fs (%d bytes stdout)",
                 agent_key, process.returncode, time.monotonic() - start, len(stdout))
        return process.returncode, stdout.decode("utf-8", errors="replace"), stderr.decode("utf-8", errors="replace")
    finally:
        if stagger:
            _cli_running -= 1
        if _active_processes.get(track_key) is process:
            del _active_processes[track_key]
            _active_started.pop(track_key, None)
            _active_labels.pop(track_key, None)


async def _run_claude_cli(agent_key: str, prompt: str, timeout: int, model: str,
                          capabilities: str, label: str = "") -> tuple[int, str, str]:
    cmd = ["claude", "-p", "--model", model]
    tools = _CLAUDE_TOOLS.get(capabilities)
    if tools:
        cmd += ["--allowedTools", tools]
    cmd += ["--dangerously-skip-permissions"]
    return await _communicate(agent_key, cmd, prompt.encode("utf-8"), timeout, stagger=True, label=label)


_OPENCODE_DB = Path.home() / ".local" / "share" / "opencode" / "opencode.db"


def _session_tokens(agent_key: str) -> dict | None:
    """Token-Zähler der neuesten OpenCode-Session mit title=agent_key (via run --title).
    Best-effort, read-only, nie fatal — None wenn DB/Zeile fehlt. Auch Timeouts haben
    Tokens verbraucht; die Verschwendung soll sichtbar bleiben."""
    try:
        con = sqlite3.connect(f"file:{_OPENCODE_DB}?mode=ro", uri=True, timeout=1)
        try:
            row = con.execute(
                "SELECT tokens_input, tokens_output, tokens_reasoning,"
                " tokens_cache_read, tokens_cache_write"
                " FROM session WHERE title=? ORDER BY time_created DESC LIMIT 1",
                (agent_key,)).fetchone()
        finally:
            con.close()
    except Exception:
        return None
    if row is None:
        return None
    keys = ("input", "output", "reasoning", "cache_read", "cache_write")
    return {k: int(v or 0) for k, v in zip(keys, row)}


async def _run_opencode(agent_key: str, prompt: str, timeout: int, model: str, capabilities: str,
                        on_line=None, label: str = "") -> tuple[int, str, str]:
    # Prompt per Temp-Datei statt argv (ARG_MAX-Schutz bei großen Chunks)
    with tempfile.NamedTemporaryFile("w", suffix=".md", delete=False, encoding="utf-8") as f:
        f.write(prompt)
        prompt_path = Path(f.name)
    # Die positionale Message MUSS vor -f stehen (-f ist ein Array-Flag).
    cmd = [
        "opencode", "run",
        "Folge exakt den Anweisungen in der angehängten Datei. Sie sind der vollständige Auftrag.",
        "-m", model,
        "--agent", _OPENCODE_AGENTS.get(capabilities, "text"),
        "--title", agent_key,  # Token-Accounting: verbindet die OpenCode-Session mit unserem Event
        "-f", str(prompt_path),
    ]
    if on_line is not None:
        cmd += ["--format", "json"]
    env = {**os.environ, "OPENCODE_CONFIG": str(OPENCODE_CONFIG)}
    try:
        rc, stdout, stderr = await _communicate(agent_key, cmd, None, timeout, stagger=True,
                                                on_line=on_line, label=label, env=env)
        return rc, (stdout if on_line is not None else _clean_opencode_output(stdout)), stderr
    finally:
        prompt_path.unlink(missing_ok=True)


# --- Direkter API-Pfad (MiniMax, Anthropic-Messages-Format) ---------------------------
_API_URL = "https://api.minimax.io/anthropic/v1/messages"
_API_VERSION = "2023-06-01"
_API_MAX_TOKENS = 32_000


def _api_model_opts(model: str) -> dict:
    """kalt-Route = niedrige Temperatur ohne Thinking; nativ = Endpoint-Defaults.
    Bewusst prefix-basiert statt Modell-Tabelle — Modell-IDs gehören nicht in den Code."""
    if model.startswith("minimax-kalt/"):
        return {"temperature": 0.3, "thinking": {"type": "disabled"}}
    return {}


async def _run_text_api(agent_key: str, prompt: str, timeout: int, model: str,
                        label: str = "") -> tuple[int, str, str, dict | None]:
    """→ (rc, text, err, tokens). Tokens direkt aus der Response-usage, auch bei rc!=0 —
    Verschwendung bleibt sichtbar. Leere Antwort und max_tokens-Abbruch sind Fehler."""
    body = {
        "model": model.split("/", 1)[1],  # ohne OpenCode-Provider-Prefix
        "max_tokens": _API_MAX_TOKENS,
        "messages": [{"role": "user", "content": prompt}],
        **_api_model_opts(model),
    }
    api_key = os.environ.get("MINIMAX_API_KEY")
    if not api_key:
        raise AgentInfraError("MINIMAX_API_KEY fehlt — in .env setzen")
    headers = {"x-api-key": api_key, "anthropic-version": _API_VERSION}
    track_key = agent_key
    n = 2
    while track_key in _active_api:
        track_key = f"{agent_key}~{n}"
        n += 1
    _active_api.add(track_key)
    _active_started[track_key] = time.time()
    _active_labels[track_key] = label
    start = time.monotonic()
    try:
        async with httpx.AsyncClient(timeout=httpx.Timeout(timeout, connect=30)) as client:
            resp = await asyncio.wait_for(
                client.post(_API_URL, json=body, headers=headers), timeout=timeout)
    except httpx.TimeoutException:
        raise asyncio.TimeoutError
    except httpx.HTTPError as e:
        return 1, "", f"{type(e).__name__}: {e}", None
    finally:
        _active_api.discard(track_key)
        _active_started.pop(track_key, None)
        _active_labels.pop(track_key, None)
        log.info("agent %s: api done after %.1fs", agent_key, time.monotonic() - start)
    if resp.status_code != 200:
        return 1, "", f"HTTP {resp.status_code}: {resp.text[:300]}", None
    data = resp.json()
    # Nur Text-Blöcke zählen — Thinking-Blöcke (native Route) werden übersprungen.
    text = "".join(b.get("text", "") for b in data.get("content", []) if b.get("type") == "text")
    u = data.get("usage") or {}
    tokens = {"input": int(u.get("input_tokens") or 0), "output": int(u.get("output_tokens") or 0),
              "reasoning": 0, "cache_read": int(u.get("cache_read_input_tokens") or 0),
              "cache_write": int(u.get("cache_creation_input_tokens") or 0)}
    if not text.strip():
        return 1, "", f"leere Antwort (stop_reason={data.get('stop_reason')})", tokens
    err = "stop_reason=max_tokens (abgeschnitten)" if data.get("stop_reason") == "max_tokens" else ""
    return 0, text, err, tokens


_ANSI_RE = re.compile(r"\x1b\[[0-9;]*m")


def _clean_opencode_output(text: str) -> str:
    """ANSI-Codes und den führenden Banner ("> agent · model") entfernen."""
    text = _ANSI_RE.sub("", text)
    lines = text.splitlines()
    while lines and (not lines[0].strip() or lines[0].lstrip().startswith(">")):
        lines.pop(0)
    return "\n".join(lines).strip()