"""Programa: CPU-only OpenAI-compatible LLM endpoint for the original Voyager agent."""
from __future__ import annotations

import ast
import json
import math
import os
import pathlib
import re
import threading
import time
import uuid
from typing import Any, Iterator

from cpu_backend import cpu_only_report, cpu_only_summary

# The Space is hosted on ZeroGPU hardware only to retain the 16-vCPU / ~97-GiB
# CPU envelope. ZeroGPU requires at least one decorated function. The sentinel
# below is NEVER called by the API or UI, so it never requests GPU time.
import spaces

# Strict CPU mode before llama.cpp is imported.
os.environ["CUDA_VISIBLE_DEVICES"] = ""
os.environ["HIP_VISIBLE_DEVICES"] = ""
os.environ["ROCR_VISIBLE_DEVICES"] = ""
os.environ["GGML_VK_VISIBLE_DEVICES"] = ""
os.environ["GGML_CUDA"] = "0"
os.environ["HF_HUB_DISABLE_PROGRESS_BARS"] = "1"
os.environ["TQDM_DISABLE"] = "1"
os.environ.setdefault("MALLOC_ARENA_MAX", "2")

import gradio as gr
import psutil
import tiktoken
import uvicorn
from fastapi import FastAPI, Header, HTTPException, Request
from fastapi.responses import JSONResponse, StreamingResponse
from huggingface_hub import hf_hub_download
from starlette.concurrency import run_in_threadpool
from minecraft_stack import STACK
from voyager_stack import VOYAGER
from gemini_bot_stack import GEMINI_BOT
from altoclef_stack import ALTOCLEF

MODEL_REPO = os.environ.get(
    "MODEL_REPO",
    "unsloth/Qwen3-Coder-30B-A3B-Instruct-GGUF",
)
MODEL_FILE = os.environ.get(
    "MODEL_FILE",
    "Qwen3-Coder-30B-A3B-Instruct-Q4_K_M.gguf",
)
MODEL_REVISION = os.environ.get(
    "MODEL_REVISION",
    "b17cb02dd882d5b6ab62fc777ad2995f19668350",
)
MODEL_ALIAS = os.environ.get("MODEL_ALIAS", "qwen3-coder-30b-a3b-instruct")
MODEL_ALIASES = {
    MODEL_ALIAS,
    "Qwen3-Coder-30B-A3B-Instruct",
    "voyager-qwen3-coder",
    # Kept so an older client or checkpoint that still names the previous brain
    # resolves instead of 404ing on the model name.
    "granite-4.0-h-tiny",
    "Granite-4.0-H-Tiny",
    "voyager-granite-4-h-tiny",
    "qwen2.5-coder-7b-instruct",
    "Qwen2.5-Coder-7B-Instruct",
    "voyager-qwen2.5-coder-7b",
    "qwen2.5-coder-3b-instruct",
    "Qwen2.5-Coder-3B-Instruct",
    "voyager-qwen2.5-coder-3b",
    "qwen3-4b-instruct-2507",
    "voyager-qwen3-4b",
}
# The model is 18.56 GB. huggingface_hub defaults to ~/.cache, which this
# container loses on every rebuild, so a rebuild re-downloaded the whole file
# before the first request could be served. /persist survives, so the download
# happens once.
HF_CACHE_DIR = os.environ.get("HF_CACHE_DIR", "/persist/hf-cache")

AUX_MODEL_REPO = os.environ.get("AUX_MODEL_REPO", "ibm-granite/granite-4.0-h-tiny-GGUF")
AUX_MODEL_FILE = os.environ.get("AUX_MODEL_FILE", "granite-4.0-h-tiny-Q4_K_M.gguf")
AUX_MODEL_REVISION = os.environ.get("AUX_MODEL_REVISION", "08d5a8a9741dd5c1a95d2d39e25253226aa1464e")
AUX_MODEL_ALIAS = os.environ.get("AUX_MODEL_ALIAS", "granite-4.0-h-tiny-aux")
AUX_MODEL_ALIASES = {AUX_MODEL_ALIAS, "voyager-aux-granite-tiny"}
AUX_CTX = int(os.environ.get("AUX_CPU_CTX", "8192"))
AUX_PRELOAD = os.environ.get("AUX_CPU_PRELOAD", "0") == "1"

EMBED_REPO = os.environ.get("EMBED_REPO", "Qwen/Qwen3-Embedding-0.6B-GGUF")
EMBED_FILE = os.environ.get("EMBED_FILE", "Qwen3-Embedding-0.6B-Q8_0.gguf")
EMBED_REVISION = os.environ.get(
    "EMBED_REVISION",
    "370f27d7550e0def9b39c1f16d3fbaa13aa67728",
)
EMBED_ALIAS = os.environ.get("EMBED_ALIAS", "text-embedding-ada-002")
EMBED_CTX = int(os.environ.get("EMBED_CTX", "8192"))
EMBED_BATCH = int(os.environ.get("EMBED_BATCH", "8192"))
PORT = int(os.environ.get("PORT", "7860"))
GRADIO_PORT = int(os.environ.get("GRADIO_PORT", "7862"))
# Qwen3-Coder-30B-A3B declares max_position_embeddings 262144 with
# rope_scaling null, so 32768 needs no YaRN and costs no quality. The previous
# 8192 window was already tight: the Action prompt measured 6909 tokens and the
# model then had to fit a 1000+ token program into what was left.
CTX = int(os.environ.get("CPU_CTX", "32768"))
BATCH = int(os.environ.get("CPU_BATCH", "2048"))
UBATCH = int(os.environ.get("CPU_UBATCH", "512"))
# The Action Agent must be able to emit the official multi-function program.
# A 512-token ceiling truncated valid JavaScript into a parse error, so the cap
# is raised rather than the upstream response contract narrowed. Curriculum,
# Critic and SkillManager still use the smaller default.
MAX_TOKENS_CAP = int(os.environ.get("MAX_TOKENS_CAP", "2048"))
DEFAULT_MAX_TOKENS = int(os.environ.get("DEFAULT_MAX_TOKENS", "384"))
API_KEY = os.environ.get("API_KEY", "").strip()
PLAYER_MODE = os.environ.get("PLAYER_MODE", "0") == "1"
PRELOAD = os.environ.get("CPU_PRELOAD", "1") == "1"

# Server-only mode. The brain lives in the Space `Erinaldordso/Mine`; this Space
# keeps the Minecraft server, the Mineflayer bridge and the Voyager runner, and
# nothing else. Every role the bot needs -- Action, Curriculum, Critic, QA,
# SkillManager -- plus embeddings is served at VOYAGER_BRAIN_BASE_URL.
#
# Measured reason: the local Granite aux and the local Qwen main were both
# loading here even though no caller used them. The Voyager runner sets
# OPENAI_API_BASE to the Mine endpoint, so its chat AND embedding calls already
# left this Space; the only remaining local user was the Gemini bot's embeddings.
# Holding two GGUF models resident for that cost RAM the Java server needs, and
# a failed local load surfaced as `aux_ready: false` in /health, which reads as a
# broken bot when it is only an unused, broken loader.
#
# SERVER_ONLY=1 skips loading them and makes the chat route refuse instead of
# silently starting an 18 GB download on first request.
SERVER_ONLY = os.environ.get("MINECRAFT_SERVER_ONLY", "1") == "1"
BRAIN_BASE_URL = os.environ.get(
    "VOYAGER_BRAIN_BASE_URL", "https://erinaldordso-mine.hf.space/v1"
)

@spaces.GPU(duration=1)
def _zerogpu_compatibility_sentinel():
    # Never invoked. Presence only satisfies the ZeroGPU host scanner.
    return {"gpu_used": False, "inference_backend": "cpu-only"}

PLAYER_SYSTEM = os.environ.get(
    "PLAYER_SYSTEM_PROMPT",
    (
        "Voce e o cerebro estrategico do jogador autonomo iaautono em Minecraft Bedrock Survival. "
        "A cada chamada escolha UMA unica proxima acao curta, segura, verificavel e executavel. "
        "Use somente fatos presentes no estado real recebido; nunca invente item, recurso, entidade ou coordenada. "
        "Prioridade de emergencia: se estiver morto, respawn; se estiver se afogando ou sem ar, sair da agua/subir; "
        "depois preservar vida, comida e equipamentos. "
        "Progressao permanente: madeira -> bancada -> ferramentas de pedra -> fornalha/comida -> ferro -> "
        "escudo/armas/armadura -> base segura e armazenamento -> fazenda irrigada com replantio -> "
        "diamante -> encantamentos/reparo -> Nether -> netherite -> infraestrutura e estoques avancados. "
        "Nao ataque jogadores nem destrua construcoes de terceiros. Nao use comandos administrativos ou cheats. "
        "O controlador local executa fisica, movimento, mineracao, crafting, colocacao, inventario, combate contra mobs, "
        "respawn e reconexao; voce decide apenas o objetivo imediato. "
        "Se uma acao anterior falhou, escolha uma alternativa segura em vez de repetir cegamente. "
        "Nunca declare sucesso sem evidencia no estado. Nao fique ocioso se houver progresso survival seguro possivel. "
        "Responda estritamente no formato solicitado pelo cliente e mantenha o motivo curto."
    ),
)

def _read_text(path: str) -> str:
    try:
        return pathlib.Path(path).read_text().strip()
    except Exception:
        return ""

def _cgroup_cpu_quota() -> int | None:
    raw = _read_text("/sys/fs/cgroup/cpu.max")
    if raw:
        parts = raw.split()
        if len(parts) >= 2 and parts[0] != "max":
            try:
                q, period = int(parts[0]), int(parts[1])
                if q > 0 and period > 0:
                    return max(1, int(math.floor(q / period)))
            except Exception:
                pass
    return None

def _affinity_cpus() -> int:
    try:
        return max(1, len(os.sched_getaffinity(0)))
    except Exception:
        return max(1, os.cpu_count() or 1)

def effective_cpus() -> int:
    affinity = _affinity_cpus()
    quota = _cgroup_cpu_quota()
    return min(affinity, quota) if quota else affinity

def _memory_limit_gib() -> float | None:
    for path in ("/sys/fs/cgroup/memory.max", "/sys/fs/cgroup/memory/memory.limit_in_bytes"):
        raw = _read_text(path)
        if raw and raw != "max":
            try:
                value = int(raw)
                if 0 < value < (1 << 60):
                    return round(value / (1024**3), 2)
            except Exception:
                pass
    return None

THREADS = int(os.environ.get("CPU_THREADS", "16"))
THREADS = max(1, min(THREADS, effective_cpus()))
THREADS_BATCH = int(os.environ.get("CPU_THREADS_BATCH", "16"))
THREADS_BATCH = max(1, min(THREADS_BATCH, effective_cpus()))
FLASH_ATTN = os.environ.get("CPU_FLASH_ATTN", "1") == "1"
CGROUP_CPU_QUOTA = _cgroup_cpu_quota()
MEMORY_LIMIT_GIB = _memory_limit_gib()
os.environ["OMP_NUM_THREADS"] = str(THREADS)
os.environ["OPENBLAS_NUM_THREADS"] = str(THREADS)
os.environ["MKL_NUM_THREADS"] = str(THREADS)
os.environ["NUMEXPR_NUM_THREADS"] = str(THREADS)
os.environ["OMP_DYNAMIC"] = "FALSE"

_model = None
_model_path: str | None = None
_model_error = ""
_model_load_seconds = 0.0
_model_loaded_at = 0.0
_loading = False
_load_lock = threading.Lock()
_llm_lock = threading.Lock()

_aux_model = None
_aux_model_path: str | None = None
_aux_model_error = ""
_aux_model_load_seconds = 0.0
_aux_model_loaded_at = 0.0
_aux_loading = False
_aux_load_lock = threading.Lock()

_embed_model = None
_embed_model_path: str | None = None
_embed_error = ""
_embed_loading = False
_embed_load_seconds = 0.0
_embed_load_lock = threading.Lock()
_embed_lock = threading.Lock()
_token_decoder = None

def _load_model() -> None:
    global _model, _model_path, _model_error, _model_load_seconds, _model_loaded_at, _loading
    if SERVER_ONLY:
        # No local brain in this Space. See the SERVER_ONLY note above.
        raise RuntimeError(
            "this Space runs the Minecraft server only; the brain is at "
            f"{BRAIN_BASE_URL}"
        )
    if _model is not None:
        return
    with _load_lock:
        if _model is not None:
            return
        _loading = True
        started = time.perf_counter()
        try:
            print(
                f"[cpu] loading {MODEL_REPO}/{MODEL_FILE} threads={THREADS} "
                f"ctx={CTX} batch={BATCH}/{UBATCH} n_gpu_layers=0",
                flush=True,
            )
            _model_path = hf_hub_download(
                repo_id=MODEL_REPO,
                filename=MODEL_FILE,
                revision=MODEL_REVISION,
                cache_dir=HF_CACHE_DIR,
            )
            from llama_cpp import Llama
            _model = Llama(
                model_path=_model_path,
                n_ctx=CTX,
                n_threads=THREADS,
                n_threads_batch=THREADS_BATCH,
                n_gpu_layers=0,
                n_batch=BATCH,
                n_ubatch=UBATCH,
                flash_attn=FLASH_ATTN,
                offload_kqv=False,
                op_offload=False,
                use_mmap=True,
                use_mlock=False,
                numa=False,
                no_perf=True,
                verbose=False,
            )
            _model_load_seconds = time.perf_counter() - started
            _model_loaded_at = time.time()
            _model_error = ""
            print(f"[cpu] ready load_seconds={_model_load_seconds:.2f}", flush=True)
        except Exception as exc:
            _model_error = repr(exc)
            print(f"[cpu] load failed: {_model_error}", flush=True)
        finally:
            _loading = False

def _get_model(timeout: float = 900.0):
    if _model is not None:
        return _model
    if not _loading:
        _load_model()
    deadline = time.time() + timeout
    while time.time() < deadline:
        if _model is not None:
            return _model
        if not _loading and _model_error:
            raise RuntimeError(_model_error)
        time.sleep(0.25)
    raise RuntimeError("model readiness timeout")


def _load_aux_model() -> None:
    global _aux_model, _aux_model_path, _aux_model_error, _aux_model_load_seconds, _aux_model_loaded_at, _aux_loading
    if SERVER_ONLY:
        # Granite used to be resident here as the QA/SkillManager model. Those
        # roles are served by the Mine Space now, so this loader is dead code
        # and is refused rather than left to download 2 GB on a stray request.
        raise RuntimeError(
            "this Space runs the Minecraft server only; the auxiliary model is "
            f"served at {BRAIN_BASE_URL}"
        )
    if _aux_model is not None:
        return
    with _aux_load_lock:
        if _aux_model is not None:
            return
        _aux_loading = True
        started = time.perf_counter()
        try:
            print(
                f"[cpu-aux] loading {AUX_MODEL_REPO}/{AUX_MODEL_FILE} threads={THREADS} "
                f"ctx={AUX_CTX} batch={BATCH}/{UBATCH} n_gpu_layers=0",
                flush=True,
            )
            _aux_model_path = hf_hub_download(
                repo_id=AUX_MODEL_REPO,
                filename=AUX_MODEL_FILE,
                revision=AUX_MODEL_REVISION,
                cache_dir=HF_CACHE_DIR,
            )
            from llama_cpp import Llama
            _aux_model = Llama(
                model_path=_aux_model_path,
                n_ctx=AUX_CTX,
                n_threads=THREADS,
                n_threads_batch=THREADS_BATCH,
                n_gpu_layers=0,
                n_batch=BATCH,
                n_ubatch=UBATCH,
                flash_attn=FLASH_ATTN,
                offload_kqv=False,
                op_offload=False,
                use_mmap=True,
                use_mlock=False,
                numa=False,
                no_perf=True,
                verbose=False,
            )
            _aux_model_load_seconds = time.perf_counter() - started
            _aux_model_loaded_at = time.time()
            _aux_model_error = ""
            print(f"[cpu-aux] ready load_seconds={_aux_model_load_seconds:.2f}", flush=True)
        except Exception as exc:
            _aux_model_error = repr(exc)
            print(f"[cpu-aux] load failed: {_aux_model_error}", flush=True)
        finally:
            _aux_loading = False


def _get_aux_model(timeout: float = 300.0):
    if _aux_model is not None:
        return _aux_model
    if not _aux_loading:
        _load_aux_model()
    deadline = time.time() + timeout
    while time.time() < deadline:
        if _aux_model is not None:
            return _aux_model
        if not _aux_loading and _aux_model_error:
            raise RuntimeError(_aux_model_error)
        time.sleep(0.25)
    raise RuntimeError("aux model readiness timeout")


def _select_chat_model(requested_model: str):
    if SERVER_ONLY:
        # Fail with a redirect instead of a load attempt. A caller that still
        # names a chat model here is pointed at the Space that actually serves
        # it, so the misroute is obvious in the response body and in the log.
        raise RuntimeError(
            f"chat is not served by this Space (model requested: {requested_model}); "
            f"the brain is at {BRAIN_BASE_URL}"
        )
    if requested_model in AUX_MODEL_ALIASES:
        return _get_aux_model(), "aux"
    if requested_model in MODEL_ALIASES:
        return _get_model(), "main"
    raise ValueError(f"Unknown model: {requested_model}")


def _load_embed_model() -> None:
    global _embed_model, _embed_model_path, _embed_error, _embed_loading, _embed_load_seconds
    if SERVER_ONLY:
        # Embeddings moved to the Mine Space with the rest of the brain. Both
        # Spaces load the same Qwen3-Embedding-0.6B and both return 1024-dim
        # vectors, so the existing Chroma indexes stay valid.
        raise RuntimeError(
            "this Space runs the Minecraft server only; embeddings are served at "
            f"{BRAIN_BASE_URL}"
        )
    if _embed_model is not None:
        return
    with _embed_load_lock:
        if _embed_model is not None:
            return
        _embed_loading = True
        started = time.perf_counter()
        try:
            _embed_model_path = hf_hub_download(
                repo_id=EMBED_REPO,
                filename=EMBED_FILE,
                revision=EMBED_REVISION,
                cache_dir=HF_CACHE_DIR,
            )
            from llama_cpp import Llama
            from llama_cpp import llama_cpp as llama_backend
            _embed_model = Llama(
                model_path=_embed_model_path,
                n_ctx=EMBED_CTX,
                n_threads=THREADS,
                n_threads_batch=THREADS_BATCH,
                n_gpu_layers=0,
                n_batch=EMBED_BATCH,
                n_ubatch=EMBED_BATCH,
                embedding=True,
                pooling_type=llama_backend.LLAMA_POOLING_TYPE_LAST,
                flash_attn=True,
                offload_kqv=False,
                op_offload=False,
                use_mmap=True,
                use_mlock=False,
                numa=False,
                verbose=False,
            )
            _embed_load_seconds = time.perf_counter() - started
            _embed_error = ""
            print(f"[cpu] embedding ready load_seconds={_embed_load_seconds:.2f}", flush=True)
        except Exception as exc:
            _embed_error = repr(exc)
            print(f"[cpu] embedding load failed: {_embed_error}", flush=True)
        finally:
            _embed_loading = False

def _get_embed_model(timeout: float = 300.0):
    if _embed_model is not None:
        return _embed_model
    if not _embed_loading:
        _load_embed_model()
    deadline = time.time() + timeout
    while time.time() < deadline:
        if _embed_model is not None:
            return _embed_model
        if not _embed_loading and _embed_error:
            raise RuntimeError(_embed_error)
        time.sleep(0.25)
    raise RuntimeError("embedding model readiness timeout")

def _auth(authorization: str | None) -> None:
    if API_KEY and authorization != f"Bearer {API_KEY}":
        raise HTTPException(status_code=401, detail="Invalid API key")

def _get_token_decoder():
    global _token_decoder
    if _token_decoder is None:
        _token_decoder = tiktoken.get_encoding("cl100k_base")
    return _token_decoder

def _normalize_embedding_inputs(inputs: Any) -> list[str]:
    if isinstance(inputs, str):
        return [inputs]
    if isinstance(inputs, list) and all(isinstance(item, str) for item in inputs):
        return inputs

    decoder = _get_token_decoder()
    if isinstance(inputs, list) and all(isinstance(item, int) for item in inputs):
        return [decoder.decode(inputs)]
    if isinstance(inputs, list) and all(
        isinstance(item, list) and all(isinstance(token, int) for token in item)
        for item in inputs
    ):
        return [decoder.decode(item) for item in inputs]

    raise HTTPException(
        status_code=400,
        detail="input must be a string, list of strings, token list, or list of token lists",
    )

def _messages(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
    out: list[dict[str, Any]] = []
    for raw_message in messages:
        if not isinstance(raw_message, dict):
            continue
        message = dict(raw_message)
        tool_calls = message.get("tool_calls")
        if isinstance(tool_calls, list):
            normalized_calls: list[dict[str, Any]] = []
            for raw_call in tool_calls:
                if not isinstance(raw_call, dict):
                    continue
                call = dict(raw_call)
                function = call.get("function")
                if isinstance(function, dict):
                    function = dict(function)
                    arguments = function.get("arguments")
                    if isinstance(arguments, str):
                        try:
                            parsed_arguments = json.loads(arguments)
                            if isinstance(parsed_arguments, dict):
                                function["arguments"] = parsed_arguments
                        except json.JSONDecodeError:
                            pass
                    call["function"] = function
                normalized_calls.append(call)
            message["tool_calls"] = normalized_calls
        out.append(message)
    if PLAYER_MODE:
        if out and out[0].get("role") == "system":
            out[0]["content"] = PLAYER_SYSTEM + "\n\n" + str(out[0].get("content") or "")
        else:
            out.insert(0, {"role": "system", "content": PLAYER_SYSTEM})
    return out

def _kwargs(payload: dict[str, Any], stream: bool) -> dict[str, Any]:
    max_tokens = payload.get("max_tokens", payload.get("max_completion_tokens", DEFAULT_MAX_TOKENS))
    out: dict[str, Any] = {
        "messages": _messages(payload.get("messages") or []),
        "max_tokens": max(1, min(int(max_tokens), MAX_TOKENS_CAP)),
        "temperature": float(payload.get("temperature", 0.0)),
        "top_p": float(payload.get("top_p", 1.0)),
        "top_k": int(payload.get("top_k", 0)),
        "min_p": float(payload.get("min_p", 0.0)),
        "presence_penalty": float(payload.get("presence_penalty", 0.0)),
        "stream": stream,
    }
    for key in ("tools", "tool_choice", "stop", "response_format", "seed"):
        if key in payload and payload[key] is not None:
            out[key] = payload[key]
    if PLAYER_MODE and not payload.get("tools") and "response_format" not in out:
        out["response_format"] = {"type": "json_object"}
    return out

def _tool_argument_config(tools: list[dict[str, Any]] | None, function_name: str) -> dict[str, Any]:
    for tool in tools or []:
        if not isinstance(tool, dict) or tool.get("type") != "function":
            continue
        function = tool.get("function")
        if not isinstance(function, dict) or function.get("name") != function_name:
            continue
        parameters = function.get("parameters")
        if isinstance(parameters, dict):
            properties = parameters.get("properties", parameters)
            return properties if isinstance(properties, dict) else {}
    return {}


def _convert_tool_value(value: str, name: str, config: dict[str, Any]) -> Any:
    spec = config.get(name, {}) if isinstance(config, dict) else {}
    param_type = spec.get("type", "string") if isinstance(spec, dict) else "string"
    if isinstance(param_type, list):
        param_type = next((item for item in param_type if item != "null"), "string")
    param_type = str(param_type).lower()
    if param_type == "null":
        return None
    if param_type.startswith("int"):
        try:
            return int(value)
        except (TypeError, ValueError):
            return value
    if param_type.startswith(("num", "float")):
        try:
            number = float(value)
            return int(number) if number.is_integer() else number
        except (TypeError, ValueError):
            return value
    if param_type in {"boolean", "bool", "binary"}:
        lowered = value.lower()
        return lowered == "true" if lowered in {"true", "false"} else False
    if param_type in {"object", "array", "arr"} or param_type.startswith(("dict", "list")):
        try:
            return json.loads(value)
        except (TypeError, ValueError, json.JSONDecodeError):
            try:
                return ast.literal_eval(value)
            except (ValueError, SyntaxError):
                return value
    return value


_TOOL_CALL_RE = re.compile(r"<tool_call>(.*?)</tool_call>|<tool_call>(.*?)$", re.DOTALL)
_TOOL_FUNCTION_RE = re.compile(r"<function=(.*?)</function>|<function=(.*)$", re.DOTALL)
_TOOL_PARAMETER_RE = re.compile(
    r"<parameter=(.*?)(?:</parameter>|(?=<parameter=)|(?=</function>)|$)",
    re.DOTALL,
)


def _parse_qwen3_coder_tool_calls(
    model_output: str,
    tools: list[dict[str, Any]] | None,
) -> tuple[str | None, list[dict[str, Any]]]:
    # Mirrors Qwen's official qwen3coder_tool_parser semantics for non-streaming output.
    if "<function=" not in model_output:
        return model_output, []

    matches = _TOOL_CALL_RE.findall(model_output)
    raw_tool_calls = [left if left else right for left, right in matches] or [model_output]

    raw_functions: list[str] = []
    for raw_tool_call in raw_tool_calls:
        for complete, partial in _TOOL_FUNCTION_RE.findall(raw_tool_call):
            raw_functions.append(complete if complete else partial)

    parsed_calls: list[dict[str, Any]] = []
    for function_call in raw_functions:
        if ">" not in function_call:
            continue
        function_name, parameters = function_call.split(">", 1)
        function_name = function_name.strip()
        if not function_name:
            continue
        config = _tool_argument_config(tools, function_name)
        arguments: dict[str, Any] = {}
        for match_text in _TOOL_PARAMETER_RE.findall(parameters):
            if ">" not in match_text:
                continue
            param_name, param_value = match_text.split(">", 1)
            if param_value.startswith("\n"):
                param_value = param_value[1:]
            if param_value.endswith("\n"):
                param_value = param_value[:-1]
            param_name = param_name.strip()
            arguments[param_name] = _convert_tool_value(param_value, param_name, config)
        parsed_calls.append(
            {
                "id": f"call_{uuid.uuid4().hex[:24]}",
                "type": "function",
                "function": {
                    "name": function_name,
                    "arguments": json.dumps(arguments, ensure_ascii=False),
                },
            }
        )

    if not parsed_calls:
        return model_output, []

    content_index = model_output.find("<tool_call>")
    if content_index < 0:
        content_index = model_output.find("<function=")
    content = model_output[:content_index] if content_index >= 0 else ""
    return content if content else None, parsed_calls


def _nonstream(payload: dict[str, Any]) -> dict[str, Any]:
    requested_model = str(payload.get("model") or MODEL_ALIAS)
    llm, model_role = _select_chat_model(requested_model)
    started = time.perf_counter()
    with _llm_lock:
        print(f"[chat] start role={model_role} model={requested_model} temperature={payload.get('temperature', 0.0)}", flush=True)
        result = llm.create_chat_completion(**_kwargs(payload, False))
    result = dict(result)
    if payload.get("tools") and result.get("choices"):
        choice = result["choices"][0]
        message = choice.get("message") or {}
        content = message.get("content")
        if isinstance(content, str):
            parsed_content, tool_calls = _parse_qwen3_coder_tool_calls(
                content,
                payload.get("tools"),
            )
            if tool_calls:
                message["content"] = parsed_content
                message["tool_calls"] = tool_calls
                choice["message"] = message
                choice["finish_reason"] = "tool_calls"
    result["model"] = requested_model
    result["backend"] = "cpu-only"
    result["program_metrics"] = {
        "elapsed_seconds": round(time.perf_counter() - started, 4),
        "threads": THREADS,
        "context": AUX_CTX if model_role == "aux" else CTX,
        "model_role": model_role,
        "n_gpu_layers": 0,
    }
    print(f"[chat] done role={model_role} model={requested_model} metrics={result['program_metrics']} usage={result.get('usage', {})}", flush=True)
    return result

def _stream(payload: dict[str, Any]) -> Iterator[str]:
    requested_model = str(payload.get("model") or MODEL_ALIAS)
    llm, _model_role = _select_chat_model(requested_model)
    with _llm_lock:
        for chunk in llm.create_chat_completion(**_kwargs(payload, True)):
            item = dict(chunk)
            item["model"] = requested_model
            item["backend"] = "cpu-only"
            yield "data: " + json.dumps(item, ensure_ascii=False) + "\n\n"
    yield "data: [DONE]\n\n"

# Computed once: the answer cannot change while the process lives, and the
# status endpoint must not walk the package directory on every request.
_CPU_ONLY = cpu_only_report()
print(f"[cpu] {cpu_only_summary(_CPU_ONLY)}", flush=True)

api = FastAPI(title="Programa Voyager OpenAI Endpoint", version="6.0.0")

_supervisor_lock = threading.Lock()
_supervisor_recoveries = 0
_supervisor_last_action = "startup"
_supervisor_last_check = 0.0
# Set during shutdown BEFORE any save/stop, so the 30s loop cannot restart a
# bot or the Java server mid-shutdown. All supervisor sleeps wait on it.
_supervisor_stop = threading.Event()


def _service_supervisor() -> None:
    global _supervisor_recoveries, _supervisor_last_action, _supervisor_last_check
    backoff = 10
    while not _supervisor_stop.is_set():
        try:
            if _supervisor_stop.wait(30):
                break
            _supervisor_last_check = time.time()
            mc = STACK.status()
            processes = mc.get("processes") or {}
            critical = ("minecraft", "viaproxy", "friends")
            dead = [name for name in critical if name in processes and not processes[name].get("running")]
            missing = [name for name in critical if mc.get("state") == "running" and name not in processes]
            if mc.get("state") == "error" or dead or missing:
                with _supervisor_lock:
                    _supervisor_recoveries += 1
                    _supervisor_last_action = f"minecraft recovery #{_supervisor_recoveries}: dead={dead} missing={missing} state={mc.get('state')}"
                    VOYAGER.stop()
                    GEMINI_BOT.stop()
                    ALTOCLEF.stop()
                    STACK.stop()
                    if _supervisor_stop.wait(backoff):
                        break
                    STACK.start_async()
                    # Voyager stays available on demand but never auto-starts:
                    # VOYAGER_AUTOSTART defaults to 0 (AltoClef is the default
                    # controller). The recovery path must respect the same
                    # flag, otherwise a restart silently re-enables it.
                    if os.environ.get("VOYAGER_AUTOSTART", "0") == "1":
                        VOYAGER.start_async()
                    if os.environ.get("GEMINI_BOT_AUTOSTART", "1") == "1":
                        GEMINI_BOT.start_async()
                    if os.environ.get("ALTOCLEF_AUTOSTART", "1") == "1":
                        ALTOCLEF.start_async()
                backoff = min(backoff * 2, 120)
                continue

            vs = VOYAGER.status()
            if (os.environ.get("VOYAGER_AUTOSTART", "0") == "1"
                    and mc.get("state") == "running" and vs.get("state") in {"error", "exited", "stopped"} and not vs.get("running")):
                with _supervisor_lock:
                    _supervisor_recoveries += 1
                    _supervisor_last_action = f"voyager recovery #{_supervisor_recoveries}: state={vs.get('state')}"
                    VOYAGER.start_async()
                backoff = min(backoff * 2, 120)
                continue

            gs = GEMINI_BOT.status()
            if (os.environ.get("GEMINI_BOT_AUTOSTART", "1") == "1" and mc.get("state") == "running"
                    and gs.get("state") in {"error", "stopped"} and not gs.get("running")):
                GEMINI_BOT.start_async()

            als = ALTOCLEF.status()
            if (os.environ.get("ALTOCLEF_AUTOSTART", "1") == "1" and mc.get("state") == "running"
                    and als.get("state") in {"error", "stopped"} and not als.get("running")):
                ALTOCLEF.start_async()

            if mc.get("access_state") == "degraded":
                # nether2rak already retries its Xbox session internally. Report
                # loss of access without restarting the Minecraft world or bot.
                _supervisor_last_action = "xbox session degraded; relay retrying internally"
                continue

            if mc.get("state") == "running" and vs.get("running"):
                backoff = 10
                _supervisor_last_action = "healthy"
        except Exception as exc:
            _supervisor_last_action = f"supervisor exception: {type(exc).__name__}: {str(exc)[:160]}"
            if _supervisor_stop.wait(min(backoff, 120)):
                break
            backoff = min(backoff * 2, 120)


@api.on_event("startup")
def _startup() -> None:
    if SERVER_ONLY:
        # Only the Minecraft server, the Mineflayer bridge and the Voyager
        # runner start here. No local GGUF is loaded and no chat preload thread
        # is spawned, so the RAM budget belongs to Java.
        print(
            "[startup] server-only mode: no local model will be loaded; the "
            f"brain is at {BRAIN_BASE_URL}",
            flush=True,
        )
    if PRELOAD and not SERVER_ONLY:
        threading.Thread(target=_load_model, daemon=True, name="cpu-model-preload").start()
        threading.Thread(target=_load_embed_model, daemon=True, name="cpu-embedding-preload").start()
    if AUX_PRELOAD and not SERVER_ONLY:
        threading.Thread(target=_load_aux_model, daemon=True, name="cpu-aux-model-preload").start()
    if os.environ.get("MINECRAFT_AUTOSTART", "1") == "1":
        STACK.start_async()
    if os.environ.get("VOYAGER_AUTOSTART", "0") == "1":
        VOYAGER.start_async()
    if os.environ.get("GEMINI_BOT_AUTOSTART", "1") == "1":
        GEMINI_BOT.start_async()
    if os.environ.get("ALTOCLEF_AUTOSTART", "1") == "1":
        ALTOCLEF.start_async()
    threading.Thread(target=_service_supervisor, daemon=True, name="service-supervisor").start()

@api.on_event("shutdown")
def _shutdown() -> None:
    # Graceful stop on SIGTERM/redeploy: stop the supervisor FIRST so the 30s
    # loop cannot restart a bot or Java mid-shutdown, then flush the world and
    # stop the Java server (vanilla saves again on "stop"). Best-effort and
    # local only; no remote command endpoint is exposed for this.
    _supervisor_stop.set()
    try:
        STACK.save_all(timeout=20.0)
    except Exception:
        pass
    for runner in (VOYAGER, GEMINI_BOT, ALTOCLEF):
        try:
            runner.stop()
        except Exception:
            pass
    try:
        STACK.stop()
    except Exception:
        pass


@api.get("/")
def root():
    return {
        "status": "ok",
        "service": "Minecraft server and autonomous bot runners" if SERVER_ONLY else "CPU model endpoint",
        "api": "/v1",
        "health": "/health",
    }

@api.get("/health")
def health():
    if SERVER_ONLY:
        # No local weights exist, so there is nothing to report as ready or
        # broken. Reporting `aux_ready: false` here (the old behaviour) was
        # actively misleading: it reads as a failed load when the truth is that
        # no local load is intended. The role map points at the real host.
        return {
            "status": "ok",
            "mode": "server-only",
            "server_only": True,
            "brain_base_url": BRAIN_BASE_URL,
            "local_models": False,
            "goal": os.environ.get("MINECRAFT_GAME_GOAL") or None,
            "roles": {
                "chat": f"{BRAIN_BASE_URL}/chat/completions",
                "embeddings": f"{BRAIN_BASE_URL}/embeddings",
            },
            "backend": "cpu-only",
            "n_gpu_layers": 0,
            "supervisor_recoveries": _supervisor_recoveries,
            "supervisor_last_action": _supervisor_last_action,
            "supervisor_last_check": _supervisor_last_check or None,
        }
    return {
        "status": "ok",
        "ready": _model is not None,
        "loading": _loading,
        "model": MODEL_ALIAS,
        "aux_model": AUX_MODEL_ALIAS,
        "aux_ready": _aux_model is not None,
        "aux_loading": _aux_loading,
        "aux_error": _aux_model_error or None,
        "embedding_model": EMBED_ALIAS,
        "embedding_ready": _embed_model is not None,
        "embedding_loading": _embed_loading,
        "embedding_error": _embed_error or None,
        "backend": "cpu-only",
        "n_gpu_layers": 0,
        "supervisor_recoveries": _supervisor_recoveries,
        "supervisor_last_action": _supervisor_last_action,
        "supervisor_last_check": _supervisor_last_check or None,
        "error": _model_error or None,
    }

@api.get("/status")
def status():
    memory = psutil.virtual_memory()
    process = psutil.Process()
    if SERVER_ONLY:
        return {
            "status": "ok",
            "mode": "server-only",
            "server_only": True,
            "brain_base_url": BRAIN_BASE_URL,
            "local_models": False,
            "roles": {
                "chat": f"{BRAIN_BASE_URL}/chat/completions",
                "embeddings": f"{BRAIN_BASE_URL}/embeddings",
            },
            "backend": "cpu-only",
            "cpu_only": _CPU_ONLY["cpu_only"],
            "gpu_offload_supported": _CPU_ONLY["gpu_offload_supported"],
            "gpu_backend_libraries": _CPU_ONLY["gpu_backend_libraries"],
            "cuda_libraries_present": _CPU_ONLY["cuda_libraries_present"],
            "gpu_enabled": bool(_CPU_ONLY["gpu_offload_supported"] is True),
            "n_gpu_layers": _CPU_ONLY["n_gpu_layers_requested"],
            "threads": THREADS,
            "threads_batch": THREADS_BATCH,
            "effective_cpus": effective_cpus(),
            "cgroup_cpu_quota": CGROUP_CPU_QUOTA,
            "player_mode": PLAYER_MODE,
            "memory_limit_gib": MEMORY_LIMIT_GIB,
            "system_memory_total_gib": round(memory.total / (1024**3), 2),
            "system_memory_available_gib": round(memory.available / (1024**3), 2),
            "process_rss_gib": round(process.memory_info().rss / (1024**3), 2),
        }
    return {
        "status": "ok",
        "ready": _model is not None,
        "loading": _loading,
        "model": MODEL_ALIAS,
        "model_repo": MODEL_REPO,
        "model_file": MODEL_FILE,
        "model_revision": MODEL_REVISION,
        "quantization": "Q4_K_M",
        "model_family": "Granite-4.0-H-Tiny",
        "total_params_b": 7.0,
        "active_params_b": 1.0,
        "reasoning_policy": "voyager-granite-h-tiny-cpu-low-latency",
        "aux_model": AUX_MODEL_ALIAS,
        "aux_model_repo": AUX_MODEL_REPO,
        "aux_model_file": AUX_MODEL_FILE,
        "aux_model_revision": AUX_MODEL_REVISION,
        "aux_context": AUX_CTX,
        "aux_ready": _aux_model is not None,
        "aux_loading": _aux_loading,
        "aux_load_seconds": round(_aux_model_load_seconds, 3),
        "aux_loaded_at": _aux_model_loaded_at or None,
        "aux_error": _aux_model_error or None,
        "embedding_repo": EMBED_REPO,
        "embedding_file": EMBED_FILE,
        "embedding_revision": EMBED_REVISION,
        "embedding_model": EMBED_ALIAS,
        "embedding_ready": _embed_model is not None,
        "embedding_loading": _embed_loading,
        "embedding_load_seconds": round(_embed_load_seconds, 3),
        "embedding_error": _embed_error or None,
        "backend": "cpu-only",
        # Derived from the linked binary and the backend libraries actually
        # present, not a literal. See cpu_backend.cpu_only_report().
        "cpu_only": _CPU_ONLY["cpu_only"],
        "gpu_offload_supported": _CPU_ONLY["gpu_offload_supported"],
        "gpu_backend_libraries": _CPU_ONLY["gpu_backend_libraries"],
        "cuda_libraries_present": _CPU_ONLY["cuda_libraries_present"],
        "gpu_enabled": bool(_CPU_ONLY["gpu_offload_supported"] is True),
        "n_gpu_layers": _CPU_ONLY["n_gpu_layers_requested"],
        "context": CTX,
        "context_mode": "native-no-yarn",
        "threads": THREADS,
        "threads_batch": THREADS_BATCH,
        "effective_cpus": effective_cpus(),
        "cgroup_cpu_quota": CGROUP_CPU_QUOTA,
        "batch": BATCH,
        "ubatch": UBATCH,
        "flash_attn": FLASH_ATTN,
        "aux_context": AUX_CTX,
        "default_max_tokens": DEFAULT_MAX_TOKENS,
        "max_tokens_cap": MAX_TOKENS_CAP,
        "player_mode": PLAYER_MODE,
        "memory_limit_gib": MEMORY_LIMIT_GIB,
        "system_memory_total_gib": round(memory.total / (1024**3), 2),
        "system_memory_available_gib": round(memory.available / (1024**3), 2),
        "process_rss_gib": round(process.memory_info().rss / (1024**3), 2),
        "model_load_seconds": round(_model_load_seconds, 3),
        "model_loaded_at": _model_loaded_at or None,
        "model_error": _model_error or None,
    }

@api.get("/v1/models")
def models(authorization: str | None = Header(default=None)):
    _auth(authorization)
    now = int(time.time())
    if SERVER_ONLY:
        # Advertising a chat model this Space will refuse to serve sends clients
        # to the wrong host on the strength of a successful discovery call. The
        # list stays empty and the response names the real endpoint.
        return {
            "object": "list",
            "data": [],
            "server_only": True,
            "brain_base_url": BRAIN_BASE_URL,
        }
    return {
        "object": "list",
        "data": [
            *[
                {"id": alias, "object": "model", "created": now, "owned_by": "Qwen"}
                for alias in sorted(MODEL_ALIASES | AUX_MODEL_ALIASES)
            ],
            {"id": EMBED_ALIAS, "object": "model", "created": now, "owned_by": "Qwen"},
        ],
    }

@api.post("/v1/embeddings")
async def embeddings(request: Request, authorization: str | None = Header(default=None)):
    _auth(authorization)
    if SERVER_ONLY:
        # Refuse before parsing the body. Without this the request would fall
        # through to `_get_embed_model()`, which raises from `_load_embed_model`
        # and is reported as a generic 503 -- a status a caller reads as "the
        # server is briefly busy, retry me" rather than "you are calling the
        # wrong host". The detail names the right host instead.
        raise HTTPException(
            status_code=503,
            detail=(
                "this Space runs the Minecraft server only; embeddings are "
                f"served at {BRAIN_BASE_URL}"
            ),
        )
    try:
        payload = await request.json()
    except Exception as exc:
        raise HTTPException(status_code=400, detail=f"Invalid JSON: {str(exc)[:160]}") from exc
    inputs = payload.get("input")
    texts = _normalize_embedding_inputs(inputs)
    if not texts:
        raise HTTPException(status_code=400, detail="input must not be empty")
    requested_model = str(payload.get("model") or EMBED_ALIAS)
    try:
        model = await run_in_threadpool(_get_embed_model)
        def _run_embedding():
            with _embed_lock:
                vectors, token_count = model.embed(texts, normalize=True, return_count=True)
            return {
                "object": "list",
                "data": [
                    {"object": "embedding", "embedding": vector, "index": index}
                    for index, vector in enumerate(vectors)
                ],
                "model": requested_model,
                "usage": {"prompt_tokens": token_count, "total_tokens": token_count},
            }
        return JSONResponse(await run_in_threadpool(_run_embedding))
    except RuntimeError as exc:
        raise HTTPException(status_code=503, detail=str(exc)[:800]) from exc


@api.get("/minecraft/status")
def minecraft_status():
    data = STACK.status().copy()
    data.pop("microsoft_device_code", None)
    data.pop("log_tail", None)
    return data

@api.get("/voyager/status")
def voyager_status():
    data = VOYAGER.status().copy()
    data.pop("log_tail", None)
    return data


@api.get("/voyager/log")
def voyager_log(lines: int = 200, find: str | None = None):
    """Expose the runner log.

    /voyager/status deliberately drops log_tail: that payload is polled by the
    dashboard and the log is large. Dropping it there left the loop's own
    traceback unobservable from outside the Space -- the Hugging Face log
    stream can lag hours behind, so a wedged runner looked identical to a
    healthy one. This endpoint reads the same file the status builds from, on
    demand, without changing the status contract.
    """
    lines = max(1, min(int(lines or 200), 5000))
    tail = VOYAGER.tail(lines)
    if find:
        kept = [row for row in tail.splitlines() if find in row]
        tail = "\n".join(kept)
    return {
        "lines": lines,
        "find": find,
        "bytes": len(tail),
        "log": tail,
    }


@api.get("/voyager/probe")
def voyager_probe(seconds: int = 6):
    """Is the runner child alive, and is it using CPU or blocked?

    A wedged runner and a slow one look identical from the outside: both keep
    the process alive and write nothing. Sampling utime/stime across a short
    interval separates them. A process whose CPU counters do not move at all
    over several seconds while producing no output is blocked on a syscall, not
    thinking -- which is the distinction that took three rounds to establish by
    reading logs alone.
    """
    seconds = max(1, min(int(seconds or 6), 30))
    snapshot = VOYAGER.probe_runner(seconds)
    return snapshot


@api.get("/gemini-bot/status")
def gemini_bot_status():
    return GEMINI_BOT.status()


@api.get("/opencode-bot/status")
@api.get("/altoclef/status")
def altoclef_status():
    return ALTOCLEF.status()


@api.get("/opencode-bot/log")
def altoclef_log(lines: int = 100):
    return {"log": ALTOCLEF.tail(lines)}

@api.post("/v1/chat/completions")
async def chat(request: Request, authorization: str | None = Header(default=None)):
    _auth(authorization)
    try:
        payload = await request.json()
    except Exception as exc:
        raise HTTPException(status_code=400, detail=f"Invalid JSON: {str(exc)[:160]}") from exc
    if not isinstance(payload.get("messages"), list):
        raise HTTPException(status_code=400, detail="messages must be a list")
    requested_model = str(payload.get("model") or MODEL_ALIAS)
    if requested_model not in (MODEL_ALIASES | AUX_MODEL_ALIASES):
        raise HTTPException(status_code=404, detail=f"Unknown model: {requested_model}")
    if payload.get("stream"):
        return StreamingResponse(
            _stream(payload),
            media_type="text/event-stream",
            headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
        )
    try:
        return JSONResponse(await run_in_threadpool(_nonstream, payload))
    except RuntimeError as exc:
        raise HTTPException(status_code=503, detail=str(exc)[:800]) from exc

with gr.Blocks(title="Programa Voyager LLM Endpoint") as demo:
    gr.Markdown(
        "# Programa — Voyager LLM Endpoint\n"
        "**Granite-4.0-H-Tiny Q4_K_M**, 7B total / ~1B ativo, contexto 8K, 16 threads de decode / 16 de prefill, inferencia 100% CPU."
    )
    output = gr.JSON(label="IA / CPU")
    gr.Button("Status IA / CPU").click(status, outputs=output)
    gr.Markdown("## Minecraft + Xbox Friends")
    mc_output = gr.JSON(label="Minecraft")
    with gr.Row():
        gr.Button("Status Minecraft").click(STACK.status, outputs=mc_output)
        gr.Button("Iniciar Minecraft", variant="primary").click(STACK.start_async, outputs=mc_output)
        gr.Button("Parar Minecraft").click(STACK.stop, outputs=mc_output)
    gr.Markdown("## Voyager (alternativa manual)\nO Voyager continua disponível se você quiser ligar manualmente; o Naldo usa o AltoClef.")
    voyager_output = gr.JSON(label="Voyager")
    with gr.Row():
        gr.Button("Status Voyager").click(VOYAGER.status, outputs=voyager_output)
        gr.Button("Iniciar Voyager").click(VOYAGER.start_async, outputs=voyager_output)
        gr.Button("Parar Voyager").click(VOYAGER.stop, outputs=voyager_output)

    gr.Markdown("## Naldo — jogador autônomo\nO Naldo joga survival sozinho para zerar o jogo (mapa e inventário atual mantidos).")
    altoclef_output = gr.JSON(label="Naldo")
    with gr.Row():
        gr.Button("Status Naldo").click(ALTOCLEF.status, outputs=altoclef_output)
        gr.Button("Iniciar Naldo", variant="primary").click(ALTOCLEF.start_async, outputs=altoclef_output)
        gr.Button("Parar Naldo").click(ALTOCLEF.stop, outputs=altoclef_output)

if __name__ == "__main__":
    print(
        f"[boot] cpu-only model={MODEL_REPO}/{MODEL_FILE} threads={THREADS} "
        f"ctx={CTX} n_gpu_layers=0 mem_limit_gib={MEMORY_LIMIT_GIB}",
        flush=True,
    )
    # Keep the Gradio SDK app on a private port; the public OpenAI-compatible API
    # owns PORT=7860 exactly as in the previously stable deployment.
    demo.queue(default_concurrency_limit=1)
    demo.launch(
        server_name="0.0.0.0",
        server_port=GRADIO_PORT,
        prevent_thread_lock=True,
        ssr_mode=False,
        show_error=True,
    )
    uvicorn.run(api, host="0.0.0.0", port=PORT, log_level="info")