"""Provider layer: runs agent calls via the Claude CLI or OpenCode (MiniMax).

Both runners are independent. If a binary/key is missing, only the
respective provider fails — the other keeps running unchanged.

Role routing: config.resolve_role maps (run_provider, role) → (provider, model)
ACROSS stacks, so one run can generate on MiniMax and judge on Claude. If the
routed provider is unavailable, the call falls back to the run's provider.
"""

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

import httpx

from config import (PROVIDERS, DEFAULT_PROVIDER, MAX_CONCURRENT_AGENTS,
                    MAX_CONCURRENT_AGENTS_PER_TOPIC, MAX_CONCURRENT_API_AGENTS,
                    MAX_CONCURRENT_INTERACTIVE, RAM_MIN_FREE_PCT, resolve_role)

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

_active_processes: dict[str, asyncio.subprocess.Process] = {}
_active_api: set[str] = set()            # agent_keys of running direct-API calls (no process to kill)
_active_started: dict[str, float] = {}   # agent_key → wall-clock start (for the live runtime display)
_active_labels: dict[str, str] = {}      # agent_key → human-readable label (for display + events)


def active_agents(scope_prefix: str | None = None) -> list[dict]:
    """Currently running agents and how long they've been running. Filter by key prefix
    (e.g. f"blocks-{topic}-") for one topic. → [{key, label, runtime}] sorted longest-first."""
    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"])

# Cancelled scopes (key prefixes, symmetric to kill_process). An agent whose
# key starts with one of these prefixes aborts BEFORE the spawn — so agents WAITING
# in the semaphore queue are also stopped immediately on abort instead of still starting.
_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)

# Caps the real CLI processes — independent of the pipeline semaphore in
# generator.py. The acquire happens BEFORE the spawn so that queue wait time
# does not count against the agent timeout.
class _PrioritySemaphore:
    """asyncio.Semaphore variant: when slots are scarce, the LOWEST priority number is served first
    (FIFO within the same priority). Lets earlier pipeline columns grab agents before later ones."""
    def __init__(self, value: int):
        self._value = value
        self._waiters: list = []  # heap of [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() hands us the slot directly (no value change)
        except BaseException:
            entry[2] = None  # tombstone so release() skips this dead waiter
            if fut.done() and not fut.cancelled():
                self.release()  # granted just before we were cancelled → pass it on
            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)  # hand the slot straight to the highest-priority waiter
                return
        self._value += 1


_batch_sem = _PrioritySemaphore(MAX_CONCURRENT_AGENTS)          # process tier (~310 MB RSS each)
_batch_sem_api = _PrioritySemaphore(MAX_CONCURRENT_API_AGENTS)  # direct-API tier (~0 RAM)
_interactive_sem = asyncio.Semaphore(MAX_CONCURRENT_INTERACTIVE)

# Per-topic caps (lazily created): each topic gets its own priority semaphore of size
# MAX_CONCURRENT_AGENTS_PER_TOPIC, nested INSIDE the global _batch_sem. Priority-based too, so the
# per-topic queue can't undo the global priority when one topic is the only load.
_topic_sems: dict[str, _PrioritySemaphore] = {}

# Smaller index = higher priority. Board 1 (inventory) first — it feeds everything.
# Within board 2 the LATE stages win (outline → artefacts → … → subblocks): finish cards
# instead of opening new WIP, so the makespan tail block gets slots before fresh work.
_STAGE_PRIORITY = ("research", "ingest", "cluster", "pair", "clarify", "naming", "filter",
                   "dedup", "grouping", "gruppierung", "supplement", "outline", "artifact",
                   "question", "relevance", "level", "facts", "subblock")


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)  # unmatched keys (guide board, …) after everything


@asynccontextmanager
async def _batch_gate(scope: str | None, priority: int, api: bool = False):
    """Per-topic slot FIRST (fair), then the GLOBAL slot by priority (earlier columns win when
    agents are scarce). Order matters — a waiter holds only its per-topic slot while queueing globally.
    The per-topic cap is shared across both tiers; only the global cap is tiered (process vs API)."""
    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()

# Space OpenCode starts: processes starting simultaneously collide on the internal
# session DB ("database is locked", exit after <1s). Token bucket instead of a lock
# held through spawn+sleep: the lock only assigns a start slot, the sleep happens
# outside — a wave of starts is spaced by OPENCODE_START_DELAY without a global convoy.
_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-adaptive admission for opencode spawns (~310 MB RSS each): below RAM_MIN_FREE_PCT
# free memory new processes wait instead of starting. Gates admission only — running
# processes are never touched.
_RAM_GATE_FLOOR = 2                # below this many running: always admit (deadlock guard)
_RAM_PER_PROC_KB = 350 * 1024      # commit estimate: RSS ramps up slowly after spawn
_RAM_COMMIT_WINDOW_S = 10.0        # fresh admissions count as already-spent RAM
_RAM_POLL_S = 2.0
_opencode_running = 0              # opencode spawns only (stagger path), not claude
_opencode_recent_starts: list[float] = []  # monotonic timestamps of admissions


def _meminfo() -> tuple[int, int] | None:
    """(MemAvailable_kB, MemTotal_kB) from /proc/meminfo; None → gate fails open (non-Linux)."""
    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) -> bool:
    """True = start admitted (commit registered), False = scope cancelled while waiting.
    Check and commit-append happen in the same synchronous block (no await between) —
    concurrent waiters on the loop cannot double-admit on the same free RAM."""
    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 _opencode_running < _RAM_GATE_FLOOR:
            break  # fail open / floor
        avail_kb, total_kb = mem
        now = time.monotonic()
        _opencode_recent_starts[:] = [t for t in _opencode_recent_starts
                                      if now - t < _RAM_COMMIT_WINDOW_S]
        if avail_kb - len(_opencode_recent_starts) * _RAM_PER_PROC_KB >= total_kb * RAM_MIN_FREE_PCT / 100:
            break
        if not waited:
            log.info("agent %s: RAM gate waiting (%.0f%% free)", agent_key, avail_kb * 100 / total_kb)
            waited = True
        await asyncio.sleep(_RAM_POLL_S)
    _opencode_recent_starts.append(time.monotonic())
    return True

_SLIM_CONFIG = Path(__file__).resolve().parent.parent / "dev-ops" / "opencode-slim.json"

# Capability → Claude --allowedTools
_CLAUDE_TOOLS = {
    "full": "Write,Bash,Read,WebSearch,WebFetch",
    "files": "Read,Bash,Write",
    "read": "Read",
    "none": None,
}

# Capability → OpenCode agent (tool permissions defined in dev-ops/opencode.json)
_OPENCODE_AGENTS = {
    "full": "full",
    "files": "files",
    "read": "readonly",
    "none": "text",
}


def _use_text_api(provider: str, model: str, capabilities: str, on_line) -> bool:
    """Direct API path only for tool-less, non-streaming MiniMax calls. Model-prefix check
    instead of cli check: "lokal" (ollama) also runs via opencode. Missing key or
    CREATOR_TEXT_API=0 (kill switch) falls back to the opencode process path."""
    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("CREATOR_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
    check_url = cfg.get("check_url")
    if check_url:
        try:
            urllib.request.urlopen(check_url, timeout=1)
        except Exception:
            return False
    return True


# Availability cache for role routing: the routed target is probed at most once per
# TTL (check_url providers would otherwise block the loop on every call).
_avail_cache: dict[str, tuple[float, bool]] = {}
_AVAIL_TTL = 60.0


def _available_cached(provider: str) -> bool:
    now = time.monotonic()
    hit = _avail_cache.get(provider)
    if hit and now - hit[0] < _AVAIL_TTL:
        return hit[1]
    ok = provider_available(provider)
    _avail_cache[provider] = (now, ok)
    return ok


def _kill(process) -> None:
    """Kill the agent and its child processes via the process group (otherwise the
    children spawned by the CLI survive, keep the pipes open and block 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:
    """Kill all active processes whose key starts with the prefix (covers -plan/-w1…)."""
    for key, process in list(_active_processes.items()):
        if process.returncode is not None:  # clean up dead entries while iterating
            _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 for the pipeline history (injected by main.py lifespan as database.add_event —
# agents.py stays DB-free). Called fire-and-forget for every finished BATCH agent.
on_event = None


async def run_agent(
    agent_key: str,
    prompt: str,
    timeout: int,
    provider: str = DEFAULT_PROVIDER,
    role: str = "fast",
    capabilities: str = "none",
    lane: str = "batch",
    scope: str | None = None,
    on_line=None,
    label: str = "",
) -> tuple[int, str, str]:
    if os.getenv("CREATOR_FAKE_AGENTS"):  # Sekunden-Smoke: deterministische Antworten statt LLM
        import fake_agents
        return await fake_agents.respond(agent_key, prompt, capabilities)
    if _scope_cancelled(agent_key):  # before queueing: don't even enter the queue
        return 1, "", "cancelled"
    if provider not in PROVIDERS:
        return 1, "", f"Unknown provider: {provider}"
    run_provider = provider
    provider, model = resolve_role(run_provider, role)
    if provider != run_provider and not _available_cached(provider):
        provider, model = run_provider, PROVIDERS[run_provider].get(role, "")
    if not model:
        return 1, "", f"No model for role '{role}' (provider: {provider})"
    use_api = _use_text_api(provider, model, capabilities, on_line)
    if not use_api and shutil.which(PROVIDERS[provider]["cli"]) is None:
        return 1, "", f"CLI '{PROVIDERS[provider]['cli']}' not installed (provider: {provider})"
    queued = time.monotonic()
    gate = _interactive_sem if lane == "interactive" else _batch_gate(scope, _agent_priority(agent_key), api=use_api)
    async with gate:
        if _scope_cancelled(agent_key):  # after the acquire: cancelled in the queue → no spawn
            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 (role %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, provider, model, capabilities, on_line=on_line, label=label)
            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:]  # diagnosis: rc=1 without stderr is opaque
            return res
        except asyncio.TimeoutError:
            status = "timeout"
            raise
        except asyncio.CancelledError:
            status = "cancelled"
            raise
        finally:
            if on_event is not None and scope is not None:  # batch pipeline only, never fatal
                try:
                    meta = {"provider": provider, "model": model, "role": role, "rc": rc}
                    if err_tail:
                        meta["stderr"] = err_tail
                    if api_tokens:  # direct-API path: usage from the response, even on rc!=0
                        meta["tokens"] = api_tokens
                    elif PROVIDERS[provider]["cli"] == "opencode":  # token accounting per agent
                        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 _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) -> 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,  # own process group → killpg also kills child processes
            env=env,
        )

    global _opencode_running
    if stagger:
        if not await _ram_gate(agent_key):
            return 1, "", "cancelled"     # like the cancelled path in run_agent
        await _opencode_slot()  # gate BEFORE the start slot: an admission wave still gets spaced
    process = await spawn()
    if stagger:
        _opencode_running += 1
    # Collision-safe tracking: identical keys (e.g. same chunk label from parallel cards)
    # get a ~n suffix — prefix-based kill/cancel still matches, nothing becomes an orphan.
    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:
                # Streaming path: read stdout line by line, hand each raw line to on_line LIVE.
                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:
            _opencode_running -= 1
        # Pop only on identity: a slot restart under the same key must not evict
        # the NEW process from tracking.
        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]:
    cfg = PROVIDERS["claude"]
    cmd = [cfg["cli"], "-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, label=label)


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


def _session_tokens(agent_key: str) -> dict | None:
    """Token counters of the newest OpenCode session titled `agent_key` (set via run --title).
    Best-effort read-only lookup — None when DB/row is missing; never fails the agent.
    Timeouts count too: their sessions consumed tokens, that waste should be visible."""
    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, provider: str, model: str, capabilities: str, on_line=None, label: str = "") -> tuple[int, str, str]:
    cfg = PROVIDERS[provider]
    # Prompt via temp file instead of argv (ARG_MAX protection for large project prompts)
    with tempfile.NamedTemporaryFile("w", suffix=".md", delete=False, encoding="utf-8", dir=tempfile.gettempdir()) as f:
        f.write(prompt)
        prompt_path = Path(f.name)
    # The positional message MUST come before -f: -f is an array flag and
    # would otherwise eat the text as a second file name ("File not found").
    cmd = [
        cfg["cli"], "run",
        "Folge exakt den Anweisungen in der angehängten Datei. Sie sind der vollständige Auftrag.",
        "-m", model,
        "--agent", _OPENCODE_AGENTS.get(capabilities, "text"),
        "--dangerously-skip-permissions",
        "--title", agent_key,  # token accounting: joins the OpenCode session to our event
        "-f", str(prompt_path),
    ]
    if on_line is not None:
        cmd += ["--format", "json"]  # raw JSON events → parsed live by on_line
    env = None
    if capabilities != "full":
        # Batch agents (files/readonly/text) never use the web MCPs, but opencode starts
        # every configured MCP server PER PROCESS (~3 procs / ~300 MB each). Point them
        # at the mcp-free config copy; only `full` (research/supplement) keeps the servers.
        env = {**os.environ, "OPENCODE_CONFIG": str(_SLIM_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)


# Direct text-API path (MiniMax, Anthropic Messages format): saves the ~310 MB RSS opencode
# process for tool-less calls. The native "minimax" and "minimax-kalt" opencode providers both
# resolve to this base URL — only the per-model options differ. Options source of truth:
# dev-ops/opencode.json (+ -slim); keep in sync on changes there.
_API_URL = "https://api.minimax.io/anthropic/v1/messages"
_API_VERSION = "2023-06-01"
_API_MAX_TOKENS = 32_000  # required Messages field; generate calls are long
_API_MODEL_OPTS = {       # prefix "minimax" (native) → endpoint defaults (no entry)
    "minimax-kalt/MiniMax-M3": {"temperature": 0.2, "thinking": {"type": "disabled"}},
    "minimax-kalt/MiniMax-M2.7-highspeed": {"temperature": 0.3},
}


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 come straight from the response usage (same key set
    as _session_tokens) and are returned even on rc!=0 — waste stays visible."""
    body = {
        "model": model.split("/", 1)[1],  # without the opencode provider prefix
        "max_tokens": _API_MAX_TOKENS,
        "messages": [{"role": "user", "content": prompt}],
        **_API_MODEL_OPTS.get(model, {}),
    }
    headers = {"x-api-key": os.environ.get("MINIMAX_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(  # belt: hard wall-clock cap like the process path
                client.post(_API_URL, json=body, headers=headers), timeout=timeout)
    except httpx.TimeoutException:
        raise asyncio.TimeoutError  # contract: run_agent/_race handle 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()
    # Only text blocks count — thinking blocks (native route) are skipped.
    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"empty response (stop_reason={data.get('stop_reason')})", tokens
    err = "stop_reason=max_tokens (truncated)" 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:
    """Strip ANSI codes and the leading banner ("> agent · model")."""
    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()