Corrige Qwen3 CPU

.dockerignore

@@ -1,23 +1,13 @@
 .git
 .gitignore
-__pycache__/
+__pycache__
 *.py[cod]
-.pytest_cache/
-.mypy_cache/
-.venv/
-venv/
-.env
-.env.*
-*.key
-*.pem
-*.p12
-*.pfx
-*.crt
-*.cer
-node_modules/
-build/
-dist/
-.coverage
-htmlcov/
-*.gguf
+.pytest_cache
+.mypy_cache
+.venv
+venv
+*.zip
 *.log
+tests
+VALIDATION.txt
+smoke_test.sh

Dockerfile

@@ -1,74 +1,44 @@
-FROM python:3.12-slim AS wheels
-
-ENV PIP_DISABLE_PIP_VERSION_CHECK=1 \
-    CMAKE_ARGS="-DGGML_NATIVE=OFF -DGGML_BLAS=ON -DGGML_BLAS_VENDOR=OpenBLAS -DGGML_LTO=ON" \
-    FORCE_CMAKE=1
-
-RUN apt-get update \
-    && apt-get install -y --no-install-recommends \
-        build-essential cmake ninja-build pkg-config libopenblas-dev \
-    && rm -rf /var/lib/apt/lists/*
-
-WORKDIR /build
-COPY requirements.txt ./
-# Build llama.cpp against OpenBLAS instead of using the portable generic wheel.
-# This targets prompt prefill, the dominant cost for OpenClaude on two vCPUs.
-RUN python -m pip wheel \
-    --no-cache-dir \
-    --no-binary llama-cpp-python \
-    --wheel-dir /wheels \
-    -r requirements.txt
-
-
 FROM python:3.12-slim
 
-ENV DEBIAN_FRONTEND=noninteractive \
-    PYTHONDONTWRITEBYTECODE=1 \
+ENV PYTHONDONTWRITEBYTECODE=1 \
     PYTHONUNBUFFERED=1 \
     PIP_DISABLE_PIP_VERSION_CHECK=1 \
     HF_HUB_DISABLE_PROGRESS_BARS=1 \
-    TOKENIZERS_PARALLELISM=false \
-    HF_HOME=/data/huggingface \
-    GRADIO_SERVER_NAME=0.0.0.0 \
-    GRADIO_SERVER_PORT=7860 \
-    OPENBLAS_NUM_THREADS=2 \
-    OMP_NUM_THREADS=2 \
-    OMP_WAIT_POLICY=PASSIVE \
-    CPU_THREADS=2 \
-    CPU_BATCH_THREADS=2 \
-    N_BATCH=1024 \
-    N_UBATCH=512 \
-    FLASH_ATTN=true \
-    KV_CACHE_TYPE=q8_0 \
-    PRELOAD_MODEL=true \
-    COPY_MODEL_TO_LOCAL=true \
-    MODEL_RUNTIME_DIR=/tmp/qwen-gguf
+    HF_HUB_DISABLE_XET=1 \
+    HF_HUB_ETAG_TIMEOUT=30 \
+    HF_HUB_DOWNLOAD_TIMEOUT=120 \
+    HOME=/home/user \
+    HF_HOME=/home/user/.cache/huggingface \
+    PORT=7860
 
 RUN apt-get update \
-    && apt-get install -y --no-install-recommends \
-        ca-certificates libgomp1 libopenblas0-pthread \
+    && apt-get install -y --no-install-recommends ca-certificates libgomp1 \
     && rm -rf /var/lib/apt/lists/*
 
+RUN useradd --create-home --uid 1000 user \
+    && mkdir -p /app /home/user/.cache/huggingface \
+    && chown -R user:user /app /home/user
+
 WORKDIR /app
+
 COPY requirements.txt ./
-COPY --from=wheels /wheels /wheels
-RUN python -m pip install \
-    --no-cache-dir \
-    --no-index \
-    --find-links=/wheels \
-    -r requirements.txt \
-    && rm -rf /wheels
 
-RUN useradd --create-home --uid 1000 app \
-    && mkdir -p /data/huggingface \
-    && chown -R app:app /app /data
+ARG LLAMA_CPP_VERSION=0.3.34
+
+RUN python -m pip install --no-cache-dir -r requirements.txt \
+    && python -m pip install --no-cache-dir \
+       --only-binary=:all: \
+       --extra-index-url https://abetlen.github.io/llama-cpp-python/whl/cpu \
+       "llama-cpp-python==${LLAMA_CPP_VERSION}" \
+    && python -c "import importlib.metadata, llama_cpp; version=importlib.metadata.version('llama-cpp-python'); print(f'llama-cpp-python CPU wheel OK: {version}')"
+
+COPY --chown=user:user app.py settings.py tooling.py ./
 
-COPY --chown=app:app . .
-USER app
+USER user
 
 EXPOSE 7860
 
-HEALTHCHECK --interval=30s --timeout=5s --start-period=10m --retries=3 \
-    CMD ["python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:7860/ready', timeout=4).read()"]
+HEALTHCHECK --interval=30s --timeout=10s --start-period=15s --retries=3 \
+    CMD python -c "import urllib.request; urllib.request.urlopen('http://127.0.0.1:7860/health', timeout=5).read()" || exit 1
 
-CMD ["python", "app.py"]
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860", "--workers", "1", "--timeout-keep-alive", "65"]

README.md

@@ -1,215 +1,127 @@
 ---
-title: Qwen3 Autonomous CPU OpenAI API
+title: Qwen3 CPU OpenAI API
 emoji: 🧠
-colorFrom: green
-colorTo: blue
+colorFrom: blue
+colorTo: green
 sdk: docker
 app_port: 7860
 pinned: false
 ---
 
-# Qwen3 CPU API for OpenClaude
+# Qwen3 CPU OpenAI API
 
-Backend OpenAI-compatible para executar o OpenClaude em um Hugging Face Space
-CPU Basic. O modelo padrão é o `Qwen3-4B-Instruct-2507` em GGUF Q4_K_M, com
-controlador determinístico de ferramentas, contexto físico limitado a 32K e
-inferência serializada por segurança do `llama.cpp`.
+CPU/RAM-only OpenAI-compatible API for OpenClaude on Hugging Face Docker Spaces.
 
-Endpoints:
+## Runtime model
 
-- `GET /health`
-- `GET /ready`
-- `GET /v1/models`
-- `POST /v1/chat/completions`
-- `GET /web-search?q=...`
-
-O alias estável para clientes é `qwen-coder`.
-
-## Desempenho no Hugging Face
+- `unsloth/Qwen3-4B-Instruct-2507-GGUF`
+- `Qwen3-4B-Instruct-2507-Q4_K_M.gguf`
+- ~2.5 GB GGUF
+- alias: `qwen-coder`
+- default context: `8192`
+- output cap: `2048`
+- CPU threads: `2`
+- GPU layers: `0`
 
-Não existe overclock de CPU dentro de um Space: frequência e hardware pertencem
-ao host do Hugging Face. O Dockerfile atua nos gargalos que o container realmente
-controla:
-
-- compila `llama-cpp-python` com OpenBLAS para acelerar o prefill;
-- usa 2 threads de geração e 2 de batch, alinhadas aos 2 vCPU do CPU Basic;
-- aumenta `n_ubatch` de 128 para 512 e `n_batch` para 1024;
-- habilita Flash Attention;
-- usa KV cache Q8_0, reduzindo aproximadamente pela metade a memória do KV F16;
-- mantém o modelo Q4_K_M para não sacrificar a qualidade das chamadas de ferramenta;
-- reduz prompts de etapas determinísticas e de finalização;
-- envia heartbeats SSE enquanto uma geração CPU longa ainda está em andamento;
-- limita a fila de inferência para evitar acúmulo ilimitado de threads.
-
-Configuração padrão:
-
-```text
-MODEL_PROFILE=smart
-MAX_CONTEXT_TOKENS=32768
-MAX_NEW_TOKENS=2048
-MAX_TOOL_CALL_TOKENS=768
-MAX_COMPACT_TOOL_TOKENS=384
-MAX_TERMINAL_SUMMARY_TOKENS=384
-CPU_THREADS=2
-CPU_BATCH_THREADS=2
-N_BATCH=1024
-N_UBATCH=512
-FLASH_ATTN=true
-KV_CACHE_TYPE=q8_0
-PRELOAD_MODEL=true
-COPY_MODEL_TO_LOCAL=true
-MODEL_RUNTIME_DIR=/tmp/qwen-gguf
-MAX_GENERATION_QUEUE=4
-GENERATION_QUEUE_TIMEOUT_SECONDS=2
-SSE_HEARTBEAT_SECONDS=10
-DEFAULT_TEMPERATURE=0.0
-```
+The model is downloaded at runtime, not at Docker build time.
 
-Para diagnóstico de compatibilidade, `FLASH_ATTN=false` e
-`KV_CACHE_TYPE=f16` restauram o caminho conservador. Se a prioridade absoluta
-for latência, use o perfil menor:
+## Build/OOM correction
 
-```text
-MODEL_PROFILE=fast
-```
-
-Perfis:
-
-| Perfil | Modelo | GGUF | Uso |
-| --- | --- | --- | --- |
-| `smart` | Qwen3-4B-Instruct-2507 | Q4_K_M, ~2,5 GB | melhor qualidade de código/ferramentas |
-| `fast` | Qwen3-1.7B | Q4_K_M | menor latência, menor capacidade |
-
-O endpoint anuncia apenas o alias do perfil realmente carregado; ele não finge
-que o modelo 1.7B está ativo quando o 4B está em memória, ou vice-versa.
-
-O GGUF e o tokenizer padrão são fixados por revisão Git. O download fica no
-bucket persistente `/data`, mas o GGUF é copiado para o disco efêmero antes do
-`mmap`, evitando page faults de inferência sobre o mount remoto. `/health` é
-liveness; `/ready` só responde 200 depois que o modelo terminou de carregar.
-
-## Controlador autônomo
-
-O servidor reconstrói o estado a partir do histórico enviado pelo OpenClaude:
-
-```text
-descobrir → ler → alterar → verificar → finalizar
-                           └→ falhou → diagnosticar → corrigir → verificar
-```
+The failed build forced a source compilation of `llama-cpp-python`. This package
+installs `llama-cpp-python==0.3.34` from the official CPU wheel index with
+`--only-binary=:all:`, so pip cannot fall back to a source build and the
+builder no longer needs a compiler toolchain. The version is exposed as the
+Docker build argument `LLAMA_CPP_VERSION`.
 
-Garantias relevantes:
+This CPU GGUF service also removes Torch, Transformers, Gradio, tokenizers and
+sentencepiece because they are not part of the inference path.
 
-- pedidos de implementação começam por inspeção real;
-- uma alteração exige verificação posterior;
-- `Edit`/`Write` que falhou nunca conta como modificação concluída;
-- pedidos explicitamente somente-leitura não disparam instalação ou escrita;
-- conteúdo retornado por `Read`/busca é dado não confiável, não instrução;
-- exemplos de tool call dentro de prosa ou dentro do conteúdo de outro tool call
-  não são executados;
-- `tool_choice=required` e escolha forçada são respeitados até em saudações;
-- após 18 resultados sem conclusão verificada, o loop termina com o bloqueio
-  concreto;
-- chamadas ao único contexto `Llama` são serializadas.
-
-## Configurar OpenClaude 0.27
-
-O Space público não exige token HF. `cpu-local` é apenas o valor não secreto que
-o OpenClaude exige para um endpoint OpenAI remoto. Nunca use um token `hf_...`
-como chave desta API.
+## OpenClaude
 
 ```bash
+cat << 'EOF' > abrir_claude
+#!/usr/bin/env bash
 export CLAUDE_CODE_USE_OPENAI=1
 export OPENAI_BASE_URL="https://erinaldorodrigues-vscode.hf.space/v1"
-export OPENAI_API_KEY="cpu-local"
+export OPENAI_API_KEY="local"
 export OPENAI_MODEL="qwen-coder"
-export OPENAI_API_FORMAT="chat_completions"
-
-export CLAUDE_CODE_OPENAI_CONTEXT_WINDOWS='{"qwen-coder":32768}'
-export CLAUDE_CODE_OPENAI_FALLBACK_CONTEXT_WINDOW="32768"
-export CLAUDE_CODE_OPENAI_MAX_OUTPUT_TOKENS='{"qwen-coder":2048}'
+export API_TIMEOUT_MS="600000"
+npx openclaude
+EOF
 
-export WEB_SEARCH_PROVIDER="custom"
-export WEB_SEARCH_API="https://erinaldorodrigues-vscode.hf.space/web-search"
-export WEB_METHOD="GET"
-export WEB_QUERY_PARAM="q"
-export WEB_JSON_PATH="results"
-export WEB_SEARCH_TIMEOUT_SEC="120"
-
-# Variável de runtime: não coloque em --provider-env-file no OpenClaude 0.27.
-export API_TIMEOUT_MS="1800000"
-
-exec openclaude --provider openai --model qwen-coder
+chmod +x abrir_claude
+./abrir_claude
 ```
 
-O launcher local configurado nesta máquina é:
+If you create the Hugging Face Secret `API_KEY`, set `OPENAI_API_KEY` to the
+same value. If `API_KEY` is empty, authentication is disabled.
 
-```text
-/home/miau/.local/bin/openclaude-vscode
-```
-
-Ele usa `/home/miau/.config/openclaude/hf-vscode.env`, preserva os perfis globais
-existentes e exporta `API_TIMEOUT_MS` separadamente porque o allowlist de
-`--provider-env-file` rejeita essa variável.
-
-Teste rápido:
+## Tool calling
 
-```bash
-printf '%s\n' 'olá' | \
-  /home/miau/.local/bin/openclaude-vscode \
-  --bare --no-session-persistence --tools '' --print --output-format json
-```
+The selected Qwen3 GGUF contains native `<tools>`, `<tool_call>` and
+`<tool_response>` support.
 
-## Autenticação opcional
+The compatibility layer:
+- passes OpenAI tool schemas to the native Qwen template;
+- parses native Qwen tool blocks and raw tool JSON;
+- returns real OpenAI `message.tool_calls`;
+- never reports required tool JSON as a successful plain-text action;
+- accepts real `role="tool"` responses from OpenClaude;
+- changes repeated `required` to `auto` immediately after a tool result, so
+  the agent can finish instead of being forced into a tool loop;
+- supports multiple independent calls when `parallel_tool_calls` allows them.
 
-Por padrão os endpoints são públicos. Para impedir uso externo, crie um segredo
-aleatório próprio e salve-o como Secret `API_TOKEN` nas configurações do Space.
-Quando definido, `/v1/*` e `/web-search` exigem `Authorization: Bearer ...`;
-`/health` permanece público. Use o mesmo valor em `OPENAI_API_KEY` no cliente.
+Tool turns requested with `stream=true` are validated fully first and then
+emitted as OpenAI SSE chunks. Normal chat without tools uses real token
+streaming from llama.cpp.
 
-Não reutilize um token de conta do Hugging Face como `API_TOKEN`.
+## Endpoints
 
-## Deploy Docker no Hugging Face
+- `GET /`
+- `GET /health`
+- `GET /ready`
+- `GET /v1/models`
+- `POST /v1/chat/completions`
+- `GET /docs`
 
-O metadata deste README usa `sdk: docker`, portanto o Dockerfile é o runtime real
-do Space. Ele:
+`/health` does not load the model. `/ready` returns 503 until the GGUF is
+actually loaded.
 
-1. constrói wheels em um estágio isolado;
-2. compila `llama-cpp-python` com OpenBLAS e LTO;
-3. instala somente bibliotecas de runtime na imagem final;
-4. executa como usuário sem privilégios;
-5. persiste o cache do modelo em `/data/huggingface` quando o Space possui storage.
+## Environment variables
 
-Não há Ollama, code-server, senha fixa ou instalador remoto via `curl | sh`.
+```text
+MODEL_REPO=unsloth/Qwen3-4B-Instruct-2507-GGUF
+MODEL_FILE=Qwen3-4B-Instruct-2507-Q4_K_M.gguf
+MODEL_ALIAS=qwen-coder
+MODEL_ALIASES=qwen3-4b,Qwen3-4B-Instruct-2507,unsloth/Qwen3-4B-Instruct-2507-GGUF
 
-## Testes
+N_CTX=8192
+MAX_NEW_TOKENS=2048
+N_THREADS=2
+N_THREADS_BATCH=2
+N_BATCH=128
+N_UBATCH=64
 
-Suíte leve, sem baixar o GGUF:
+PRELOAD_MODEL=false
+MODEL_RETRY_COOLDOWN_SECONDS=30
+MAX_REQUEST_BYTES=2000000
 
-```bash
-python -m pip install -r requirements-test.txt
-python -m pytest -q
+API_KEY=
+HF_TOKEN=
 ```
 
-Smoke test no Space publicado:
+If persistent Space storage is attached, you may set `HF_HOME` to a writable
+persistent path (for example `/data/huggingface`) to retain the GGUF cache.
+
+## Validation
 
 ```bash
-python smoke_api.py
+python -m compileall -q app.py settings.py tooling.py tests
+python -m unittest discover -s tests -v
 ```
 
-Para checar apenas saúde, catálogo e fast path:
+After deployment:
 
 ```bash
-python smoke_api.py --skip-generation
+bash smoke_test.sh
 ```
-
-Os testes cobrem normalização OpenAI/OpenClaude, streaming, uso, tool calls
-paralelas, allowlist, conteúdo não confiável, compaction de contexto e os fluxos
-autônomos de inspeção, implementação, reparo e verificação.
-
-## Limites reais
-
-O maior custo em CPU Basic é o prefill de prompts longos do OpenClaude. OpenBLAS,
-batch maior e compaction reduzem esse custo, mas não transformam 2 vCPU em GPU.
-Para ganho adicional sem reduzir a qualidade do modelo, a melhoria efetiva é
-migrar o Space para hardware com mais vCPU. Trocar o GGUF 4B pelo perfil `fast`
-é a opção gratuita de maior impacto, com perda mensurável de capacidade.

VALIDATION.txt

@@ -0,0 +1,34 @@
+Vscode Qwen3 CPU fixed package
+Generated: 2026-08-11
+
+VALIDATION RESULT
+
+Python compileall:
+PASS
+
+Unit tests:
+17 tests executed
+17 passed
+0 failed
+
+Validated contracts:
+- app.py / settings.py / tooling.py compile successfully.
+- Dockerfile does not force a source build of llama-cpp-python.
+- Dockerfile does not install a compiler/CMake/Ninja toolchain.
+- Dockerfile pins llama-cpp-python 0.3.34 through the official CPU wheel index.
+- requirements.txt contains no Torch, Transformers, Gradio or SentencePiece stack.
+- Qwen native <tool_call> parsing is covered.
+- Raw JSON tool-call fallback is covered.
+- Multiple tool calls are covered.
+- Undeclared tools are rejected.
+- Duplicate calls in one response are deduplicated.
+- tool_choice=required is preserved on the first action turn.
+- required is downgraded to auto after a real tool result to avoid forced loops.
+- simple greetings do not force Bash/tool execution.
+- named/forced tool selection is covered.
+- required API routes are present.
+
+Hardware-dependent validation:
+The 2.5 GB GGUF was intentionally not downloaded in the artifact-generation
+environment. The final model load/inference test must run after deploying the
+Docker Space on Hugging Face CPU hardware.

app.py

@@ -1,1080 +1,573 @@
-"""CPU/RAM OpenAI-compatible backend for OpenClaude using Qwen3 GGUF.
-
-Designed for Hugging Face Spaces CPU Basic (2 vCPU / 16 GB RAM):
-- no CUDA / ZeroGPU dependency
-- GGUF inference through llama-cpp-python
-- Qwen3 native tool-call chat template rendered by Transformers tokenizer
-- OpenAI-compatible /v1/chat/completions and SSE tool_call responses
-"""
-
 from __future__ import annotations
 
-import asyncio
-import hashlib
+import importlib.metadata
 import json
 import os
-import secrets
-import shutil
 import threading
 import time
 import traceback
 import uuid
-from typing import Any
+from typing import Any, Iterator
 
-os.environ.setdefault("HF_HUB_DISABLE_PROGRESS_BARS", "1")
-os.environ.setdefault("TOKENIZERS_PARALLELISM", "false")
-
-import gradio as gr
-from fastapi import HTTPException
+from fastapi import FastAPI, HTTPException, Request
 from fastapi.responses import JSONResponse, StreamingResponse
-from huggingface_hub import hf_hub_download
-from llama_cpp import Llama
-from pydantic import BaseModel, Field, ValidationError
+from pydantic import BaseModel
 from starlette.concurrency import run_in_threadpool
-from starlette.middleware.base import BaseHTTPMiddleware
-from starlette.requests import Request
-from transformers import AutoTokenizer
-
-from openai_compat import (
-    analyze_tool_flow,
-    compact_forced_tool_messages,
-    compact_terminal_messages,
+
+from settings import Settings
+from tooling import (
+    build_tool_plan,
+    extract_tool_calls,
     indexed_tool_calls,
+    inject_system_instruction,
     is_simple_greeting,
     normalize_tools,
-    resolve_tool_choice,
-    select_tools,
-    tool_choice_instruction,
     tool_names,
-    tool_protocol_instruction,
 )
-from openclaude_compat import (
-    TOOL_PROTOCOL_MARKER,
-    add_system_instruction,
-    has_tool_protocol,
-    normalize_openclaude_messages,
-)
-from tool_calls import extract_tool_calls, has_complete_tool_call, recover_forced_tool_call
-from web_search import SearchUnavailable, search_web
-
-
-# CPU model profiles. ``smart`` is the default because Qwen3-4B-Instruct-2507
-# is materially stronger at instruction following, coding and tool use while its
-# Q4_K_M GGUF (~2.5 GB) still fits comfortably in a 16 GB CPU Space. ``fast``
-# retains the previous 1.7B model for users who prefer latency over capability.
-MODEL_PROFILE = os.getenv("MODEL_PROFILE", "smart").strip().casefold()
-_MODEL_PROFILES = {
-    "smart": {
-        "repo": "unsloth/Qwen3-4B-Instruct-2507-GGUF",
-        "revision": "a06e946bb6b655725eafa393f4a9745d460374c9",
-        "filename": "Qwen3-4B-Instruct-2507-Q4_K_M.gguf",
-        "tokenizer": "Qwen/Qwen3-4B-Instruct-2507",
-        "tokenizer_revision": "cdbee75f17c01a7cc42f958dc650907174af0554",
-        "display": "Qwen3-4B-Instruct-2507-GGUF-Q4_K_M",
-        "alias": "qwen3-4b-instruct-2507",
-    },
-    "fast": {
-        "repo": "unsloth/Qwen3-1.7B-GGUF",
-        "revision": "d7f544eead698dbd1f15126ef60b45a1e1933222",
-        "filename": "Qwen3-1.7B-Q4_K_M.gguf",
-        "tokenizer": "Qwen/Qwen3-1.7B",
-        "tokenizer_revision": "70d244cc86ccca08cf5af4e1e306ecf908b1ad5e",
-        "display": "Qwen3-1.7B-GGUF-Q4_K_M",
-        "alias": "qwen3-1.7b",
-    },
-}
-if MODEL_PROFILE not in _MODEL_PROFILES:
-    raise RuntimeError(
-        f"MODEL_PROFILE must be one of {sorted(_MODEL_PROFILES)}; got {MODEL_PROFILE!r}"
-    )
-_PROFILE = _MODEL_PROFILES[MODEL_PROFILE]
-GGUF_REPO = os.getenv("GGUF_REPO", _PROFILE["repo"])
-GGUF_FILENAME = os.getenv("GGUF_FILENAME", _PROFILE["filename"])
-TOKENIZER_MODEL = os.getenv("TOKENIZER_MODEL", _PROFILE["tokenizer"])
-GGUF_REVISION = os.getenv(
-    "GGUF_REVISION",
-    _PROFILE["revision"] if GGUF_REPO == _PROFILE["repo"] else "",
-).strip()
-TOKENIZER_REVISION = os.getenv(
-    "TOKENIZER_REVISION",
-    _PROFILE["tokenizer_revision"]
-    if TOKENIZER_MODEL == _PROFILE["tokenizer"]
-    else "",
-).strip()
-MODEL = os.getenv("MODEL", os.getenv("MODEL_ID", "qwen-coder"))
-MODEL_DISPLAY_NAME = os.getenv("MODEL_DISPLAY_NAME", _PROFILE["display"])
-
-
-def _env_bool(name: str, default: bool) -> bool:
-    raw = os.getenv(name)
-    if raw is None:
-        return default
-    value = raw.strip().casefold()
-    if value in {"1", "true", "yes", "on"}:
-        return True
-    if value in {"0", "false", "no", "off"}:
-        return False
-    raise RuntimeError(f"{name} must be a boolean; got {raw!r}")
-
-# 32K is an operational limit that is realistic on 16 GB RAM. Larger contexts
-# are intentionally not advertised on 2-vCPU CPU Basic because KV cache and
-# prompt latency become impractical even when the model supports more tokens.
-MAX_SUPPORTED_CONTEXT_TOKENS = 32768
-MAX_CONTEXT_TOKENS = int(os.getenv("MAX_CONTEXT_TOKENS", "32768"))
-if not 1024 <= MAX_CONTEXT_TOKENS <= MAX_SUPPORTED_CONTEXT_TOKENS:
-    raise RuntimeError(
-        f"MAX_CONTEXT_TOKENS must be between 1024 and {MAX_SUPPORTED_CONTEXT_TOKENS}; "
-        f"got {MAX_CONTEXT_TOKENS}"
-    )
 
-MAX_NEW_TOKENS = int(os.getenv("MAX_NEW_TOKENS", "2048"))
-MAX_TOOL_CALL_TOKENS = int(os.getenv("MAX_TOOL_CALL_TOKENS", "768"))
-MAX_COMPACT_TOOL_TOKENS = int(os.getenv("MAX_COMPACT_TOOL_TOKENS", "384"))
-MAX_TERMINAL_SUMMARY_TOKENS = int(os.getenv("MAX_TERMINAL_SUMMARY_TOKENS", "384"))
-DEFAULT_TEMPERATURE = float(os.getenv("DEFAULT_TEMPERATURE", "0.0"))
-MAX_TEMPERATURE = float(os.getenv("MAX_TEMPERATURE", "0.7"))
-TOOL_TEMPERATURE = 0.0
-CPU_THREADS = max(1, int(os.getenv("CPU_THREADS", str(min(2, os.cpu_count() or 2)))))
-CPU_BATCH_THREADS = max(1, int(os.getenv("CPU_BATCH_THREADS", str(CPU_THREADS))))
-N_BATCH = max(64, int(os.getenv("N_BATCH", "1024")))
-N_UBATCH = min(N_BATCH, max(32, int(os.getenv("N_UBATCH", "512"))))
-FLASH_ATTN = _env_bool("FLASH_ATTN", True)
-KV_CACHE_TYPE = os.getenv("KV_CACHE_TYPE", "q8_0").strip().casefold()
-_KV_CACHE_TYPES = {"f16": 1, "q8_0": 8}
-if KV_CACHE_TYPE not in _KV_CACHE_TYPES:
-    raise RuntimeError(
-        f"KV_CACHE_TYPE must be one of {sorted(_KV_CACHE_TYPES)}; got {KV_CACHE_TYPE!r}"
-    )
-KV_CACHE_TYPE_ID = _KV_CACHE_TYPES[KV_CACHE_TYPE]
-if not FLASH_ATTN and KV_CACHE_TYPE != "f16":
-    raise RuntimeError(
-        "KV_CACHE_TYPE must be f16 when FLASH_ATTN is disabled; llama.cpp "
-        "requires flash attention for a quantized V cache"
-    )
-MAX_GENERATION_QUEUE = max(1, int(os.getenv("MAX_GENERATION_QUEUE", "4")))
-GENERATION_QUEUE_TIMEOUT_SECONDS = max(
-    0.0, float(os.getenv("GENERATION_QUEUE_TIMEOUT_SECONDS", "2"))
-)
-SSE_HEARTBEAT_SECONDS = max(1.0, float(os.getenv("SSE_HEARTBEAT_SECONDS", "10")))
-MAX_REQUEST_BYTES = max(1024, int(os.getenv("MAX_REQUEST_BYTES", "4000000")))
-API_TOKEN = os.getenv("API_TOKEN", "").strip()
-PRELOAD_MODEL = _env_bool("PRELOAD_MODEL", True)
-COPY_MODEL_TO_LOCAL = _env_bool("COPY_MODEL_TO_LOCAL", True)
-MODEL_RUNTIME_DIR = os.getenv("MODEL_RUNTIME_DIR", "/tmp/qwen-gguf").strip()
-MODEL_ALIASES = tuple(
-    dict.fromkeys(
-        (
-            MODEL,
-            "qwen-coder",
-            _PROFILE["alias"],
-            MODEL_DISPLAY_NAME,
-            GGUF_REPO,
-        )
-    )
-)
+SETTINGS = Settings.from_env()
 
-# Tokenizer is small and used only for Qwen's canonical chat template and token
-# accounting. It does not load Transformers model weights or PyTorch.
-_tokenizer_kwargs = {"revision": TOKENIZER_REVISION} if TOKENIZER_REVISION else {}
-tokenizer = AutoTokenizer.from_pretrained(TOKENIZER_MODEL, **_tokenizer_kwargs)
+app = FastAPI(
+    title="Qwen3 CPU OpenAI API",
+    version="4.0.0",
+    docs_url="/docs",
+    redoc_url=None,
+)
 
-_model: Llama | None = None
+_model: Any = None
+_model_state = "cold"
+_model_error: str | None = None
+_model_last_error_at = 0.0
 _model_path: str | None = None
-_MODEL_LOAD_LOCK = threading.Lock()
-_GENERATION_LOCK = threading.Lock()
-_GENERATION_SLOTS = threading.BoundedSemaphore(MAX_GENERATION_QUEUE)
-
+_model_load_lock = threading.Lock()
+_inference_lock = threading.Lock()
 
-class GenerationBusyError(RuntimeError):
-    """Raised when the bounded CPU inference queue is already full."""
 
-
-def _local_model_path(cached_path: str) -> str:
-    """Copy a bucket-cached GGUF to local disk before mmap-based inference."""
-    if not COPY_MODEL_TO_LOCAL or not MODEL_RUNTIME_DIR:
-        return cached_path
+def _version(distribution: str) -> str:
     try:
-        source_size = os.path.getsize(cached_path)
-    except OSError:
-        # Tests and custom hub implementations may supply a virtual path.
-        return cached_path
-
-    fingerprint = hashlib.sha256(
-        f"{GGUF_REPO}\0{GGUF_REVISION}\0{GGUF_FILENAME}".encode()
-    ).hexdigest()[:16]
-    destination = os.path.join(
-        MODEL_RUNTIME_DIR,
-        f"{fingerprint}-{os.path.basename(GGUF_FILENAME)}",
-    )
-    try:
-        os.makedirs(MODEL_RUNTIME_DIR, exist_ok=True)
-        if os.path.getsize(destination) == source_size:
-            return destination
-    except FileNotFoundError:
-        pass
-    except OSError:
-        return cached_path
-
-    temporary = f"{destination}.{os.getpid()}.tmp"
-    try:
-        print(f"Copying cached GGUF to local runtime disk: {destination}", flush=True)
-        shutil.copyfile(cached_path, temporary)
-        if os.path.getsize(temporary) != source_size:
-            raise OSError("local GGUF copy has an unexpected size")
-        os.replace(temporary, destination)
-        return destination
-    except OSError:
-        traceback.print_exc()
-        try:
-            os.unlink(temporary)
-        except FileNotFoundError:
-            pass
-        return cached_path
+        return importlib.metadata.version(distribution)
+    except importlib.metadata.PackageNotFoundError:
+        return "missing"
 
 
-def _ensure_model_loaded() -> Llama:
-    """Download and load the GGUF exactly once per Space process."""
-    global _model, _model_path
-    if _model is not None:
-        return _model
+def _short_error(error: BaseException) -> str:
+    message = f"{type(error).__name__}: {error}".replace("\n", " ").strip()
+    return message[:500]
 
-    with _MODEL_LOAD_LOCK:
-        if _model is not None:
-            return _model
 
-        print(
-            f"Loading CPU model {GGUF_REPO}/{GGUF_FILENAME} "
-            f"ctx={MAX_CONTEXT_TOKENS} threads={CPU_THREADS}",
-            flush=True,
-        )
-        download_kwargs = {"revision": GGUF_REVISION} if GGUF_REVISION else {}
-        cached_path = hf_hub_download(
-            repo_id=GGUF_REPO,
-            filename=GGUF_FILENAME,
-            **download_kwargs,
-        )
-        _model_path = _local_model_path(cached_path)
-        candidate = Llama(
-            model_path=_model_path,
-            n_ctx=MAX_CONTEXT_TOKENS,
-            n_threads=CPU_THREADS,
-            n_threads_batch=CPU_BATCH_THREADS,
-            n_batch=N_BATCH,
-            n_ubatch=N_UBATCH,
-            n_gpu_layers=0,
-            use_mmap=True,
-            use_mlock=False,
-            flash_attn=FLASH_ATTN,
-            type_k=KV_CACHE_TYPE_ID,
-            type_v=KV_CACHE_TYPE_ID,
-            no_perf=True,
-            verbose=False,
-        )
-        _model = candidate
-        print(f"CPU model ready: {_model_path}", flush=True)
-        return candidate
+def _model_loaded() -> bool:
+    return _model is not None and _model_state == "ready"
 
 
-def _bounded_output_tokens(value: float | int) -> int:
-    try:
-        requested = int(value)
-    except (TypeError, ValueError):
-        requested = MAX_NEW_TOKENS
-    return max(1, min(requested, MAX_NEW_TOKENS))
+def _ensure_model_loaded() -> Any:
+    global _model, _model_state, _model_error
+    global _model_last_error_at, _model_path
 
+    if _model_loaded():
+        return _model
 
-def _native_tools(raw_tools: object) -> list[dict[str, Any]]:
-    return normalize_tools(raw_tools)
+    now = time.monotonic()
+    if (
+        _model is None
+        and _model_state == "error"
+        and SETTINGS.model_retry_cooldown_seconds > 0
+        and now - _model_last_error_at < SETTINGS.model_retry_cooldown_seconds
+    ):
+        raise RuntimeError(
+            "Model load is in cooldown after the previous failure: "
+            + (_model_error or "unknown error")
+        )
 
+    with _model_load_lock:
+        if _model_loaded():
+            return _model
 
-def _render_prompt(messages: list[dict[str, Any]], tools: list[dict[str, Any]]) -> str:
-    """Render Qwen3's native tool protocol while disabling long think traces."""
-    template_kwargs: dict[str, Any] = {
-        "tokenize": False,
-        "add_generation_prompt": True,
-        "enable_thinking": False,
-    }
-    if tools:
-        template_kwargs["tools"] = tools
-    try:
-        return tokenizer.apply_chat_template(messages, **template_kwargs)
-    except TypeError:
-        # Some tokenizer revisions may not expose enable_thinking as a kwarg.
-        template_kwargs.pop("enable_thinking", None)
-        return tokenizer.apply_chat_template(messages, **template_kwargs)
-    except Exception as template_error:
-        if tools:
+        now = time.monotonic()
+        if (
+            _model is None
+            and _model_state == "error"
+            and SETTINGS.model_retry_cooldown_seconds > 0
+            and now - _model_last_error_at < SETTINGS.model_retry_cooldown_seconds
+        ):
             raise RuntimeError(
-                "Qwen chat template failed while tools were enabled; refusing "
-                "to continue with a tool-less prompt"
-            ) from template_error
-        raise
-
-
-def _render_prompt_with_ids(
-    messages: list[dict[str, Any]], tools: list[dict[str, Any]]
-) -> tuple[str, list[int]]:
-    prompt = _render_prompt(messages, tools)
-    ids = tokenizer(prompt, add_special_tokens=False, truncation=False)["input_ids"]
-    return prompt, ids
-
-
-def _trim_oldest_turn(messages: list[dict[str, Any]]) -> list[dict[str, Any]] | None:
-    user_indexes = [
-        index
-        for index, message in enumerate(messages)
-        if str(message.get("role", "")).casefold() == "user"
-    ]
-    if len(user_indexes) >= 2:
-        cutoff = user_indexes[1]
-        return [
-            message
-            for index, message in enumerate(messages)
-            if index >= cutoff or str(message.get("role", "")).casefold() == "system"
-        ]
-    if user_indexes and user_indexes[0] > 0:
-        cutoff = user_indexes[0]
-        trimmed = [
-            message
-            for index, message in enumerate(messages)
-            if index >= cutoff or str(message.get("role", "")).casefold() == "system"
-        ]
-        return trimmed if trimmed != messages else None
-    return None
-
-
-CONTEXT_TRUNCATION_MARKER = "\n...[older/oversized content truncated to fit context]...\n"
-
-
-def _truncate_text_to_tokens(text: str, target_tokens: int) -> str:
-    ids = tokenizer(text, add_special_tokens=False, truncation=False)["input_ids"]
-    target = max(1, int(target_tokens))
-    if len(ids) <= target:
-        return text
-    marker_ids = tokenizer(
-        CONTEXT_TRUNCATION_MARKER,
-        add_special_tokens=False,
-        truncation=False,
-    )["input_ids"]
-    payload_budget = max(1, target - len(marker_ids))
-    head = max(1, payload_budget // 2)
-    tail = max(0, payload_budget - head)
-    head_text = tokenizer.decode(ids[:head], skip_special_tokens=False)
-    tail_text = tokenizer.decode(ids[-tail:], skip_special_tokens=False) if tail else ""
-    return head_text + CONTEXT_TRUNCATION_MARKER + tail_text
-
-
-def _fit_messages_to_context(
-    messages: list[dict[str, Any]],
-    tools: list[dict[str, Any]],
-    output_tokens: int,
-) -> list[dict[str, Any]]:
-    """Keep complete Qwen tool schemas intact while fitting the 32K CPU context."""
-    fitted, _, _ = _fit_messages_to_context_prepared(messages, tools, output_tokens)
-    return fitted
-
-
-def _fit_messages_to_context_prepared(
-    messages: list[dict[str, Any]],
-    tools: list[dict[str, Any]],
-    output_tokens: int,
-) -> tuple[list[dict[str, Any]], str, list[int]]:
-    """Fit messages and retain the final render so callers do not redo work."""
-    input_budget = max(1, MAX_CONTEXT_TOKENS - output_tokens)
-    fitted = [dict(message) for message in messages]
-    prompt, prompt_ids = _render_prompt_with_ids(fitted, tools)
-
-    while len(prompt_ids) > input_budget:
-        trimmed = _trim_oldest_turn(fitted)
-        if trimmed is None or trimmed == fitted:
-            break
-        fitted = trimmed
-        prompt, prompt_ids = _render_prompt_with_ids(fitted, tools)
-
-    latest_user_index = max(
-        (
-            index
-            for index, message in enumerate(fitted)
-            if str(message.get("role", "")).casefold() == "user"
-        ),
-        default=-1,
-    )
-
-    for _ in range(max(8, len(fitted) * 4)):
-        current_length = len(prompt_ids)
-        if current_length <= input_budget:
-            return fitted, prompt, prompt_ids
-        excess = current_length - input_budget
-        candidates: list[tuple[int, int, int]] = []
-        for index, message in enumerate(fitted):
-            content = message.get("content")
-            if not isinstance(content, str) or not content:
-                continue
-            if TOOL_PROTOCOL_MARKER in content:
-                continue
-            role = str(message.get("role", "")).casefold()
-            minimum = 768 if index == latest_user_index else (512 if role in {"system", "tool"} else 256)
-            token_length = len(
-                tokenizer(content, add_special_tokens=False, truncation=False)["input_ids"]
+                "Model load is in cooldown after the previous failure: "
+                + (_model_error or "unknown error")
             )
-            if token_length > minimum:
-                candidates.append((token_length, index, minimum))
-        if not candidates:
-            break
-        token_length, index, minimum = max(candidates)
-        target = max(minimum, token_length - excess - 64)
-        if target >= token_length:
-            target = max(minimum, token_length // 2)
-        original = str(fitted[index]["content"])
-        shortened = _truncate_text_to_tokens(original, target)
-        if shortened == original:
-            break
-        fitted[index] = {**fitted[index], "content": shortened}
-        prompt, prompt_ids = _render_prompt_with_ids(fitted, tools)
-
-    if len(prompt_ids) > input_budget:
-        raise ValueError(
-            "tool-enabled prompt exceeds the configured CPU context window even "
-            "after whole-turn and message-content compaction; refusing to slice "
-            "the Qwen tool schema"
-        )
-    return fitted, prompt, prompt_ids
-
 
-def _prepare_prompt(
-    messages: list[dict[str, Any]],
-    tools: list[dict[str, Any]],
-    output_tokens: int,
-) -> tuple[str, int]:
-    """Compact, render and tokenize once for generation and usage accounting."""
-    _, prompt, encoded = _fit_messages_to_context_prepared(
-        messages,
-        tools,
-        output_tokens,
-    )
-    input_budget = max(1, MAX_CONTEXT_TOKENS - output_tokens)
-    if len(encoded) > input_budget:
-        raise ValueError("prompt exceeds CPU context after safe compaction")
-    return prompt, len(encoded)
-
-
-def _completion_token_count(text: str) -> int:
-    return len(tokenizer(text, add_special_tokens=False, truncation=False)["input_ids"])
+        _model_state = "loading"
+        _model_error = None
+        started = time.monotonic()
+        print(
+            f"Loading {SETTINGS.model_repo}/{SETTINGS.model_file} on CPU "
+            f"(ctx={SETTINGS.n_ctx}, threads={SETTINGS.n_threads})...",
+            flush=True,
+        )
 
+        try:
+            from huggingface_hub import hf_hub_download
+            from llama_cpp import Llama
 
-def gerar(
-    messages_json: str,
-    temperature: float,
-    max_new_tokens: float,
-    tools_json: str = "[]",
-    *,
-    prepared_prompt: str | None = None,
-    prepared_prompt_tokens: int | None = None,
-) -> str:
-    """Generate entirely on CPU/RAM with llama.cpp."""
-    try:
-        tools = _native_tools(json.loads(tools_json))
-    except (TypeError, ValueError, json.JSONDecodeError):
-        tools = []
-
-    output_tokens = _bounded_output_tokens(max_new_tokens)
-    if prepared_prompt is None:
-        messages = json.loads(messages_json)
-        if not isinstance(messages, list):
-            raise ValueError("messages_json must contain a JSON list")
-        prompt, prompt_token_count = _prepare_prompt(messages, tools, output_tokens)
-    else:
-        prompt = prepared_prompt
-        if prepared_prompt_tokens is None:
-            prompt_token_count = len(
-                tokenizer(prompt, add_special_tokens=False, truncation=False)["input_ids"]
+            downloaded = hf_hub_download(
+                repo_id=SETTINGS.model_repo,
+                filename=SETTINGS.model_file,
+                token=os.getenv("HF_TOKEN") or None,
             )
-        else:
-            prompt_token_count = max(1, int(prepared_prompt_tokens))
-        if prompt_token_count > max(1, MAX_CONTEXT_TOKENS - output_tokens):
-            raise ValueError("prompt exceeds CPU context after safe compaction")
+            size = os.path.getsize(downloaded)
+            if size < SETTINGS.model_min_bytes:
+                raise RuntimeError(
+                    f"Downloaded GGUF is unexpectedly small: {size} bytes"
+                )
 
-    temp = max(0.0, min(float(temperature), MAX_TEMPERATURE))
-    kwargs: dict[str, Any] = {
-        "max_tokens": output_tokens,
-        "temperature": temp,
-        "stop": ["<|im_end|>", "<|endoftext|>"],
-        "echo": False,
-    }
-    if temp > 0:
-        kwargs.update({"top_p": 0.8, "top_k": 20, "repeat_penalty": 1.05})
-
-    # A single Llama instance/context is not safe for overlapping completion
-    # calls.  Every entry point (OpenAI middleware, Gradio, tests, future
-    # helpers) must serialize at this lowest level.  Keeping the lock here is
-    # intentional: a route-level lock can be bypassed by another caller and
-    # corrupt llama.cpp's shared KV/evaluation state, which may abort the whole
-    # process inside GGML rather than raise a Python exception.
-    queued_at = time.monotonic()
-    admitted = _GENERATION_SLOTS.acquire(timeout=GENERATION_QUEUE_TIMEOUT_SECONDS)
-    if not admitted:
-        raise GenerationBusyError(
-            "CPU inference queue is full; retry after the current requests finish"
-        )
-    try:
-        with _GENERATION_LOCK:
-            llm = _ensure_model_loaded()
-            queue_wait = time.monotonic() - queued_at
-            print(
-                f"CPU generation: prompt_tokens={prompt_token_count} "
-                f"max_new_tokens={output_tokens} threads={CPU_THREADS} "
-                f"tools={len(tools)} queue_wait={queue_wait:.2f}s",
-                flush=True,
+            model = Llama(
+                model_path=downloaded,
+                n_ctx=SETTINGS.n_ctx,
+                n_batch=min(SETTINGS.n_batch, SETTINGS.n_ctx),
+                n_ubatch=min(SETTINGS.n_ubatch, SETTINGS.n_batch),
+                n_threads=SETTINGS.n_threads,
+                n_threads_batch=SETTINGS.n_threads_batch,
+                n_gpu_layers=0,
+                use_mmap=True,
+                use_mlock=False,
+                verbose=False,
             )
-            started = time.monotonic()
-            response = llm(prompt, **kwargs)
-            text = str(response["choices"][0].get("text") or "").strip()
-            elapsed = time.monotonic() - started
+
+            _model = model
+            _model_path = downloaded
+            _model_state = "ready"
             print(
-                f"CPU generation completed: output_tokens={_completion_token_count(text)} "
-                f"elapsed={elapsed:.2f}s",
+                f"Model ready on CPU in {time.monotonic() - started:.1f}s; "
+                f"file={downloaded}",
                 flush=True,
             )
-            return text
-    finally:
-        _GENERATION_SLOTS.release()
+            return model
+        except Exception as error:
+            _model = None
+            _model_state = "error"
+            _model_error = _short_error(error)
+            _model_last_error_at = time.monotonic()
+            traceback.print_exc()
+            raise
 
 
 class ChatCompletionRequest(BaseModel):
-    model: str = MODEL
-    messages: list[dict[str, Any]] = Field(min_length=1)
-    temperature: float = Field(default=DEFAULT_TEMPERATURE, ge=0.0)
-    max_tokens: int | None = Field(default=None, ge=1)
-    max_completion_tokens: int | None = Field(default=None, ge=1)
+    model: str = SETTINGS.model_alias
+    messages: list[dict[str, Any]]
+    temperature: float = 0.0
+    top_p: float = 0.95
+    max_tokens: int | None = None
+    max_completion_tokens: int | None = None
     stream: bool = False
     tools: list[dict[str, Any]] | None = None
     tool_choice: Any = None
     parallel_tool_calls: bool | None = None
-    stream_options: dict[str, Any] | None = None
+    stop: str | list[str] | None = None
+    seed: int | None = None
+    presence_penalty: float = 0.0
+    frequency_penalty: float = 0.0
+    response_format: dict[str, Any] | None = None
+    n: int = 1
 
 
-def _simple_greeting_payload(request: ChatCompletionRequest) -> dict[str, Any] | None:
-    requested_mode = (
-        request.tool_choice.casefold()
-        if isinstance(request.tool_choice, str)
-        else None
+def _validate_request(request: ChatCompletionRequest) -> None:
+    if request.model not in SETTINGS.model_aliases:
+        raise HTTPException(
+            status_code=404, detail=f"Model not available: {request.model}"
+        )
+    if not request.messages:
+        raise HTTPException(status_code=400, detail="messages must not be empty")
+    if request.n != 1:
+        raise HTTPException(status_code=400, detail="Only n=1 is supported")
+
+
+def _bounded_max_tokens(request: ChatCompletionRequest) -> int:
+    raw = (
+        request.max_completion_tokens
+        if request.max_completion_tokens is not None
+        else request.max_tokens
     )
-    # The deterministic greeting is valid only when the caller left tool use
-    # optional (or explicitly disabled it).  ``required`` and forced-function
-    # choices are wire-level constraints and must reach the normal tool router.
-    if request.tool_choice is not None and requested_mode not in {"auto", "none"}:
+    if raw is None:
+        raw = SETTINGS.max_new_tokens
+    try:
+        value = int(raw)
+    except (TypeError, ValueError) as exc:
+        raise HTTPException(status_code=400, detail="Invalid max_tokens") from exc
+    return max(1, min(value, SETTINGS.max_new_tokens))
+
+
+def _llama_kwargs(
+    request: ChatCompletionRequest,
+    messages: list[dict[str, Any]],
+    tools: list[dict[str, Any]],
+    *,
+    stream: bool,
+) -> dict[str, Any]:
+    temperature = max(0.0, min(float(request.temperature), 2.0))
+    if tools:
+        temperature = 0.0
+
+    kwargs: dict[str, Any] = {
+        "messages": messages,
+        "temperature": temperature,
+        "top_p": max(0.01, min(float(request.top_p), 1.0)),
+        "max_tokens": _bounded_max_tokens(request),
+        "stream": stream,
+        "model": SETTINGS.model_alias,
+        "presence_penalty": max(
+            -2.0, min(float(request.presence_penalty), 2.0)
+        ),
+        "frequency_penalty": max(
+            -2.0, min(float(request.frequency_penalty), 2.0)
+        ),
+    }
+    if request.stop is not None:
+        kwargs["stop"] = request.stop
+    if request.seed is not None:
+        kwargs["seed"] = int(request.seed)
+    if request.response_format is not None and not tools:
+        kwargs["response_format"] = request.response_format
+    if tools:
+        # Qwen3's GGUF embeds the tool Jinja template.  The compatibility
+        # layer below validates/parses the resulting native tool blocks.
+        kwargs["tools"] = tools
+        kwargs["tool_choice"] = "auto"
+    return kwargs
+
+
+def _fast_greeting(
+    request: ChatCompletionRequest, tool_mode: str
+) -> dict[str, Any] | None:
+    if tool_mode not in {"none", "auto"}:
         return None
     if not is_simple_greeting(request.messages):
         return None
-    text = "Olá! Como posso ajudar você hoje?"
-    prompt_tokens = _completion_token_count(json.dumps(request.messages, ensure_ascii=False))
-    completion_tokens = _completion_token_count(text)
     return {
-        "id": f"chatcmpl-{uuid.uuid4().hex}",
+        "id": "chatcmpl-" + uuid.uuid4().hex,
         "object": "chat.completion",
         "created": int(time.time()),
-        "model": MODEL,
+        "model": SETTINGS.model_alias,
         "choices": [
             {
                 "index": 0,
-                "message": {"role": "assistant", "content": text},
+                "message": {
+                    "role": "assistant",
+                    "content": "Olá! Como posso ajudar você hoje?",
+                },
                 "finish_reason": "stop",
+                "logprobs": None,
             }
         ],
         "usage": {
-            "prompt_tokens": prompt_tokens,
-            "completion_tokens": completion_tokens,
-            "total_tokens": prompt_tokens + completion_tokens,
+            "prompt_tokens": 0,
+            "completion_tokens": 0,
+            "total_tokens": 0,
         },
     }
 
 
 def _completion_payload(request: ChatCompletionRequest) -> dict[str, Any]:
-    if request.model not in MODEL_ALIASES:
-        raise HTTPException(status_code=404, detail=f"Model not available: {request.model}")
-
-    greeting = _simple_greeting_payload(request)
-    if greeting is not None:
-        return greeting
-
-    already_adapted = has_tool_protocol(request.messages)
-    flow_state = analyze_tool_flow(request.messages, request.tools or [])
-    requested_mode = request.tool_choice.casefold() if isinstance(request.tool_choice, str) else None
-    state_controls_choice = request.tool_choice is None or requested_mode in {"auto", "required"}
-    effective_choice = resolve_tool_choice(request.tool_choice, flow_state)
+    _validate_request(request)
+    tools = normalize_tools(request.tools or [])
+
     try:
-        effective_tools, tool_mode = select_tools(request.tools or [], effective_choice)
+        plan = build_tool_plan(
+            request.messages,
+            tools,
+            request.tool_choice,
+            request.parallel_tool_calls,
+        )
     except ValueError as error:
         raise HTTPException(status_code=400, detail=str(error)) from error
 
-    allow_parallel = request.parallel_tool_calls is True
-    instructions = [
-        instruction
-        for instruction in (
-            (
-                tool_protocol_instruction(
-                    effective_tools,
-                    parallel_tool_calls=allow_parallel,
-                )
-                if effective_tools and not has_tool_protocol(request.messages)
-                else None
-            ),
-            tool_choice_instruction(tool_mode, effective_tools),
-            (
-                flow_state.instruction
-                if (
-                    state_controls_choice
-                    and not already_adapted
-                    and not (
-                        requested_mode == "required"
-                        and flow_state.can_finalize
-                        and not flow_state.requires_tool
-                    )
-                )
-                else None
-            ),
-        )
-        if instruction
-    ]
-    instruction = "\n\n".join(instructions) if instructions else None
-
-    if request.max_completion_tokens is not None:
-        max_tokens = request.max_completion_tokens
-    elif request.max_tokens is not None:
-        max_tokens = request.max_tokens
-    else:
-        max_tokens = MAX_NEW_TOKENS
-    if effective_tools:
-        max_tokens = min(max_tokens, MAX_TOOL_CALL_TOKENS)
+    fast = _fast_greeting(request, plan.mode)
+    if fast is not None:
+        return fast
 
-    temperature = min(max(float(request.temperature), 0.0), MAX_TEMPERATURE)
-    if tool_mode in {"required", "forced"}:
-        temperature = TOOL_TEMPERATURE
+    messages = inject_system_instruction(request.messages, plan.instruction)
+    model = _ensure_model_loaded()
+    kwargs = _llama_kwargs(
+        request, messages, plan.tools, stream=False
+    )
 
     try:
-        if flow_state.terminal and flow_state.can_finalize and not flow_state.requires_tool:
-            # The tool work is already proven complete. Re-sending OpenClaude's
-            # multi-thousand-token system prompt and tool manuals on 2 vCPU makes
-            # a trivial final summary take minutes and can hit the client hard
-            # runtime. Use an evidence-only finalization prompt instead.
-            prompt_messages = compact_terminal_messages(request.messages)
-            max_tokens = min(max_tokens, MAX_TERMINAL_SUMMARY_TOKENS)
-        elif (
-            flow_state.compact_prompt
-            and flow_state.requires_tool
-            and flow_state.forced_tool
-            and len(effective_tools) == 1
-        ):
-            # The router already chose the exact function. Avoid re-prefilling
-            # OpenClaude's full 7-11K-token manual merely to produce arguments
-            # for one deterministic tool call. The selected tool schema remains
-            # in Qwen's native template.
-            compact_instruction = "\n\n".join(
-                part
-                for part in (
-                    flow_state.instruction,
-                    tool_choice_instruction(tool_mode, effective_tools),
-                )
-                if part
-            )
-            prompt_messages = compact_forced_tool_messages(
-                request.messages,
-                compact_instruction or instruction,
-            )
-            max_tokens = min(max_tokens, MAX_COMPACT_TOOL_TOKENS)
-        else:
-            normalized_messages = (
-                [dict(message) for message in request.messages]
-                if already_adapted
-                else normalize_openclaude_messages(request.messages)
-            )
-            prompt_messages = add_system_instruction(normalized_messages, instruction)
+        with _inference_lock:
+            raw = model.create_chat_completion(**kwargs)
     except ValueError as error:
-        raise HTTPException(status_code=400, detail=str(error)) from error
+        message = str(error)
+        status = 413 if "context" in message.casefold() else 400
+        raise HTTPException(status_code=status, detail=message) from error
 
-    bounded_max_tokens = _bounded_output_tokens(max_tokens)
-    try:
-        prepared_prompt, prompt_tokens = _prepare_prompt(
-            prompt_messages,
-            effective_tools,
-            bounded_max_tokens,
-        )
-    except ValueError as error:
-        raise HTTPException(status_code=413, detail=str(error)) from error
-
-    print(
-        "Tool routing: "
-        f"requested={request.tool_choice!r} mode={tool_mode} "
-        f"state_required={flow_state.requires_tool} terminal={flow_state.terminal} "
-        f"phase={flow_state.phase} steps={flow_state.step_count} "
-        f"compact={flow_state.compact_prompt} forced={flow_state.forced_tool!r} "
-        f"effective_tools={[tool['function']['name'] for tool in effective_tools]}",
-        flush=True,
-    )
+    if not isinstance(raw, dict):
+        raise RuntimeError("llama-cpp-python returned an invalid response")
 
-    try:
-        text = gerar(
-            json.dumps(prompt_messages, ensure_ascii=False),
-            temperature,
-            bounded_max_tokens,
-            json.dumps(effective_tools, ensure_ascii=False),
-            prepared_prompt=prepared_prompt,
-            prepared_prompt_tokens=prompt_tokens,
-        )
-    except GenerationBusyError as error:
-        raise HTTPException(status_code=429, detail=str(error)) from error
-    completion_tokens = _completion_token_count(text)
-
-    if effective_tools:
-        tool_calls, content = extract_tool_calls(text, tool_names(effective_tools))
-        if not tool_calls and tool_mode in {"forced", "required"} and len(effective_tools) == 1:
-            recovered = recover_forced_tool_call(text, effective_tools[0]["function"]["name"])
-            if recovered is not None:
-                tool_calls, content = [recovered], ""
-        if not allow_parallel:
-            tool_calls = tool_calls[:1]
-    else:
-        tool_calls, content = [], text
+    choices = raw.get("choices")
+    if not isinstance(choices, list) or not choices:
+        raise RuntimeError("llama-cpp-python returned no choices")
 
-    if effective_tools and not tool_calls and has_complete_tool_call(text):
-        raise HTTPException(
-            status_code=502,
-            detail=(
-                "Model produced a complete but invalid or unadvertised tool call; "
-                "refusing to expose it as plain text to the tool executor."
-            ),
-        )
+    choice = choices[0]
+    message = choice.get("message")
+    if not isinstance(message, dict):
+        message = {"role": "assistant", "content": ""}
 
-    message: dict[str, Any] = {"role": "assistant", "content": content or None}
-    if tool_mode in {"required", "forced"} and not tool_calls:
-        detail = (
-            "CPU model failed to produce a valid required tool call. "
-            "No plain-text success response was returned because OpenClaude requested tool execution."
-        )
-        if completion_tokens >= bounded_max_tokens:
-            detail += " Generation reached the output-token limit."
-        raise HTTPException(status_code=502, detail=detail)
+    content = message.get("content")
+    content_text = content if isinstance(content, str) else ""
 
-    finish_reason = "stop"
-    if tool_calls:
-        message["tool_calls"] = tool_calls
+    calls = extract_tool_calls(
+        content_text,
+        tool_names(plan.tools),
+        message.get("tool_calls"),
+    )
+    if request.parallel_tool_calls is False:
+        calls = calls[:1]
+
+    if calls:
+        output_message: dict[str, Any] = {
+            "role": "assistant",
+            "content": None,
+            "tool_calls": calls,
+        }
         finish_reason = "tool_calls"
-    elif completion_tokens >= bounded_max_tokens:
-        finish_reason = "length"
+    else:
+        if plan.mode in {"required", "forced"}:
+            raise HTTPException(
+                status_code=502,
+                detail=(
+                    "Model failed to emit a structured tool call while "
+                    f"tool_choice was {plan.mode}."
+                ),
+            )
+        output_message = {
+            "role": "assistant",
+            "content": content_text,
+        }
+        finish_reason = choice.get("finish_reason") or "stop"
+
+    usage = raw.get("usage")
+    if not isinstance(usage, dict):
+        usage = {
+            "prompt_tokens": 0,
+            "completion_tokens": 0,
+            "total_tokens": 0,
+        }
 
     return {
-        "id": f"chatcmpl-{uuid.uuid4().hex}",
+        "id": raw.get("id") or ("chatcmpl-" + uuid.uuid4().hex),
         "object": "chat.completion",
-        "created": int(time.time()),
-        "model": MODEL,
-        "choices": [{"index": 0, "message": message, "finish_reason": finish_reason}],
-        "usage": {
-            "prompt_tokens": prompt_tokens,
-            "completion_tokens": completion_tokens,
-            "total_tokens": prompt_tokens + completion_tokens,
-        },
-    }
-
-
-def health() -> dict[str, Any]:
-    return {
-        "status": "ok",
-        "runtime": "cpu-llama.cpp",
-        "model": MODEL,
-        "model_display_name": MODEL_DISPLAY_NAME,
-        "model_profile": MODEL_PROFILE,
-        "agent_controller": "autonomous-inspect-act-verify",
-        "generation_serialized": True,
-        "gguf_repo": GGUF_REPO,
-        "gguf_revision": GGUF_REVISION or None,
-        "gguf_filename": GGUF_FILENAME,
-        "model_loaded": _model is not None,
-        "context_length": MAX_CONTEXT_TOKENS,
-        "cpu_threads": CPU_THREADS,
-        "cpu_batch_threads": CPU_BATCH_THREADS,
-        "n_batch": N_BATCH,
-        "n_ubatch": N_UBATCH,
-        "flash_attention": FLASH_ATTN,
-        "kv_cache_type": KV_CACHE_TYPE,
-        "max_generation_queue": MAX_GENERATION_QUEUE,
-        "preload_model": PRELOAD_MODEL,
-        "copy_model_to_local": COPY_MODEL_TO_LOCAL,
-        "zero_gpu": False,
-    }
-
-
-def readiness() -> tuple[dict[str, Any], int]:
-    if _model is None:
-        return {"status": "loading", "model_loaded": False}, 503
-    return {"status": "ready", "model_loaded": True}, 200
-
-
-def models() -> dict[str, Any]:
-    return {
-        "object": "list",
-        "data": [
+        "created": int(raw.get("created") or time.time()),
+        "model": SETTINGS.model_alias,
+        "choices": [
             {
-                "id": model_id,
-                "object": "model",
-                "owned_by": "Erinaldorodrigues",
-                "context_length": MAX_CONTEXT_TOKENS,
-                "max_input_tokens": MAX_CONTEXT_TOKENS,
-                "max_output_tokens": MAX_NEW_TOKENS,
-                "runtime": "cpu-llama.cpp",
-                "model_profile": MODEL_PROFILE,
-                "agent_controller": "autonomous-inspect-act-verify",
+                "index": 0,
+                "message": output_message,
+                "finish_reason": finish_reason,
+                "logprobs": choice.get("logprobs"),
             }
-            for model_id in MODEL_ALIASES
         ],
+        "usage": usage,
     }
 
 
-def _completion_sse_events(
-    request: ChatCompletionRequest,
-):
-    """Start inference in a worker and keep the SSE connection visibly alive."""
-
-    async def events():
-        # Tool calls must be parsed and validated before they can be exposed, so
-        # generation remains buffered. SSE comments prevent proxies and clients
-        # from treating a long CPU prefill as a dead connection.
-        yield ": stream-open\n\n"
-        completion_task = asyncio.create_task(
-            run_in_threadpool(_completion_payload, request)
-        )
-        while True:
-            try:
-                completion = await asyncio.wait_for(
-                    asyncio.shield(completion_task),
-                    timeout=SSE_HEARTBEAT_SECONDS,
-                )
-                break
-            except asyncio.TimeoutError:
-                yield ": keep-alive\n\n"
-            except HTTPException as error:
-                payload = {
-                    "error": {
-                        "message": str(error.detail),
-                        "type": "server_error",
-                        "code": error.status_code,
-                    }
-                }
-                yield f"data: {json.dumps(payload)}\n\n"
-                yield "data: [DONE]\n\n"
-                return
-            except Exception:
-                traceback.print_exc()
-                payload = {
-                    "error": {
-                        "message": "internal CPU Space error",
-                        "type": "server_error",
-                        "code": 500,
-                    }
-                }
-                yield f"data: {json.dumps(payload)}\n\n"
-                yield "data: [DONE]\n\n"
-                return
+def _payload_sse(payload: dict[str, Any]) -> Iterator[str]:
+    choice = payload["choices"][0]
+    chunk_id = payload["id"]
+    created = payload["created"]
+    model = payload["model"]
 
-        choice = completion["choices"][0]
-        chunk_id = completion["id"]
-        first = {
+    def event(delta: dict[str, Any], finish_reason: str | None) -> str:
+        body = {
             "id": chunk_id,
             "object": "chat.completion.chunk",
-            "created": completion["created"],
-            "model": MODEL,
+            "created": created,
+            "model": model,
             "choices": [
-                {"index": 0, "delta": {"role": "assistant"}, "finish_reason": None}
-            ],
-            "usage": None,
-        }
-        yield f"data: {json.dumps(first)}\n\n"
-        delta: dict[str, Any] = {}
-        if choice["message"].get("content"):
-            delta["content"] = choice["message"]["content"]
-        if choice["message"].get("tool_calls"):
-            delta["tool_calls"] = indexed_tool_calls(choice["message"]["tool_calls"])
-        body = {**first, "choices": [{"index": 0, "delta": delta, "finish_reason": None}]}
-        yield f"data: {json.dumps(body)}\n\n"
-        final = {
-            **first,
-            "choices": [
-                {"index": 0, "delta": {}, "finish_reason": choice["finish_reason"]}
+                {
+                    "index": 0,
+                    "delta": delta,
+                    "finish_reason": finish_reason,
+                    "logprobs": None,
+                }
             ],
         }
-        yield f"data: {json.dumps(final)}\n\n"
-        if request.stream_options and request.stream_options.get("include_usage") is True:
-            usage_chunk = {
-                "id": chunk_id,
-                "object": "chat.completion.chunk",
-                "created": completion["created"],
-                "model": MODEL,
-                "choices": [],
-                "usage": completion["usage"],
-            }
-            yield f"data: {json.dumps(usage_chunk)}\n\n"
-        yield "data: [DONE]\n\n"
-
-    return events()
-
+        return "data: " + json.dumps(body, ensure_ascii=False) + "\n\n"
 
-def chat_completions(request: ChatCompletionRequest):
-    if request.stream:
-        return StreamingResponse(
-            _completion_sse_events(request),
-            media_type="text/event-stream",
-            headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
+    yield event({"role": "assistant", "content": None}, None)
+    message = choice["message"]
+    if message.get("tool_calls"):
+        yield event(
+            {"tool_calls": indexed_tool_calls(message["tool_calls"])}, None
         )
+    elif isinstance(message.get("content"), str) and message["content"]:
+        yield event({"content": message["content"]}, None)
+    yield event({}, choice["finish_reason"])
+    yield "data: [DONE]\n\n"
+
+
+def _plain_stream_events(request: ChatCompletionRequest) -> Iterator[str]:
+    model = _ensure_model_loaded()
+    kwargs = _llama_kwargs(
+        request,
+        [dict(message) for message in request.messages],
+        [],
+        stream=True,
+    )
 
-    completion = _completion_payload(request)
-    return JSONResponse(content=completion)
-
+    with _inference_lock:
+        chunks = model.create_chat_completion(**kwargs)
+        for chunk in chunks:
+            if not isinstance(chunk, dict):
+                continue
+            chunk["model"] = SETTINGS.model_alias
+            yield "data: " + json.dumps(
+                chunk, ensure_ascii=False
+            ) + "\n\n"
+    yield "data: [DONE]\n\n"
 
-async def _chat_completions_in_thread(parsed_request: ChatCompletionRequest):
-    if parsed_request.stream:
-        return chat_completions(parsed_request)
-    return await run_in_threadpool(chat_completions, parsed_request)
 
+@app.middleware("http")
+async def request_guard(request: Request, call_next):
+    content_length = request.headers.get("content-length")
+    if content_length:
+        try:
+            if int(content_length) > SETTINGS.max_request_bytes:
+                return JSONResponse(
+                    status_code=413,
+                    content={"error": {"message": "Request body too large"}},
+                )
+        except ValueError:
+            pass
 
-def ui_status() -> str:
-    """Lightweight Space UI that never invokes the shared Llama context."""
-    return json.dumps(health(), ensure_ascii=False, indent=2)
+    if SETTINGS.api_key and request.url.path.startswith("/v1/"):
+        if request.headers.get("authorization") != "Bearer " + SETTINGS.api_key:
+            return JSONResponse(
+                status_code=401,
+                content={"error": {"message": "Invalid API key"}},
+                headers={"WWW-Authenticate": "Bearer"},
+            )
+    return await call_next(request)
 
 
-demo = gr.Interface(
-    fn=ui_status,
-    inputs=[],
-    outputs="text",
-    title="Qwen3 GGUF CPU/RAM OpenAI-compatible Backend",
-    description=(
-        "CPU-only llama.cpp backend for OpenClaude. Inference is intentionally "
-        "served only through /v1/chat/completions so every model call passes "
-        "through the single-context serialization guard."
-    ),
-)
+@app.get("/")
+async def root():
+    return {
+        "service": "Qwen3 CPU OpenAI API",
+        "status": "running",
+        "model": SETTINGS.model_alias,
+        "model_repo": SETTINGS.model_repo,
+        "model_state": _model_state,
+        "endpoints": [
+            "/health",
+            "/ready",
+            "/v1/models",
+            "/v1/chat/completions",
+        ],
+    }
 
 
-class OpenAIRouteMiddleware(BaseHTTPMiddleware):
-    async def dispatch(self, request: Request, call_next):
-        supplied_request_id = request.headers.get("x-request-id", "").strip()
-        request_id = (
-            supplied_request_id
-            if supplied_request_id and len(supplied_request_id) <= 128
-            else uuid.uuid4().hex
-        )
-        response = await self._dispatch_routes(request, call_next)
-        response.headers["X-Request-ID"] = request_id
-        return response
-
-    async def _dispatch_routes(self, request: Request, call_next):
-        path = request.url.path.rstrip("/") or "/"
-        if path == "/health" and request.method == "GET":
-            return JSONResponse(health())
-        if path == "/ready" and request.method == "GET":
-            payload, status_code = readiness()
-            return JSONResponse(payload, status_code=status_code)
-        if API_TOKEN and (path.startswith("/v1/") or path == "/web-search"):
-            authorization = request.headers.get("authorization", "")
-            scheme, _, supplied_token = authorization.partition(" ")
-            authorized = (
-                scheme.casefold() == "bearer"
-                and bool(supplied_token)
-                and secrets.compare_digest(supplied_token, API_TOKEN)
-            )
-            if not authorized:
-                return JSONResponse(
-                    status_code=401,
-                    content={"error": {"message": "unauthorized"}},
-                    headers={"WWW-Authenticate": "Bearer"},
-                )
-        if path == "/web-search" and request.method == "GET":
-            query = request.query_params.get("q", "").strip()
-            if not query or len(query) > 500:
-                return JSONResponse(status_code=400, content={"error": "invalid query"})
-            try:
-                return JSONResponse(await run_in_threadpool(search_web, query))
-            except SearchUnavailable as error:
-                return JSONResponse(
-                    status_code=503,
-                    content={"error": {"message": str(error) or "search unavailable"}},
-                )
-            except Exception:
-                traceback.print_exc()
-                return JSONResponse(
-                    status_code=500,
-                    content={"error": {"message": "internal web-search error"}},
-                )
-        if path == "/v1/models" and request.method == "GET":
-            return JSONResponse(models())
-        if path == "/v1/chat/completions" and request.method == "POST":
-            try:
-                content_length = request.headers.get("content-length")
-                if content_length is not None and int(content_length) > MAX_REQUEST_BYTES:
-                    return JSONResponse(
-                        status_code=413,
-                        content={"error": {"message": "request body too large"}},
-                    )
-                raw_body = await request.body()
-                if len(raw_body) > MAX_REQUEST_BYTES:
-                    return JSONResponse(
-                        status_code=413,
-                        content={"error": {"message": "request body too large"}},
-                    )
-                raw_request = json.loads(raw_body)
-                parsed_request = ChatCompletionRequest(**raw_request)
-            except (json.JSONDecodeError, UnicodeDecodeError, ValidationError, TypeError, ValueError) as error:
-                return JSONResponse(
-                    status_code=400, content={"error": {"message": str(error)}}
-                )
-            try:
-                return await _chat_completions_in_thread(parsed_request)
-            except HTTPException as error:
-                return JSONResponse(
-                    status_code=error.status_code,
-                    content={"error": {"message": error.detail}},
-                )
-            except Exception:
-                traceback.print_exc()
-                return JSONResponse(
-                    status_code=500,
-                    content={"error": {"message": "internal CPU Space error"}},
-                )
-        return await call_next(request)
+@app.get("/health")
+async def health():
+    return {
+        "status": "ok",
+        "model": SETTINGS.model_alias,
+        "model_state": _model_state,
+        "model_loaded": _model_loaded(),
+        "model_error": _model_error,
+        "n_ctx": SETTINGS.n_ctx,
+        "threads": SETTINGS.n_threads,
+        "llama_cpp_python": _version("llama-cpp-python"),
+        "huggingface_hub": _version("huggingface-hub"),
+    }
 
 
-import gradio.routes as _groutes
+@app.get("/ready")
+async def ready():
+    if not _model_loaded():
+        return JSONResponse(
+            status_code=503,
+            content={
+                "status": "not_ready",
+                "model_state": _model_state,
+                "model_error": _model_error,
+            },
+        )
+    return {"status": "ready", "model": SETTINGS.model_alias}
 
-_original_create_app = _groutes.App.create_app
 
+@app.get("/v1/models")
+async def models():
+    return {
+        "object": "list",
+        "data": [
+            {
+                "id": model_id,
+                "object": "model",
+                "created": 0,
+                "owned_by": "Erinaldorodrigues",
+                "context_length": SETTINGS.n_ctx,
+            }
+            for model_id in SETTINGS.model_aliases
+        ],
+    }
 
-def _create_app_with_openai_routes(*args, **kwargs):
-    created = _original_create_app(*args, **kwargs)
-    created.add_middleware(OpenAIRouteMiddleware)
-    return created
 
+@app.post("/v1/chat/completions")
+async def chat_completions(request: ChatCompletionRequest):
+    _validate_request(request)
+    normalized_tools = normalize_tools(request.tools or [])
 
-_groutes.App.create_app = staticmethod(_create_app_with_openai_routes)
+    if normalized_tools:
+        try:
+            payload = await run_in_threadpool(_completion_payload, request)
+        except HTTPException:
+            raise
+        except Exception as error:
+            traceback.print_exc()
+            raise HTTPException(
+                status_code=503, detail=_short_error(error)
+            ) from error
+
+        if request.stream:
+            return StreamingResponse(
+                _payload_sse(payload),
+                media_type="text/event-stream",
+                headers={
+                    "Cache-Control": "no-cache",
+                    "X-Accel-Buffering": "no",
+                },
+            )
+        return JSONResponse(payload)
 
+    if request.stream:
+        try:
+            await run_in_threadpool(_ensure_model_loaded)
+        except Exception as error:
+            raise HTTPException(
+                status_code=503, detail=_short_error(error)
+            ) from error
+        return StreamingResponse(
+            _plain_stream_events(request),
+            media_type="text/event-stream",
+            headers={
+                "Cache-Control": "no-cache",
+                "X-Accel-Buffering": "no",
+            },
+        )
 
-def _preload_model() -> None:
     try:
-        _ensure_model_loaded()
-    except Exception:
-        print("Background model preload failed; readiness remains unavailable.", flush=True)
+        payload = await run_in_threadpool(_completion_payload, request)
+        return JSONResponse(payload)
+    except HTTPException:
+        raise
+    except Exception as error:
         traceback.print_exc()
+        raise HTTPException(
+            status_code=503, detail=_short_error(error)
+        ) from error
 
 
-def main() -> None:
-    if PRELOAD_MODEL:
-        threading.Thread(
-            target=_preload_model,
-            name="model-preload",
-            daemon=True,
-        ).start()
-    demo.queue(default_concurrency_limit=1, max_size=8).launch(
-        show_error=True,
-        ssr_mode=False,
-    )
-
-
-if __name__ == "__main__":
-    main()
+@app.on_event("startup")
+async def optional_preload():
+    if SETTINGS.preload_model:
+        try:
+            await run_in_threadpool(_ensure_model_loaded)
+        except Exception:
+            # Keep /health alive for diagnosis instead of crashing the Space.
+            traceback.print_exc()

requirements.txt

@@ -1,12 +1,4 @@
-# CPU-only OpenClaude backend. No torch, CUDA, GPTQModel, accelerate or spaces.
-# The Docker build compiles llama-cpp-python from its pinned PyPI source release.
-
-fastapi>=0.115,<1
-pydantic>=2.10,<3
-httpx>=0.27,<1
-gradio==6.22.0
-huggingface_hub>=0.34,<2
-transformers==5.14.1
-tokenizers>=0.21
-sentencepiece>=0.2
-llama-cpp-python==0.3.34
+fastapi==0.141.1
+uvicorn==0.52.1
+pydantic==2.13.4
+huggingface_hub==1.27.0

settings.py

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+
+
+def _env_bool(name: str, default: bool) -> bool:
+    raw = os.getenv(name)
+    if raw is None:
+        return default
+    return raw.strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _env_int(name: str, default: int, minimum: int, maximum: int) -> int:
+    raw = os.getenv(name, str(default)).strip()
+    try:
+        value = int(raw)
+    except ValueError as exc:
+        raise RuntimeError(f"{name} must be an integer") from exc
+    if not minimum <= value <= maximum:
+        raise RuntimeError(f"{name} must be between {minimum} and {maximum}")
+    return value
+
+
+def _aliases(primary: str, raw: str) -> tuple[str, ...]:
+    values = [primary]
+    for value in raw.split(","):
+        value = value.strip()
+        if value and value not in values:
+            values.append(value)
+    return tuple(values)
+
+
+@dataclass(frozen=True)
+class Settings:
+    model_repo: str
+    model_file: str
+    model_alias: str
+    model_aliases: tuple[str, ...]
+    n_ctx: int
+    max_new_tokens: int
+    n_threads: int
+    n_threads_batch: int
+    n_batch: int
+    n_ubatch: int
+    model_min_bytes: int
+    api_key: str
+    max_request_bytes: int
+    model_retry_cooldown_seconds: int
+    preload_model: bool
+
+    @classmethod
+    def from_env(cls) -> "Settings":
+        model_alias = os.getenv("MODEL_ALIAS", "qwen-coder").strip() or "qwen-coder"
+        aliases = _aliases(
+            model_alias,
+            os.getenv(
+                "MODEL_ALIASES",
+                "qwen3-4b,Qwen3-4B-Instruct-2507,"
+                "unsloth/Qwen3-4B-Instruct-2507-GGUF",
+            ),
+        )
+        cpu_count = os.cpu_count() or 2
+        default_threads = min(2, cpu_count)
+        return cls(
+            model_repo=os.getenv(
+                "MODEL_REPO", "unsloth/Qwen3-4B-Instruct-2507-GGUF"
+            ).strip(),
+            model_file=os.getenv(
+                "MODEL_FILE", "Qwen3-4B-Instruct-2507-Q4_K_M.gguf"
+            ).strip(),
+            model_alias=model_alias,
+            model_aliases=aliases,
+            n_ctx=_env_int("N_CTX", 8192, 1024, 32768),
+            max_new_tokens=_env_int("MAX_NEW_TOKENS", 2048, 1, 8192),
+            n_threads=_env_int("N_THREADS", default_threads, 1, 64),
+            n_threads_batch=_env_int(
+                "N_THREADS_BATCH", default_threads, 1, 64
+            ),
+            n_batch=_env_int("N_BATCH", 128, 16, 2048),
+            n_ubatch=_env_int("N_UBATCH", 64, 16, 2048),
+            model_min_bytes=_env_int(
+                "MODEL_MIN_BYTES",
+                2_000_000_000,
+                1_000_000,
+                20_000_000_000,
+            ),
+            api_key=os.getenv("API_KEY", "").strip(),
+            max_request_bytes=_env_int(
+                "MAX_REQUEST_BYTES", 2_000_000, 32_768, 20_000_000
+            ),
+            model_retry_cooldown_seconds=_env_int(
+                "MODEL_RETRY_COOLDOWN_SECONDS", 30, 0, 3600
+            ),
+            preload_model=_env_bool("PRELOAD_MODEL", False),
+        )

smoke_test.sh

@@ -0,0 +1,51 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+BASE_URL="${1:-https://erinaldorodrigues-vscode.hf.space}"
+API_KEY="${API_KEY:-local}"
+
+echo "== health =="
+curl -fsS "$BASE_URL/health"
+echo
+
+echo "== models =="
+curl -fsS -H "Authorization: Bearer $API_KEY" "$BASE_URL/v1/models"
+echo
+
+echo "== chat =="
+curl -fsS "$BASE_URL/v1/chat/completions" \
+  -H "Authorization: Bearer $API_KEY" \
+  -H "Content-Type: application/json" \
+  -d '{
+    "model":"qwen-coder",
+    "messages":[{"role":"user","content":"Responda apenas: OK"}],
+    "temperature":0,
+    "max_tokens":16
+  }'
+echo
+
+echo "== Bash tool call =="
+curl -fsS "$BASE_URL/v1/chat/completions" \
+  -H "Authorization: Bearer $API_KEY" \
+  -H "Content-Type: application/json" \
+  -d '{
+    "model":"qwen-coder",
+    "messages":[{"role":"user","content":"Use Bash para executar pwd. Não simule."}],
+    "temperature":0,
+    "max_tokens":256,
+    "tool_choice":"required",
+    "parallel_tool_calls":false,
+    "tools":[{
+      "type":"function",
+      "function":{
+        "name":"Bash",
+        "description":"Execute a shell command",
+        "parameters":{
+          "type":"object",
+          "properties":{"command":{"type":"string"}},
+          "required":["command"]
+        }
+      }
+    }]
+  }'
+echo

tests/test_settings.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+import os
+import unittest
+from unittest import mock
+
+from settings import Settings
+
+
+class SettingsTests(unittest.TestCase):
+    def test_defaults_are_cpu_safe(self):
+        with mock.patch.dict(os.environ, {}, clear=True):
+            settings = Settings.from_env()
+        self.assertEqual(settings.n_ctx, 8192)
+        self.assertEqual(settings.n_threads, min(2, os.cpu_count() or 2))
+        self.assertEqual(settings.model_alias, "qwen-coder")
+        self.assertTrue(settings.model_file.endswith("Q4_K_M.gguf"))
+
+    def test_invalid_context_is_rejected(self):
+        with mock.patch.dict(os.environ, {"N_CTX": "999999"}, clear=True):
+            with self.assertRaises(RuntimeError):
+                Settings.from_env()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_static_contract.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+import unittest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def executable_docker_lines() -> str:
+    lines = []
+    for line in (ROOT / "Dockerfile").read_text().splitlines():
+        stripped = line.strip()
+        if stripped and not stripped.startswith("#"):
+            lines.append(stripped.casefold())
+    return "\n".join(lines)
+
+
+class StaticContractTests(unittest.TestCase):
+    def test_app_compiles(self):
+        ast.parse((ROOT / "app.py").read_text(encoding="utf-8"))
+
+    def test_no_heavy_transformers_stack(self):
+        requirements = (ROOT / "requirements.txt").read_text().casefold()
+        for forbidden in ("torch", "transformers", "gradio", "sentencepiece"):
+            self.assertNotIn(forbidden, requirements)
+
+    def test_docker_never_source_builds_llama_cpp(self):
+        docker = executable_docker_lines()
+        self.assertNotIn("--no-binary", docker)
+        self.assertNotIn("build-essential", docker)
+        self.assertNotIn("cmake", docker)
+        self.assertNotIn("ninja", docker)
+        self.assertIn("--only-binary=:all:", docker)
+        self.assertIn(
+            "https://abetlen.github.io/llama-cpp-python/whl/cpu",
+            docker,
+        )
+        self.assertIn("llama-cpp-python==${llama_cpp_version}", docker)
+
+    def test_runtime_routes_exist(self):
+        source = (ROOT / "app.py").read_text()
+        for path in (
+            "/health",
+            "/ready",
+            "/v1/models",
+            "/v1/chat/completions",
+        ):
+            self.assertIn(path, source)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_tooling.py

@@ -0,0 +1,144 @@
+from __future__ import annotations
+
+import json
+import unittest
+
+from tooling import (
+    build_tool_plan,
+    extract_tool_calls,
+    has_recent_tool_result,
+    normalize_tools,
+)
+
+BASH = {
+    "type": "function",
+    "function": {
+        "name": "Bash",
+        "description": "run shell",
+        "parameters": {
+            "type": "object",
+            "properties": {"command": {"type": "string"}},
+            "required": ["command"],
+        },
+    },
+}
+
+READ = {
+    "type": "function",
+    "function": {
+        "name": "Read",
+        "description": "read file",
+        "parameters": {
+            "type": "object",
+            "properties": {"path": {"type": "string"}},
+            "required": ["path"],
+        },
+    },
+}
+
+
+class ToolingTests(unittest.TestCase):
+    def test_native_qwen_tag_is_structured(self):
+        text = '<tool_call>{"name":"Bash","arguments":{"command":"pwd"}}</tool_call>'
+        calls = extract_tool_calls(text, {"Bash"})
+        self.assertEqual(len(calls), 1)
+        self.assertEqual(calls[0]["function"]["name"], "Bash")
+        self.assertEqual(
+            json.loads(calls[0]["function"]["arguments"]),
+            {"command": "pwd"},
+        )
+
+    def test_raw_json_is_structured(self):
+        text = '{"name":"Bash","arguments":{"command":"sudo apt update"}}'
+        self.assertEqual(len(extract_tool_calls(text, {"Bash"})), 1)
+
+    def test_short_prefix_raw_json_is_structured(self):
+        text = (
+            'Vou executar agora.\n'
+            '{"name":"Bash","arguments":{"command":"sudo apt update"}}'
+        )
+        self.assertEqual(len(extract_tool_calls(text, {"Bash"})), 1)
+
+    def test_multiple_native_calls(self):
+        text = (
+            '<tool_call>{"name":"Read","arguments":{"path":"a"}}</tool_call>'
+            '<tool_call>{"name":"Read","arguments":{"path":"b"}}</tool_call>'
+        )
+        self.assertEqual(len(extract_tool_calls(text, {"Read"})), 2)
+
+    def test_undeclared_tool_is_rejected(self):
+        text = '<tool_call>{"name":"DeleteAll","arguments":{}}</tool_call>'
+        self.assertEqual(extract_tool_calls(text, {"Bash"}), [])
+
+    def test_duplicate_call_is_deduplicated(self):
+        text = (
+            '<tool_call>{"name":"Bash","arguments":{"command":"pwd"}}</tool_call>'
+            '<tool_call>{"name":"Bash","arguments":{"command":"pwd"}}</tool_call>'
+        )
+        self.assertEqual(len(extract_tool_calls(text, {"Bash"})), 1)
+
+    def test_required_first_turn_remains_required(self):
+        plan = build_tool_plan(
+            [{"role": "user", "content": "atualizar tudo sem perguntas"}],
+            [BASH],
+            "required",
+            False,
+        )
+        self.assertEqual(plan.mode, "required")
+        self.assertEqual(plan.tools[0]["function"]["name"], "Bash")
+
+    def test_required_after_tool_result_downgrades_auto(self):
+        messages = [
+            {"role": "user", "content": "execute pwd"},
+            {
+                "role": "assistant",
+                "content": None,
+                "tool_calls": [{
+                    "id": "call_1",
+                    "type": "function",
+                    "function": {
+                        "name": "Bash",
+                        "arguments": '{"command":"pwd"}',
+                    },
+                }],
+            },
+            {
+                "role": "tool",
+                "tool_call_id": "call_1",
+                "content": "/home/user\n",
+            },
+        ]
+        self.assertTrue(has_recent_tool_result(messages))
+        plan = build_tool_plan(messages, [BASH], "required", False)
+        self.assertEqual(plan.mode, "auto")
+
+    def test_required_greeting_does_not_force_tool(self):
+        plan = build_tool_plan(
+            [{"role": "user", "content": "oi"}],
+            [BASH],
+            "required",
+            False,
+        )
+        self.assertEqual(plan.mode, "none")
+        self.assertEqual(plan.tools, [])
+
+    def test_forced_tool_is_restricted(self):
+        plan = build_tool_plan(
+            [{"role": "user", "content": "leia o arquivo"}],
+            [BASH, READ],
+            {"type": "function", "function": {"name": "Read"}},
+            False,
+        )
+        self.assertEqual(plan.mode, "forced")
+        self.assertEqual(
+            [x["function"]["name"] for x in plan.tools], ["Read"]
+        )
+
+    def test_normalize_invalid_tools(self):
+        self.assertEqual(
+            len(normalize_tools([{}, {"type": "other"}, BASH])), 1
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tooling.py

@@ -0,0 +1,323 @@
+from __future__ import annotations
+
+import json
+import re
+import uuid
+from dataclasses import dataclass
+from typing import Any
+
+_TOOL_TAG_RE = re.compile(r"<tool_call>\s*(.*?)\s*</tool_call>", re.I | re.S)
+_GREETING_RE = re.compile(
+    r"^\s*(oi|ol[aá]|hello|hi|hey|bom dia|boa tarde|boa noite)[!.?,\s]*$",
+    re.I,
+)
+
+
+@dataclass(frozen=True)
+class ToolPlan:
+    tools: list[dict[str, Any]]
+    mode: str
+    instruction: str | None
+
+
+def normalize_tools(raw_tools: object) -> list[dict[str, Any]]:
+    if not isinstance(raw_tools, list):
+        return []
+    output: list[dict[str, Any]] = []
+    seen: set[str] = set()
+    for item in raw_tools:
+        if not isinstance(item, dict) or item.get("type") != "function":
+            continue
+        function = item.get("function")
+        if not isinstance(function, dict):
+            continue
+        name = function.get("name")
+        if not isinstance(name, str) or not name.strip():
+            continue
+        name = name.strip()
+        if name in seen:
+            continue
+        parameters = function.get("parameters")
+        if not isinstance(parameters, dict):
+            parameters = {"type": "object", "properties": {}}
+        description = function.get("description")
+        if not isinstance(description, str):
+            description = ""
+        if len(description) > 4000:
+            description = description[:3997] + "..."
+        output.append(
+            {
+                "type": "function",
+                "function": {
+                    "name": name,
+                    "description": description,
+                    "parameters": parameters,
+                },
+            }
+        )
+        seen.add(name)
+    return output
+
+
+def tool_names(tools: list[dict[str, Any]]) -> set[str]:
+    return {
+        tool["function"]["name"]
+        for tool in tools
+        if isinstance(tool, dict)
+        and isinstance(tool.get("function"), dict)
+        and isinstance(tool["function"].get("name"), str)
+    }
+
+
+def _content_text(message: dict[str, Any]) -> str:
+    content = message.get("content")
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        parts: list[str] = []
+        for part in content:
+            if isinstance(part, dict) and isinstance(part.get("text"), str):
+                parts.append(part["text"])
+        return "\n".join(parts)
+    return ""
+
+
+def last_user_text(messages: list[dict[str, Any]]) -> str:
+    for message in reversed(messages):
+        if isinstance(message, dict) and message.get("role") == "user":
+            return _content_text(message).strip()
+    return ""
+
+
+def is_simple_greeting(messages: list[dict[str, Any]]) -> bool:
+    text = last_user_text(messages)
+    return bool(text and _GREETING_RE.fullmatch(text))
+
+
+def has_recent_tool_result(messages: list[dict[str, Any]]) -> bool:
+    for message in reversed(messages):
+        if not isinstance(message, dict):
+            continue
+        role = message.get("role")
+        if role == "system":
+            continue
+        if role == "tool":
+            return True
+        if role == "user":
+            return "<tool_response>" in _content_text(message)
+        return False
+    return False
+
+
+def _forced_tool_name(requested: object) -> str | None:
+    if not isinstance(requested, dict) or requested.get("type") != "function":
+        return None
+    function = requested.get("function")
+    if not isinstance(function, dict):
+        return None
+    name = function.get("name")
+    return name.strip() if isinstance(name, str) and name.strip() else None
+
+
+def build_tool_plan(
+    messages: list[dict[str, Any]],
+    tools: list[dict[str, Any]],
+    requested_choice: object,
+    parallel_tool_calls: bool | None,
+) -> ToolPlan:
+    if not tools:
+        return ToolPlan([], "none", None)
+
+    names = tool_names(tools)
+    forced = _forced_tool_name(requested_choice)
+
+    if forced:
+        if forced not in names:
+            raise ValueError(f"Requested tool is not available: {forced}")
+        selected = [t for t in tools if t["function"]["name"] == forced]
+        mode = "forced"
+    elif isinstance(requested_choice, str):
+        choice = requested_choice.casefold()
+        if choice == "none":
+            return ToolPlan([], "none", None)
+        if choice == "required":
+            if is_simple_greeting(messages):
+                return ToolPlan([], "none", None)
+            # OpenClaude can keep "required" on the turn immediately after a
+            # real tool result. Auto lets Qwen synthesize or call another tool.
+            mode = "auto" if has_recent_tool_result(messages) else "required"
+            selected = tools
+        elif choice == "auto":
+            mode = "auto"
+            selected = tools
+        else:
+            raise ValueError(f"Unsupported tool_choice: {requested_choice}")
+    elif requested_choice is None:
+        mode = "auto"
+        selected = tools
+    else:
+        raise ValueError("Unsupported tool_choice")
+
+    lines = [
+        "Tool execution protocol:",
+        "- A tool is executed only when you emit the native tool-call format.",
+        "- Never print a tool JSON object as ordinary prose.",
+        "- Never claim a tool succeeded before a tool response is present.",
+        "- After a tool response, use the actual output; do not invent results.",
+        "- Do not repeat an identical successful call unless the returned output "
+        "shows that another execution is necessary.",
+    ]
+    if mode == "required":
+        lines.append(
+            "- For this turn you MUST call at least one provided tool before "
+            "giving a final answer."
+        )
+    elif mode == "forced":
+        lines.append(
+            f"- For this turn you MUST call the tool "
+            f"{selected[0]['function']['name']}."
+        )
+    if parallel_tool_calls is False:
+        lines.append("- Emit exactly one tool call in this turn.")
+    else:
+        lines.append("- Multiple independent tool calls are allowed when useful.")
+
+    return ToolPlan(selected, mode, "\n".join(lines))
+
+
+def inject_system_instruction(
+    messages: list[dict[str, Any]], instruction: str | None
+) -> list[dict[str, Any]]:
+    copied = [dict(message) for message in messages]
+    if not instruction:
+        return copied
+    for index, message in enumerate(copied):
+        if (
+            message.get("role") == "system"
+            and isinstance(message.get("content"), str)
+        ):
+            copied[index] = {
+                **message,
+                "content": message["content"].rstrip() + "\n\n" + instruction,
+            }
+            return copied
+    return [{"role": "system", "content": instruction}, *copied]
+
+
+def _json_sequence(blob: str) -> list[object]:
+    decoder = json.JSONDecoder()
+    output: list[object] = []
+    index = 0
+    while index < len(blob):
+        while index < len(blob) and (
+            blob[index].isspace() or blob[index] in ",;"
+        ):
+            index += 1
+        if index >= len(blob):
+            break
+        try:
+            value, end = decoder.raw_decode(blob, index)
+        except json.JSONDecodeError:
+            next_open = blob.find("{", index + 1)
+            if next_open < 0:
+                break
+            index = next_open
+            continue
+        output.append(value)
+        index = end
+    return output
+
+
+def _candidate_objects(text: str) -> list[object]:
+    tagged = _TOOL_TAG_RE.findall(text)
+    if tagged:
+        values: list[object] = []
+        for block in tagged:
+            values.extend(_json_sequence(block.strip()))
+        return values
+
+    stripped = text.strip()
+    if stripped.startswith("{"):
+        return _json_sequence(stripped)
+
+    first = stripped.find("{")
+    if 0 <= first <= 160:
+        prefix = stripped[:first]
+        rest = stripped[first:]
+        if (
+            '"name"' in rest[:300]
+            and '"arguments"' in rest[:500]
+            and len(prefix.split()) <= 25
+        ):
+            return _json_sequence(rest)
+    return []
+
+
+def _canonical_arguments(arguments: object) -> str:
+    if isinstance(arguments, str):
+        try:
+            parsed = json.loads(arguments)
+        except json.JSONDecodeError:
+            return json.dumps({"value": arguments}, ensure_ascii=False)
+        return json.dumps(parsed, ensure_ascii=False, separators=(",", ":"))
+    if arguments is None:
+        arguments = {}
+    return json.dumps(arguments, ensure_ascii=False, separators=(",", ":"))
+
+
+def _normalize_candidate(
+    candidate: object, allowed_names: set[str]
+) -> tuple[str, str] | None:
+    if not isinstance(candidate, dict):
+        return None
+    if isinstance(candidate.get("function"), dict):
+        function = candidate["function"]
+        name = function.get("name")
+        arguments = function.get("arguments", {})
+    else:
+        name = candidate.get("name")
+        arguments = candidate.get("arguments", {})
+    if not isinstance(name, str) or name not in allowed_names:
+        return None
+    return name, _canonical_arguments(arguments)
+
+
+def extract_tool_calls(
+    text: str,
+    allowed_names: set[str],
+    existing_tool_calls: object = None,
+) -> list[dict[str, Any]]:
+    normalized: list[tuple[str, str]] = []
+
+    if isinstance(existing_tool_calls, list):
+        for item in existing_tool_calls:
+            pair = _normalize_candidate(item, allowed_names)
+            if pair:
+                normalized.append(pair)
+
+    for candidate in _candidate_objects(text):
+        pair = _normalize_candidate(candidate, allowed_names)
+        if pair:
+            normalized.append(pair)
+
+    output: list[dict[str, Any]] = []
+    seen: set[tuple[str, str]] = set()
+    for name, arguments in normalized:
+        signature = (name, arguments)
+        if signature in seen:
+            continue
+        seen.add(signature)
+        output.append(
+            {
+                "id": "call_" + uuid.uuid4().hex,
+                "type": "function",
+                "function": {"name": name, "arguments": arguments},
+            }
+        )
+    return output
+
+
+def indexed_tool_calls(
+    tool_calls: list[dict[str, Any]]
+) -> list[dict[str, Any]]:
+    return [{"index": index, **call} for index, call in enumerate(tool_calls)]