Upload 4 files

README_final.md

@@ -0,0 +1,108 @@
+---
+title: Qwen2.5 Coder 32B AWQ OpenAI API
+emoji: 🚀
+colorFrom: green
+colorTo: yellow
+sdk: gradio
+sdk_version: 6.22.0
+python_version: '3.12'
+app_file: app.py
+pinned: false
+---
+
+# Qwen2.5-Coder-32B AWQ — OpenAI-compatible ZeroGPU API
+
+This Space serves `Qwen/Qwen2.5-Coder-32B-Instruct-AWQ` through an
+OpenAI-compatible Chat Completions endpoint on Hugging Face ZeroGPU.
+
+The runtime is pinned to the versions that were validated during the Space
+startup work:
+
+- PyTorch 2.11.0 / CUDA 13.0 wheels
+- torchvision 0.26.0
+- Transformers 5.14.1
+- GPTQModel 7.3.2
+- Gradio 6.22.0
+
+The AWQ model is intentionally loaded lazily from inside the `@spaces.GPU`
+function. This is required by this deployment because AWQ/Marlin performs real
+CUDA work while `from_pretrained()` is running. Do not move model loading back
+to module startup without retesting the Space on ZeroGPU.
+
+## API
+
+- `GET /health`
+- `GET /v1/models`
+- `POST /v1/chat/completions`
+- `GET /web-search?q=...`
+
+Accepted model names are the real model ID plus the compatibility aliases
+`qwen2.5-coder-32b` and `qwen-coder`. Aliases for unrelated Qwen3 or 14B
+weights are deliberately not accepted.
+
+Example client configuration:
+
+```text
+OPENAI_BASE_URL=https://erinaldorodrigues-qwen-coder-api.hf.space/v1
+OPENAI_API_BASE=https://erinaldorodrigues-qwen-coder-api.hf.space/v1
+OPENAI_MODEL=qwen2.5-coder-32b
+WEB_SEARCH_PROVIDER=custom
+WEB_SEARCH_API=https://erinaldorodrigues-qwen-coder-api.hf.space/web-search
+WEB_METHOD=GET
+WEB_QUERY_PARAM=q
+```
+
+If a client requires a non-empty `OPENAI_API_KEY`, it may send one, but the
+current `app.py` does not implement application-level Bearer-token validation.
+Add authentication before exposing private quota or sensitive tools.
+
+## Tool calling
+
+The backend accepts OpenAI-style `tools`, `tool_choice`, and
+`parallel_tool_calls`. Tool definitions are normalized for Qwen's native chat
+template and textual `<tool_call>...</tool_call>` outputs are translated back
+to OpenAI `message.tool_calls` objects.
+
+`tool_choice="required"` is reconciled with conversation state for OpenClaude:
+when the user explicitly identifies a tool, the choice is narrowed to that
+function; once usable tool evidence is already present, repeated `required`
+choices may be downgraded to `none` so the model can synthesize the answer
+instead of looping.
+
+When `parallel_tool_calls=true`, the prompt permits multiple independent tool
+calls. Otherwise generation stops after the first complete tool call.
+
+The Space generates tool calls; the calling client remains responsible for
+executing client-side tools and returning their results in subsequent `tool`
+messages. `/web-search` is a separate server-side search endpoint.
+
+## Generation and compatibility
+
+The default context limit is 16,384 tokens and default maximum output is 2,048
+tokens. Both can be changed with Space environment variables.
+
+`temperature=0` is preserved as greedy generation. Responses report prompt and
+completion token counts, and `finish_reason="length"` is returned when the
+configured output budget is exhausted without a completed tool call.
+
+`stream=true` returns OpenAI-style SSE framing. The current implementation
+finishes model generation before emitting the content delta, so it is protocol
+streaming rather than token-by-token low-latency streaming. This is intentional
+until a ZeroGPU-safe streamer/thread implementation is validated.
+
+## Health
+
+`/health` reports both the configured model and `model_loaded`. Because model
+loading is lazy, a healthy freshly started process can report
+`model_loaded=false` until the first GPU inference initializes the AWQ model.
+
+## Tests
+
+Run:
+
+```bash
+python -m unittest discover -p 'test_*.py'
+```
+
+The suite covers generation helpers, OpenAI/OpenClaude tool flow, tool-call
+parsing, web-search fallbacks, and static integration regressions in `app.py`.

app_final.py

@@ -0,0 +1,567 @@
+"""Reliable ZeroGPU backend for the local OpenAI-compatible proxy."""
+
+from __future__ import annotations
+
+import json
+import os
+import time
+import uuid
+import traceback
+from typing import Any
+
+os.environ.setdefault("HF_HUB_DISABLE_PROGRESS_BARS", "1")
+os.environ.setdefault("TRANSFORMERS_DISABLE_DEEPGEMM_LINEAR", "1")
+
+import gradio as gr
+import spaces
+import torch
+from fastapi import HTTPException
+from fastapi.responses import JSONResponse, StreamingResponse
+from pydantic import BaseModel, ValidationError
+from starlette.concurrency import run_in_threadpool
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.requests import Request
+from transformers import (
+    AutoModelForCausalLM,
+    AutoTokenizer,
+    StoppingCriteria,
+    StoppingCriteriaList,
+)
+from generation import (
+    gpu_duration_seconds,
+    head_tail_token_counts,
+    merge_eos_token_ids,
+)
+from openai_compat import (
+    analyze_tool_flow,
+    indexed_tool_calls,
+    normalize_tools,
+    resolve_tool_choice,
+    select_tools,
+    tool_choice_instruction,
+    tool_protocol_instruction,
+    tool_names,
+)
+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,
+)
+from web_search import SearchUnavailable, search_web
+
+
+# O Titã: Qwen2.5-Coder-32B nativamente quantizado em 4-bits (AWQ)
+MODEL = os.getenv(
+    "MODEL",
+    os.getenv("MODEL_ID", "Qwen/Qwen2.5-Coder-32B-Instruct-AWQ"),
+)
+
+MAX_CONTEXT_TOKENS = int(os.getenv("MAX_CONTEXT_TOKENS", "16384"))
+MAX_NEW_TOKENS = int(os.getenv("MAX_NEW_TOKENS", "2048"))
+MAX_TOOL_CALL_TOKENS = int(os.getenv("MAX_TOOL_CALL_TOKENS", "2048"))
+MAX_TEMPERATURE = float(os.getenv("MAX_TEMPERATURE", "0.2"))
+PRESERVED_PREFIX_TOKENS = int(os.getenv("PRESERVED_PREFIX_TOKENS", "4096"))
+
+tokenizer = AutoTokenizer.from_pretrained(MODEL)
+
+# O ZeroGPU só anexa uma GPU real dentro de funções decoradas com
+# @spaces.GPU; no escopo do módulo (startup) não existe CUDA de verdade,
+# apenas uma emulação que aceita `.to("cuda")`/`device_map="auto"` como
+# simples posicionamento de tensores. O carregamento deste modelo AWQ,
+# porém, dispara o kernel Marlin (`awq_marlin_repack`) de forma síncrona
+# dentro do próprio from_pretrained — isso é execução real de kernel CUDA,
+# não posicionamento, e por isso não existe backend CPU para ele (era
+# exatamente esse o erro do seu log). Por isso o carregamento precisa ser
+# adiado para dentro de `gerar`, a única função com GPU real anexada.
+model: AutoModelForCausalLM | None = None
+
+
+def _ensure_model_loaded() -> None:
+    """Carrega o modelo uma única vez, já dentro do contexto com GPU real."""
+    global model
+    if model is not None:
+        return
+    print(f"Loading {MODEL} on ZeroGPU (NATIVE AWQ)...", flush=True)
+    model = AutoModelForCausalLM.from_pretrained(
+        MODEL,
+        dtype="auto",
+        device_map="auto",
+        low_cpu_mem_usage=True,
+    )
+    model.eval()
+    print(f"Model ready on {next(model.parameters()).device}", flush=True)
+
+
+def _bounded_output_tokens(value: float) -> int:
+    try:
+        requested = int(value)
+    except (TypeError, ValueError):
+        requested = MAX_NEW_TOKENS
+    return max(1, min(requested, MAX_NEW_TOKENS))
+
+
+# Buffer para cobrir a compilação JIT do kernel Marlin + carregamento dos
+# pesos quando `gerar` cai num worker "frio" (sem o modelo em memória).
+# É uma estimativa (baseada nos ~99s de compilação que aparecem no seu log);
+# meça o cold start real do seu Space e ajuste. Confira também o teto de
+# duração por chamada da sua tier em
+# https://huggingface.co/docs/hub/spaces-zerogpu antes de subir esse valor —
+# se o teto for menor que isso, a chamada falha com "illegal duration".
+COLD_START_BUFFER_SECONDS = 180
+
+
+def _gpu_duration(
+    messages_json: str,
+    __: float,
+    max_new_tokens: float,
+    *tool_arguments: object,
+) -> int:
+    output_tokens = _bounded_output_tokens(max_new_tokens)
+    tool_characters = sum(
+        len(value) for value in tool_arguments if isinstance(value, str)
+    )
+    duration = gpu_duration_seconds(
+        len(messages_json) + tool_characters,
+        output_tokens,
+        MAX_CONTEXT_TOKENS,
+    )
+    return duration + COLD_START_BUFFER_SECONDS
+
+
+def _tool_protocol_active(messages: list[object]) -> bool:
+    return any(
+        isinstance(message, dict)
+        and isinstance(message.get("content"), str)
+        and TOOL_PROTOCOL_MARKER in message["content"]
+        for message in messages
+    )
+
+
+def _native_tools(raw_tools: object) -> list[dict[str, Any]]:
+    return normalize_tools(raw_tools)
+
+
+def _render_prompt(
+    messages: list[dict[str, Any]],
+    tools: list[dict[str, Any]],
+) -> str:
+    """Render the exact prompt used for generation, with a safe tool fallback."""
+    template_kwargs: dict[str, Any] = {
+        "tokenize": False,
+        "add_generation_prompt": True,
+    }
+    if tools:
+        template_kwargs["tools"] = tools
+
+    try:
+        return tokenizer.apply_chat_template(messages, **template_kwargs)
+    except Exception as template_error:
+        print(
+            f"Jinja Template Warning: {template_error}. Applying fallback.",
+            flush=True,
+        )
+        template_kwargs.pop("tools", None)
+        return tokenizer.apply_chat_template(messages, **template_kwargs)
+
+
+def _prompt_token_count(
+    messages: list[dict[str, Any]],
+    tools: list[dict[str, Any]],
+    output_tokens: int,
+) -> int:
+    """Count the prompt tokens that survive the same context bound as generation."""
+    prompt = _render_prompt(messages, tools)
+    encoded = tokenizer(
+        prompt,
+        add_special_tokens=False,
+        truncation=False,
+    )["input_ids"]
+    input_budget = max(1, MAX_CONTEXT_TOKENS - output_tokens)
+    return min(len(encoded), input_budget)
+
+
+def _completion_token_count(text: str) -> int:
+    """Count visible generated tokens for OpenAI-compatible usage reporting."""
+    return len(
+        tokenizer(
+            text,
+            add_special_tokens=False,
+            truncation=False,
+        )["input_ids"]
+    )
+
+
+class StopAfterToolCall(StoppingCriteria):
+    def __init__(self, prompt_length: int) -> None:
+        self.prompt_length = prompt_length
+
+    def __call__(self, input_ids, scores, **_: object):
+        completed = []
+        for sequence in input_ids:
+            generated = sequence[self.prompt_length :]
+            text = tokenizer.decode(generated, skip_special_tokens=False)
+            completed.append(has_complete_tool_call(text))
+        return torch.tensor(completed, dtype=torch.bool, device=input_ids.device)
+
+
+@spaces.GPU(duration=_gpu_duration)
+def gerar(
+    messages_json: str,
+    temperature: float,
+    max_new_tokens: float,
+    tools_json: str = "[]",
+    stop_after_first_tool: bool = True,
+) -> str:
+    _ensure_model_loaded()
+    messages = json.loads(messages_json)
+    if not isinstance(messages, list):
+        raise ValueError("messages_json must contain a JSON list")
+    try:
+        tools = _native_tools(json.loads(tools_json))
+    except (TypeError, ValueError, json.JSONDecodeError):
+        tools = []
+    if not isinstance(tools, list):
+        tools = []
+
+    output_tokens = _bounded_output_tokens(max_new_tokens)
+    tool_mode = _tool_protocol_active(messages) or bool(tools)
+    prompt = _render_prompt(messages, tools)
+
+    inputs = tokenizer(
+        prompt,
+        return_tensors="pt",
+        add_special_tokens=False,
+        truncation=False,
+    )
+    input_budget = max(1, MAX_CONTEXT_TOKENS - output_tokens)
+    input_length = inputs["input_ids"].shape[1]
+    
+    if input_length > input_budget:
+        head_tokens, tail_tokens = head_tail_token_counts(
+            input_length,
+            input_budget,
+            PRESERVED_PREFIX_TOKENS,
+        )
+        for key, value in inputs.items():
+            if (
+                isinstance(value, torch.Tensor)
+                and value.ndim == 2
+                and value.shape[1] == input_length
+            ):
+                parts = []
+                if head_tokens:
+                    parts.append(value[:, :head_tokens])
+                if tail_tokens:
+                    parts.append(value[:, -tail_tokens:])
+                inputs[key] = torch.cat(parts, dim=1)
+                
+    inputs = inputs.to("cuda")
+
+    print(
+        f"Generation started: input_tokens={inputs['input_ids'].shape[1]} "
+        f"max_new_tokens={output_tokens} tool_mode={tool_mode}",
+        flush=True,
+    )
+
+    eos_token_ids = merge_eos_token_ids(
+        model.generation_config.eos_token_id,
+        tokenizer.eos_token_id,
+    )
+    
+    generation_kwargs = {
+        "max_new_tokens": output_tokens,
+        "do_sample": float(temperature) > 0,
+        "pad_token_id": tokenizer.pad_token_id or tokenizer.eos_token_id,
+    }
+    if eos_token_ids is not None:
+        generation_kwargs["eos_token_id"] = eos_token_ids
+        
+    if generation_kwargs["do_sample"]:
+        generation_kwargs["temperature"] = max(0.01, float(temperature))
+        generation_kwargs["top_p"] = 0.8
+        generation_kwargs["top_k"] = 20
+        generation_kwargs["repetition_penalty"] = 1.05
+        
+    if tool_mode and stop_after_first_tool:
+        generation_kwargs["stopping_criteria"] = StoppingCriteriaList(
+            [StopAfterToolCall(inputs["input_ids"].shape[1])]
+        )
+
+    with torch.inference_mode():
+        output = model.generate(**inputs, **generation_kwargs)
+
+    generated = output[0][inputs["input_ids"].shape[1] :]
+    response = tokenizer.decode(generated, skip_special_tokens=True).strip()
+    print(f"Generation completed: output_tokens={generated.shape[0]}", flush=True)
+    return response
+
+
+class ChatCompletionRequest(BaseModel):
+    model: str = MODEL
+    messages: list[dict[str, Any]]
+    temperature: float = 0.2
+    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
+
+
+def _completion_payload(request: ChatCompletionRequest) -> dict[str, Any]:
+    if request.model not in {
+        MODEL,
+        "qwen-coder",
+        "qwen2.5-coder-32b",
+    }:
+        raise HTTPException(status_code=404, detail=f"Model not available: {request.model}")
+
+    already_adapted = has_tool_protocol(request.messages)
+    flow_state = analyze_tool_flow(request.messages, request.tools or [])
+    state_controls_choice = request.tool_choice is None or (
+        isinstance(request.tool_choice, str)
+        and request.tool_choice.casefold() in {"auto", "required"}
+    )
+    effective_choice = resolve_tool_choice(request.tool_choice, flow_state)
+    
+    try:
+        effective_tools, tool_mode = select_tools(
+            request.tools or [], effective_choice
+        )
+    except ValueError as error:
+        raise HTTPException(status_code=400, detail=str(error)) from error
+
+    instructions = [
+        instruction
+        for instruction in (
+            (
+                tool_protocol_instruction(
+                    effective_tools,
+                    parallel_tool_calls=request.parallel_tool_calls is True,
+                )
+                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
+                else None
+            ),
+        )
+        if instruction
+    ]
+    instruction = "\n\n".join(instructions) if instructions else None
+    max_tokens = request.max_completion_tokens or request.max_tokens or MAX_NEW_TOKENS
+    
+    if effective_tools:
+        max_tokens = min(max_tokens, MAX_TOOL_CALL_TOKENS)
+    temperature = min(max(float(request.temperature), 0.0), MAX_TEMPERATURE)
+    
+    try:
+        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,
+        )
+    except ValueError as error:
+        raise HTTPException(status_code=400, detail=str(error)) from error
+    
+    bounded_max_tokens = _bounded_output_tokens(max_tokens)
+    prompt_tokens = _prompt_token_count(
+        prompt_messages,
+        effective_tools,
+        bounded_max_tokens,
+    )
+    text = gerar(
+        json.dumps(prompt_messages),
+        temperature,
+        bounded_max_tokens,
+        json.dumps(effective_tools, ensure_ascii=False),
+        request.parallel_tool_calls is not True,
+    )
+    completion_tokens = _completion_token_count(text)
+    
+    if effective_tools:
+        tool_calls, content = extract_tool_calls(text, tool_names(effective_tools))
+        if request.parallel_tool_calls is False:
+            tool_calls = tool_calls[:1]
+    else:
+        tool_calls, content = [], text
+        
+    message: dict[str, Any] = {"role": "assistant", "content": content or None}
+    
+    finish_reason = "stop"
+    if tool_calls:
+        message["tool_calls"] = tool_calls
+        finish_reason = "tool_calls"
+    elif completion_tokens >= bounded_max_tokens:
+        finish_reason = "length"
+    elif effective_tools and has_complete_tool_call(text):
+        finish_reason = "stop"
+    elif tool_mode in {"required", "forced"}:
+        finish_reason = "stop"
+
+    return {
+        "id": f"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",
+        "model": MODEL,
+        "model_loaded": model is not None,
+    }
+
+
+def models() -> dict[str, Any]:
+    return {
+        "object": "list",
+        "data": [
+            {
+                "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,
+            }
+            for model_id in dict.fromkeys(("qwen2.5-coder-32b", MODEL))
+        ],
+    }
+
+
+def chat_completions(request: ChatCompletionRequest):
+    completion = _completion_payload(request)
+    if not request.stream:
+        return JSONResponse(content=completion)
+
+    choice = completion["choices"][0]
+    chunk_id = completion["id"]
+
+    def events():
+        first = {
+            "id": chunk_id,
+            "object": "chat.completion.chunk",
+            "created": completion["created"],
+            "model": MODEL,
+            "choices": [{"index": 0, "delta": {"role": "assistant"}, "finish_reason": 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"]}]}
+        yield f"data: {json.dumps(final)}\n\n"
+        yield "data: [DONE]\n\n"
+
+    return StreamingResponse(
+        events(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "X-Accel-Buffering": "no",
+        },
+    )
+
+
+demo = gr.Interface(
+    fn=gerar,
+    inputs=[
+        gr.Textbox(label="Messages JSON"),
+        gr.Number(value=0.2, label="Temperature"),
+        gr.Number(value=512, label="Max Tokens"),
+        gr.Textbox(value="[]", label="Tools JSON"),
+        gr.Checkbox(value=True, label="Stop after first complete tool call"),
+    ],
+    outputs="text",
+    title="Qwen2.5-Coder-32B AWQ OpenAI-compatible ZeroGPU Backend",
+)
+
+class OpenAIRouteMiddleware(BaseHTTPMiddleware):
+    async def dispatch(self, request: Request, call_next):
+        path = request.url.path.rstrip("/") or "/"
+        
+        if path == "/health" and request.method == "GET":
+            return JSONResponse(health())
+            
+        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 as error:
+                traceback.print_exc()
+                return JSONResponse(
+                    status_code=500,
+                    content={"error": {"message": f"search error: {error}"}},
+                )
+                
+        if path == "/v1/models" and request.method == "GET":
+            return JSONResponse(models())
+            
+        if path == "/v1/chat/completions" and request.method == "POST":
+            try:
+                raw_request = await request.json()
+                parsed_request = ChatCompletionRequest(**raw_request)
+            except (json.JSONDecodeError, ValidationError, TypeError) as error:
+                return JSONResponse(status_code=400, content={"error": {"message": str(error)}})
+            try:
+                return chat_completions(parsed_request)
+            except HTTPException as error:
+                return JSONResponse(status_code=error.status_code, content={"error": {"message": error.detail}})
+            except Exception as error:
+                traceback.print_exc()
+                return JSONResponse(
+                    status_code=500, 
+                    content={"error": {"message": f"internal Space error: {str(error)}"}}
+                )
+                
+        return await call_next(request)
+
+
+import gradio.routes as _groutes
+_original_create_app = _groutes.App.create_app
+
+def _create_app_with_openai_routes(*args, **kwargs):
+    created = _original_create_app(*args, **kwargs)
+    created.add_middleware(OpenAIRouteMiddleware)
+    return created
+
+_groutes.App.create_app = staticmethod(_create_app_with_openai_routes)
+
+demo.queue(default_concurrency_limit=1, max_size=8).launch(show_error=True, ssr_mode=False)

openai_compat_final.py

@@ -0,0 +1,1075 @@
+"""Pure OpenAI compatibility helpers used by the Space endpoint."""
+
+from __future__ import annotations
+
+import json
+import re
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any
+
+from tool_calls import normalize_openai_tool_arguments
+
+
+EMPTY_PARAMETERS = {"type": "object", "properties": {}}
+# OpenClaude includes human-facing operational manuals in tool descriptions.
+# They are useful to its native client but can consume most of the Qwen context
+# once the same catalog is rendered again in the model prompt.  Keep enough
+# context to select and call a tool while preserving the full JSON-schema shape.
+MAX_TOOL_DESCRIPTION_CHARS = 800
+MAX_SCHEMA_DESCRIPTION_CHARS = 240
+
+FAILED_RESULT_RE = re.compile(
+    r"(?im)(?:"
+    r"<tool_use_error>|"
+    r"\bexit\s*(?:code)?\s*[:=]?\s*[1-9]\d*\b|"
+    r"\bstatus\s*(?:code)?\s*[:=]?\s*[345]\d\d\b|"
+    r"^\s*(?:FAILED|ERROR)(?:\s|:)|"
+    r"\b[1-9]\d*\s+(?:failed|errors?)\b|"
+    r"\b(?:command not found|no such file|permission denied|timed out)\b|"
+    r"\b(?:invalid api key|invalid token|unauthorized|forbidden)\b|"
+    r"\b(?:invalid tool parameters|inputvalidationerror)\b|"
+    r"\b(?:required parameter|schema)[^\n]*(?:missing|not sent)\b|"
+    r'"status"\s*:\s*"(?:error|401|403)"|'
+    r'"status"\s*:\s*(?:401|403)\b|'
+    r"\bHTTP/\S+\s+(?:3\d\d|4\d\d|5\d\d)\b"
+    r")"
+)
+VERIFICATION_COMMAND_RE = re.compile(
+    r"(?i)(?:"
+    r"\bpytest\b|"
+    r"\bpython(?:3)?\s+-m\s+(?:unittest|pytest)\b|"
+    r"\bpython(?:3)?\s+[^\n;&|]*test[^\n;&|]*\.py\b|"
+    r"\b(?:npm|pnpm|yarn|bun)\s+(?:run\s+)?test\b|"
+    r"\b(?:cargo|go)\s+test\b|"
+    r"\b(?:cargo)\s+check\b|"
+    r"\b(?:mvn|gradle)\s+(?:test|check|build)\b|"
+    r"\bmake\s+(?:check|test)\b|"
+    r"\b(?:npm|pnpm|yarn|bun)\s+(?:run\s+)?(?:build|check|lint)\b|"
+    r"(?:^|[\s/])(?:bash\s+)?[^\s;&|]*test[^\s;&|]*\.sh\b|"
+    r"\bpython(?:3)?\s+-m\s+py_compile\b|"
+    r"\b(?:ruff|mypy|eslint|tsc)\b"
+    r")"
+)
+POSITIVE_VERIFICATION_RE = re.compile(
+    r"(?im)(?:"
+    r"^\s*OK\s*$|"
+    r"\bRan\s+\d+\s+tests?\b|"
+    r"\b\d+\s+passed\b|"
+    r"\bBUILD\s+SUCCESS(?:FUL)?\b|"
+    r"\b(?:tests?|checks?)\s+(?:passed|successful)\b|"
+    r"\b[A-Z][A-Z0-9_]+_OK\b|"
+    r"\(?(?:Bash )?completed (?:successfully )?"
+    r"(?:with no|without)(?: textual)? output\)?"
+    r")"
+)
+INSPECTION_COMMAND_RE = re.compile(
+    r"(?i)^\s*(?:"
+    r"cd\b[^;&|]*(?:&&|;)\s*)?"
+    r"(?:ls|pwd|find|rg|grep|cat|sed|head|tail|wc|stat|tree|git|cd)"
+    r"\b"
+)
+WEB_REQUEST_RE = re.compile(
+    r"(?i)\b(?:"
+    r"pesquis(?:e|ar|a)|busque|procure|not[ií]cias?|[uú]ltimas?|"
+    r"hoje|agora|atual(?:izado|izada|mente)?|search|latest|news|browser|web"
+    r")\b"
+)
+WEB_SUBJECT_RE = re.compile(
+    r"(?i)\b(?:"
+    r"web|internet|pesquis\w*|busc\w*|procur\w*|not[ií]cias?|"
+    r"search|latest|news|info|site|p[aá]gina"
+    r")\b"
+)
+LOCAL_INSPECTION_RE = re.compile(
+    r"(?i)\b(?:"
+    r"mem[oó]ria|ram|cpu|processador|disco|armazenamento|hardware|"
+    r"sistema|kernel|processos?|servi[cç]os?|rede|endere[cç]o\s+ip|"
+    r"gpu|temperatura|bateria|swap|arquivos?|diret[oó]rios?|pastas?"
+    r")\b"
+)
+INSPECTION_INTENT_RE = re.compile(
+    r"(?i)\b(?:"
+    r"verifi(?:que|car|ca[cç][aã]o)|confira|cheque|inspecione|"
+    r"mostre|liste|diagnostique|analise|check|inspect|show|list|explore"
+    r")\b"
+)
+READ_REQUEST_RE = re.compile(
+    r"(?i)\b(?:leia|ler|read|veja|ver|open|abra)\b"
+)
+EXPLICIT_TOOL_REQUEST_RE = re.compile(
+    r"(?i)\b(?:use|usar|utilize|utilizar|chame|chamar|call|invoke|"
+    r"execute|executar)\s+"
+    r"(?:(?:obrigatoriamente|necessariamente|somente|only|just|"
+    r"a|o|as|os|the|ferramenta|tool)\s+)*"
+    r"(?P<tool>bash|read|write|edit|glob|grep|websearch|webfetch|"
+    r"task|agent|notebookedit|lsp)\b"
+)
+IMPLEMENTATION_REQUEST_RE = re.compile(
+    r"(?i)\b(?:"
+    r"implemente|implement|corrija|corrigir|fix|edite|editar|modify|"
+    r"altere|alterar|crie|criar|create|write|escreva|instale|install|"
+    r"baixe|download|execute|rode|run|teste|testar|automatiz\w*"
+    r")\b"
+)
+PROGRAMMING_CONTEXT_RE = re.compile(
+    r"(?i)\b(?:"
+    r"arquivo|file|c[oó]digo|code|projeto|project|reposit[oó]rio|repo|"
+    r"script|programa|aplica[cç][aã]o|app|fun[cç][aã]o|function|classe|"
+    r"m[oó]dulo|module|teste|test|bug|erro|error|build|site|endpoint|"
+    r"proxy|api|depend[eê]ncia|package|solu[cç][aã]o|funcionalidade|feature"
+    r")\b"
+)
+ACTION_NOW_RE = re.compile(
+    r"(?i)\b(?:fa[cç]a|execute|rode|run|do)\s+(?:isso\s+)?agora\b|"
+    r"\bdo\s+it\s+now\b"
+)
+NO_TOOLS_RE = re.compile(
+    r"(?i)\b(?:"
+    r"n[aã]o\s+(?:use|usar|chame|chamar)|"
+    r"sem|"
+    r"do\s+not\s+(?:use|call)|"
+    r"never\s+(?:use|call)|"
+    r"without"
+    r")\s+(?:as?\s+)?(?:ferramentas?|tools?)\b"
+)
+SIMPLE_GREETING_RE = re.compile(
+    r"(?i)^\s*(?:oi|ol[aá]|hello|hi|hey|bom\s+dia|boa\s+tarde|boa\s+noite)"
+    r"[\s!,.?]*$"
+)
+OPENCLAUDE_METADATA_BLOCK_RE = re.compile(
+    r"<(?P<tag>available-deferred-tools|system-reminder)\b[^>]*>.*?</(?P=tag)>",
+    re.DOTALL | re.IGNORECASE,
+)
+
+
+@dataclass(frozen=True)
+class ToolFlowState:
+    """Request-local progress state; no conversation state is stored globally."""
+
+    active: bool = False
+    requires_tool: bool = False
+    can_finalize: bool = False
+    reason: str = ""
+    instruction: str | None = None
+    forced_tool: str | None = None
+
+
+@dataclass(frozen=True)
+class _ToolResultEvent:
+    name: str
+    arguments: dict[str, Any]
+    content: str
+    is_error: bool
+    batch: int
+
+
+def _bounded_description(value: Any, limit: int) -> str:
+    """Return a compact single-line description suitable for a model prompt."""
+    text = re.sub(r"\s+", " ", str(value or "")).strip()
+    if len(text) <= limit:
+        return text
+    shortened = text[: max(1, limit - 1)].rsplit(" ", 1)[0].rstrip()
+    return (shortened or text[: limit - 1]).rstrip() + "…"
+
+
+def _compact_schema_descriptions(value: Any) -> Any:
+    """Bound schema prose without removing structural validation information."""
+    if isinstance(value, Mapping):
+        return {
+            key: (
+                _bounded_description(raw_value, MAX_SCHEMA_DESCRIPTION_CHARS)
+                if key == "description"
+                else _compact_schema_descriptions(raw_value)
+            )
+            for key, raw_value in value.items()
+        }
+    if isinstance(value, list):
+        return [_compact_schema_descriptions(item) for item in value]
+    return value
+
+
+def _content_text(content: Any) -> str:
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        parts: list[str] = []
+        for block in content:
+            if isinstance(block, Mapping):
+                text = block.get("text", block.get("content", ""))
+                if text:
+                    parts.append(str(text))
+            elif block is not None:
+                parts.append(str(block))
+        return "\n".join(parts)
+    return "" if content is None else str(content)
+
+
+def _user_request_text(content: Any) -> str:
+    """Remove OpenClaude's injected metadata before classifying user intent.
+
+    OpenClaude places deferred-tool lists, skill descriptions, and snip markers
+    inside a user-role message.  Those blocks can contain words such as
+    ``create``, ``code``, or ``test``; treating them as the user's request can
+    incorrectly force ``tool_choice=required`` for a plain greeting.
+    """
+    text = _content_text(content)
+    previous = None
+    while text != previous:
+        previous = text
+        text = OPENCLAUDE_METADATA_BLOCK_RE.sub("", text)
+    return text.strip()
+
+
+def _call_arguments(value: Any) -> dict[str, Any]:
+    if isinstance(value, Mapping):
+        return dict(value)
+    if isinstance(value, str):
+        try:
+            parsed = json.loads(value)
+        except json.JSONDecodeError:
+            return {}
+        return dict(parsed) if isinstance(parsed, Mapping) else {}
+    return {}
+
+
+def _is_synthetic_continuation(message: Mapping[str, Any]) -> bool:
+    content = message.get("content")
+    if isinstance(content, list) and any(
+        isinstance(block, Mapping) and block.get("type") == "tool_result"
+        for block in content
+    ):
+        return True
+    text = _content_text(content).casefold()
+    return (
+        not text.strip()
+        or "[tool results received]" in text
+        or (
+            "continue with the task" in text
+            and "resume your thought" in text
+        )
+        or (
+            "<system-reminder" in text
+            and not re.sub(
+                r"<system-reminder\b[^>]*>.*?</system-reminder>",
+                "",
+                text,
+                flags=re.DOTALL | re.IGNORECASE,
+            ).strip()
+        )
+    )
+
+
+def _current_turn_messages(messages: object) -> list[object]:
+    if not isinstance(messages, list):
+        return []
+    start = 0
+    for index, message in enumerate(messages):
+        if (
+            isinstance(message, Mapping)
+            and str(message.get("role", "")).casefold() == "user"
+            and not _is_synthetic_continuation(message)
+        ):
+            start = index
+    return messages[start:]
+
+
+def _tool_result_events(messages: object) -> list[_ToolResultEvent]:
+    current_messages = _current_turn_messages(messages)
+    calls_by_id: dict[str, tuple[str, dict[str, Any], int]] = {}
+    pending_order: list[str] = []
+    events: list[_ToolResultEvent] = []
+    batch = 0
+
+    for message in current_messages:
+        if not isinstance(message, Mapping):
+            continue
+        role = str(message.get("role", "")).casefold()
+        if role == "assistant":
+            raw_calls = message.get("tool_calls") or []
+            if raw_calls:
+                batch += 1
+            for index, raw_call in enumerate(raw_calls):
+                if not isinstance(raw_call, Mapping):
+                    continue
+                function = raw_call.get("function")
+                if not isinstance(function, Mapping):
+                    continue
+                name = function.get("name")
+                if not isinstance(name, str) or not name:
+                    continue
+                call_id = raw_call.get("id")
+                if not isinstance(call_id, str) or not call_id:
+                    call_id = f"__ordered_{len(calls_by_id)}_{index}"
+                calls_by_id[call_id] = (
+                    name,
+                    _call_arguments(function.get("arguments", {})),
+                    batch,
+                )
+                pending_order.append(call_id)
+            continue
+        if role != "tool":
+            continue
+
+        call_id = message.get("tool_call_id")
+        call: tuple[str, dict[str, Any], int] | None = None
+        if isinstance(call_id, str) and call_id:
+            call = calls_by_id.pop(call_id, None)
+            if call_id in pending_order:
+                pending_order.remove(call_id)
+        elif pending_order:
+            fallback_id = pending_order.pop(0)
+            call = calls_by_id.pop(fallback_id, None)
+
+        if call is None:
+            explicit_name = message.get("name")
+            if not isinstance(explicit_name, str) or not explicit_name:
+                continue
+            call = (explicit_name, {}, batch)
+
+        content = _content_text(message.get("content"))
+        structured_error = message.get("is_error") is True
+        if isinstance(message.get("content"), list):
+            structured_error = structured_error or any(
+                isinstance(block, Mapping) and block.get("is_error") is True
+                for block in message["content"]
+            )
+        events.append(
+            _ToolResultEvent(
+                name=call[0],
+                arguments=call[1],
+                content=content,
+                is_error=structured_error or bool(FAILED_RESULT_RE.search(content)),
+                batch=call[2],
+            )
+        )
+    return events
+
+
+def _bash_command(event: _ToolResultEvent) -> str:
+    command = event.arguments.get("command", event.arguments.get("cmd", ""))
+    return command if isinstance(command, str) else str(command)
+
+
+def _bash_proves_completion(event: _ToolResultEvent) -> bool:
+    if event.is_error:
+        return False
+    command = _bash_command(event)
+    if not VERIFICATION_COMMAND_RE.search(command):
+        return False
+    return bool(POSITIVE_VERIFICATION_RE.search(event.content))
+
+
+def _latest_user_request(messages: object) -> str:
+    requests: list[str] = []
+    if not isinstance(messages, list):
+        return ""
+    for message in messages:
+        if (
+            isinstance(message, Mapping)
+            and str(message.get("role", "")).casefold() == "user"
+            and not _is_synthetic_continuation(message)
+        ):
+            text = _user_request_text(message.get("content"))
+            if text:
+                requests.append(text)
+    if not requests:
+        return ""
+    latest = requests[-1]
+    if len(requests) > 1 and ACTION_NOW_RE.search(latest):
+        return requests[-2] + "\n" + latest
+    return latest
+
+
+def is_simple_greeting(messages: object) -> bool:
+    """Identify a greeting that does not need a model or tool prompt.
+
+    OpenClaude sends its complete tool catalog even for ``ola``. Calling a
+    model on ZeroGPU for that turn adds unnecessary queue time, so the API can
+    answer it deterministically before inference.
+    """
+    return bool(SIMPLE_GREETING_RE.fullmatch(_latest_user_request(messages)))
+
+
+def _explicitly_disables_tools(messages: object) -> bool:
+    if not isinstance(messages, list):
+        return False
+    return any(
+        isinstance(message, Mapping)
+        and str(message.get("role", "")).casefold()
+        in {"system", "developer", "user"}
+        and bool(NO_TOOLS_RE.search(_content_text(message.get("content"))))
+        for message in messages
+    )
+
+
+def _initial_tool_flow(
+    messages: object,
+    available_by_fold: Mapping[str, str],
+) -> ToolFlowState:
+    """Force action for concrete first-turn requests instead of accepting plans."""
+    request = _latest_user_request(messages)
+    if not request or not available_by_fold:
+        return ToolFlowState()
+
+    explicit_tool = EXPLICIT_TOOL_REQUEST_RE.search(request)
+    if explicit_tool:
+        requested_name = explicit_tool.group("tool").casefold()
+        forced_tool = available_by_fold.get(requested_name)
+        if forced_tool is None:
+            forced_tool = available_by_fold.get(
+                {"agent": "task", "task": "agent"}.get(requested_name, "")
+            )
+        if forced_tool is not None:
+            return ToolFlowState(
+                active=True,
+                requires_tool=True,
+                reason=f"the user explicitly requested the {forced_tool} tool",
+                instruction=(
+                    f"OPENCLAUDE FLOW STATE: call {forced_tool} now because the "
+                    "user explicitly requested it. Do not print a sample call "
+                    "as prose and do not answer with a plan."
+                ),
+                forced_tool=forced_tool,
+            )
+
+    if (
+        "websearch" in available_by_fold
+        and WEB_REQUEST_RE.search(request)
+        and WEB_SUBJECT_RE.search(request)
+    ):
+        return ToolFlowState(
+            active=True,
+            requires_tool=True,
+            reason="the user requested current web research",
+            instruction=(
+                "OPENCLAUDE FLOW STATE: perform the requested research now. "
+                "Call WebSearch with a concise query; do not merely describe how "
+                "you would search and do not substitute curl or invented APIs."
+            ),
+            forced_tool=available_by_fold["websearch"],
+        )
+
+    if (
+        "bash" in available_by_fold
+        and LOCAL_INSPECTION_RE.search(request)
+        and INSPECTION_INTENT_RE.search(request)
+    ):
+        return ToolFlowState(
+            active=True,
+            requires_tool=True,
+            reason="the user requested inspection of the local system",
+            instruction=(
+                "OPENCLAUDE FLOW STATE: inspect the local system now. Call Bash "
+                "with a safe read-only command that directly answers the request; "
+                "do not print a command as prose and do not ask for confirmation."
+            ),
+            forced_tool=available_by_fold["bash"],
+        )
+
+    if "read" in available_by_fold and READ_REQUEST_RE.search(request):
+        return ToolFlowState(
+            active=True,
+            requires_tool=True,
+            reason="the user explicitly requested reading a file",
+            instruction=(
+                "OPENCLAUDE FLOW STATE: call Read now for the relevant file. "
+                "Do not describe a future read operation."
+            ),
+            forced_tool=available_by_fold["read"],
+        )
+
+    concrete_implementation = bool(
+        IMPLEMENTATION_REQUEST_RE.search(request)
+        and (
+            PROGRAMMING_CONTEXT_RE.search(request)
+            or re.search(r"(?i)\bautomatiz\w*\b", request)
+        )
+    )
+    if ACTION_NOW_RE.search(request) or concrete_implementation:
+        return ToolFlowState(
+            active=True,
+            requires_tool=True,
+            reason="the user requested immediate tool-backed action",
+            instruction=(
+                "OPENCLAUDE FLOW STATE: act on the request now by calling one "
+                "appropriate available tool. Do not answer with a plan, example "
+                "commands, or a request for the user to repeat the task."
+            ),
+        )
+
+    # OpenClaude may send ``tool_choice=required`` even for greetings and
+    # other conversational turns.  Those turns must be allowed to finalize;
+    # requiring a synthetic tool call makes a harmless "oi" become a 502.
+    return ToolFlowState(reason="no concrete tool action was requested")
+
+
+def analyze_tool_flow(
+    messages: object,
+    raw_tools: object,
+) -> ToolFlowState:
+    """Derive whether an agent must continue or may emit its final response."""
+    if _explicitly_disables_tools(messages):
+        return ToolFlowState(
+            can_finalize=True,
+            reason="the request explicitly disables all tools",
+        )
+    available_by_fold = {
+        tool["function"]["name"].casefold(): tool["function"]["name"]
+        for tool in normalize_tools(raw_tools)
+    }
+    available = set(available_by_fold)
+    events = _tool_result_events(messages)
+    if not events:
+        return _initial_tool_flow(messages, available_by_fold)
+
+    # A successful search/fetch is terminal evidence for a research request.
+    # This intentionally prevents WebSearch -> WebFetch -> repeated curl loops.
+    web_evidence = any(
+        event.name.casefold() in {"websearch", "webfetch"}
+        and not event.is_error
+        and bool(event.content.strip())
+        for event in events
+    )
+
+    request = _latest_user_request(messages)
+    agentic_intent = not request or bool(
+        IMPLEMENTATION_REQUEST_RE.search(request)
+        and (
+            PROGRAMMING_CONTEXT_RE.search(request)
+            or re.search(r"(?i)\bautomatiz\w*\b", request)
+        )
+    )
+    agentic = (
+        agentic_intent
+        and "bash" in available
+        and bool({"edit", "write"} & available)
+    )
+    dirty = False
+    dirty_batch = -1
+    agentic_started = False
+    last_reason = ""
+
+    if agentic:
+        for event in events:
+            name = event.name.casefold()
+            if name == "read":
+                agentic_started = True
+                dirty = True
+                dirty_batch = max(dirty_batch, event.batch)
+                last_reason = "files were inspected but implementation is still pending"
+            elif name in {"edit", "write"}:
+                agentic_started = True
+                dirty = True
+                dirty_batch = max(dirty_batch, event.batch)
+                last_reason = "files changed and must be verified with Bash"
+            elif event.is_error and agentic_started:
+                dirty = True
+                dirty_batch = max(dirty_batch, event.batch)
+                last_reason = f"{event.name} returned an error that must be recovered"
+            elif name == "bash":
+                command = _bash_command(event)
+                if event.is_error:
+                    agentic_started = True
+                    dirty = True
+                    dirty_batch = max(dirty_batch, event.batch)
+                    last_reason = "the Bash command or test failed"
+                elif INSPECTION_COMMAND_RE.search(command):
+                    agentic_started = True
+                    dirty = True
+                    dirty_batch = max(dirty_batch, event.batch)
+                    last_reason = "inspection output is not completion evidence"
+                elif (
+                    agentic_started
+                    and dirty
+                    and event.batch > dirty_batch
+                    and _bash_proves_completion(event)
+                ):
+                    dirty = False
+                    last_reason = "a Bash verification passed after the latest change"
+                elif agentic_started and dirty:
+                    last_reason = "Bash did not provide positive test evidence"
+
+    if agentic_started and dirty:
+        return ToolFlowState(
+            active=True,
+            requires_tool=True,
+            reason=last_reason,
+            instruction=(
+                "OPENCLAUDE FLOW STATE: the task is not complete. "
+                f"Reason: {last_reason}. Call exactly one appropriate tool now; "
+                "do not describe a future plan. After reading, edit or write the "
+                "implementation. After changes, use Bash to run the requested "
+                "tests and continue fixing failures until the output proves success."
+            ),
+        )
+
+    if agentic_started and not dirty:
+        return ToolFlowState(
+            active=True,
+            can_finalize=True,
+            reason=last_reason,
+            instruction=(
+                "OPENCLAUDE FLOW STATE: verification passed after the latest "
+                "change. Do not call another tool. Report the completed work and "
+                "the test evidence directly in Brazilian Portuguese."
+            ),
+        )
+
+    if web_evidence:
+        return ToolFlowState(
+            active=True,
+            can_finalize=True,
+            reason="usable web evidence is available",
+            instruction=(
+                "OPENCLAUDE FLOW STATE: usable WebSearch/WebFetch results are "
+                "already available. Do not call WebFetch, Bash, curl, or another "
+                "tool. Synthesize a concrete answer now from the supplied results, "
+                "include useful source links, and never invent API keys or facts."
+            ),
+        )
+
+    last_webfetch_error = max(
+        (
+            index
+            for index, event in enumerate(events)
+            if event.name.casefold() == "webfetch" and event.is_error
+        ),
+        default=-1,
+    )
+    last_websearch_error = max(
+        (
+            index
+            for index, event in enumerate(events)
+            if event.name.casefold() == "websearch" and event.is_error
+        ),
+        default=-1,
+    )
+    toolsearch_recovered = (
+        last_webfetch_error >= 0
+        and any(
+            index > last_webfetch_error
+            and event.name.casefold() == "toolsearch"
+            and not event.is_error
+            for index, event in enumerate(events)
+        )
+    )
+
+    forced_tool: str | None = None
+    recovery = ""
+    web_error_name = ""
+    if last_webfetch_error >= 0:
+        web_error_name = "WebFetch"
+        if toolsearch_recovered and "webfetch" in available:
+            forced_tool = available_by_fold["webfetch"]
+            recovery = (
+                "Retry WebFetch now with both required fields: url and prompt."
+            )
+        elif "webfetch" not in available and "toolsearch" in available:
+            forced_tool = available_by_fold["toolsearch"]
+            recovery = (
+                "Load WebFetch by calling ToolSearch with query select:WebFetch."
+            )
+        elif "webfetch" in available:
+            forced_tool = available_by_fold["webfetch"]
+            recovery = (
+                "Retry WebFetch with both required fields: url and prompt."
+            )
+        elif "websearch" in available:
+            forced_tool = available_by_fold["websearch"]
+            recovery = "Recover with WebSearch using a concise, relevant query."
+    elif last_websearch_error >= 0 and "websearch" in available:
+        web_error_name = "WebSearch"
+        forced_tool = available_by_fold["websearch"]
+        recovery = "Retry WebSearch using a concise, relevant query."
+
+    if forced_tool:
+        return ToolFlowState(
+            active=True,
+            requires_tool=True,
+            reason=f"{web_error_name} returned an error",
+            instruction=(
+                f"OPENCLAUDE FLOW STATE: {web_error_name} failed. "
+                f"{recovery} Do not answer with a plan and do not invent "
+                "credentials, endpoints, or placeholder tokens."
+            ),
+            forced_tool=forced_tool,
+        )
+
+    # Read already provides the requested evidence. Mark it terminal so
+    # OpenClaude's repeated ``tool_choice=required`` does not make a small
+    # model call Read forever. Keep generic Bash inspection neutral: the
+    # existing flow still lets the model decide how to summarize it.
+    last_event = events[-1]
+    if (
+        last_event.name.casefold() == "read"
+        and not last_event.is_error
+        and bool(last_event.content.strip())
+    ):
+        return ToolFlowState(
+            active=True,
+            can_finalize=True,
+            reason="a successful Read result is available",
+            instruction=(
+                "OPENCLAUDE FLOW STATE: the requested Read tool returned usable "
+                "evidence. Do not call another tool; synthesize the answer "
+                "directly from the result in Brazilian Portuguese."
+            ),
+        )
+
+    return ToolFlowState()
+
+
+def resolve_tool_choice(
+    requested_choice: object,
+    state: ToolFlowState,
+) -> object:
+    """Resolve client tool choice against the reconstructed conversation state.
+
+    A concrete function choice remains authoritative. ``required`` is slightly
+    different for OpenClaude: the client repeats it across turns, so the Space
+    must narrow it to a detected function, or downgrade it to ``none`` once a
+    usable tool result is already available / no concrete tool action exists.
+    """
+    if isinstance(requested_choice, Mapping):
+        return requested_choice
+
+    requested_mode = (
+        requested_choice.casefold()
+        if isinstance(requested_choice, str)
+        else None
+    )
+
+    if requested_mode == "required":
+        if state.requires_tool:
+            if state.forced_tool:
+                return {
+                    "type": "function",
+                    "function": {"name": state.forced_tool},
+                }
+            return "required"
+        if state.can_finalize or state.reason == "no concrete tool action was requested":
+            return "none"
+        return "required"
+
+    is_auto = requested_choice is None or requested_mode == "auto"
+    if not is_auto:
+        return requested_choice
+
+    if state.can_finalize or state.reason == "no concrete tool action was requested":
+        return "none"
+    if state.requires_tool:
+        if state.forced_tool:
+            return {
+                "type": "function",
+                "function": {"name": state.forced_tool},
+            }
+        return "required"
+    return requested_choice
+
+
+def normalize_tools(raw_tools: object) -> list[dict[str, Any]]:
+    """Return valid function definitions for Qwen's native tool template."""
+    if not isinstance(raw_tools, list):
+        return []
+
+    normalized: list[dict[str, Any]] = []
+    for raw_tool in raw_tools:
+        if not isinstance(raw_tool, Mapping):
+            continue
+        function = raw_tool.get("function")
+        candidate = function if isinstance(function, Mapping) else raw_tool
+        name = candidate.get("name")
+        if not isinstance(name, str) or not name:
+            continue
+        parameters = candidate.get(
+            "parameters", candidate.get("input_schema", EMPTY_PARAMETERS)
+        )
+        if not isinstance(parameters, Mapping):
+            parameters = EMPTY_PARAMETERS
+        normalized.append(
+            {
+                "type": "function",
+                "function": {
+                    "name": name,
+                    "description": _bounded_description(
+                        candidate.get("description"), MAX_TOOL_DESCRIPTION_CHARS
+                    ),
+                    "parameters": _compact_schema_descriptions(parameters),
+                },
+            }
+        )
+    return normalized
+
+
+def select_tools(
+    raw_tools: object,
+    tool_choice: object,
+) -> tuple[list[dict[str, Any]], str]:
+    """Apply OpenAI ``tool_choice`` semantics before prompting the model.
+
+    The returned mode is one of ``auto``, ``none``, ``required``, or
+    ``forced``. A forced choice only exposes the selected function to Qwen,
+    which is the most reliable way to enforce it with a native tool template.
+    """
+    tools = normalize_tools(raw_tools)
+    if tool_choice is None:
+        return tools, "auto"
+
+    if isinstance(tool_choice, str):
+        mode = tool_choice.casefold()
+        if mode == "none":
+            return [], "none"
+        if mode in {"auto", "required"}:
+            if mode == "required" and not tools:
+                raise ValueError("tool_choice='required' needs at least one tool")
+            return tools, mode
+        raise ValueError(f"Unsupported tool_choice: {tool_choice}")
+
+    if not isinstance(tool_choice, Mapping):
+        raise ValueError("tool_choice must be 'auto', 'none', 'required', or a function")
+    function = tool_choice.get("function")
+    name = function.get("name") if isinstance(function, Mapping) else None
+    if tool_choice.get("type") != "function" or not isinstance(name, str) or not name:
+        raise ValueError("Forced tool_choice must contain function.name")
+
+    selected = [
+        tool
+        for tool in tools
+        if tool["function"]["name"].casefold() == name.casefold()
+    ]
+    if not selected:
+        raise ValueError(f"Forced tool is not defined in tools: {name}")
+    return selected[:1], "forced"
+
+
+def tool_names(tools: list[dict[str, Any]]) -> set[str]:
+    return {tool["function"]["name"] for tool in tools}
+
+
+def indexed_tool_calls(calls: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Add the per-call index required in streamed OpenAI deltas."""
+    return [{**call, "index": index} for index, call in enumerate(calls)]
+
+
+def tool_choice_instruction(mode: str, tools: list[dict[str, Any]]) -> str | None:
+    """Supply the constraint that Qwen's template cannot express directly."""
+    if mode == "required":
+        return "You must call one or more of the available tools in this response."
+    if mode == "forced":
+        return (
+            f"You must call the {tools[0]['function']['name']} tool in this response. "
+            "Do not answer with plain text."
+        )
+    return None
+
+
+def _schema_example(parameters: object) -> dict[str, Any]:
+    if not isinstance(parameters, Mapping):
+        return {}
+    properties = parameters.get("properties")
+    if not isinstance(properties, Mapping):
+        return {}
+    required = parameters.get("required")
+    keys = required if isinstance(required, list) and required else list(properties)[:1]
+    example: dict[str, Any] = {}
+    for key in keys:
+        if not isinstance(key, str):
+            continue
+        raw_schema = properties.get(key)
+        schema = raw_schema if isinstance(raw_schema, Mapping) else {}
+        value_type = schema.get("type")
+        if value_type in {"integer", "number"}:
+            value: Any = 1
+        elif value_type == "boolean":
+            value = True
+        elif value_type == "array":
+            value = []
+        elif value_type == "object":
+            value = {}
+        elif "path" in key.casefold():
+            value = "/absolute/path"
+        elif "query" in key.casefold():
+            value = "search terms"
+        elif key.casefold() == "url":
+            value = "https://example.com"
+        else:
+            value = "value"
+        example[key] = value
+    return example
+
+
+def tool_protocol_instruction(
+    tools: list[dict[str, Any]],
+    parallel_tool_calls: bool = False,
+) -> str | None:
+    """Return the complete notebook-agent contract enforced by the Space."""
+    if not tools:
+        return None
+    lines = [
+        "OPENAI TOOL CALL FORMAT — MANDATORY",
+        "You are operating on the user's real notebook, not a simulation.",
+        "Always communicate with the user in Brazilian Portuguese (pt-BR).",
+        "Perform requested implementation, diagnosis, download, execution, "
+        "testing, local inspection, or current web research with the available "
+        "tools instead of describing commands or a future plan.",
+        "Never claim that a file changed, a command ran, or a test passed unless "
+        "a tool result in this conversation proves it.",
+        "After WebSearch or WebFetch returns usable evidence, synthesize the "
+        "answer from it. Do not fall back to repeated curl calls.",
+        "Never invent API keys, tokens, endpoints, or placeholder credentials.",
+        "For greetings, small talk, or a self-contained factual answer, respond "
+        "directly without a tool unless the flow state below requires one.",
+        (
+            "When calling tools, emit one or more complete tool calls and no prose, "
+            "Markdown, or code fence. Multiple calls are allowed only when they are "
+            "independent and can run in parallel."
+            if parallel_tool_calls
+            else "When calling a tool, emit exactly one call and no prose, Markdown, "
+            "or code fence."
+        ),
+        'Exact syntax: <tool_call>{"name":"TOOL_NAME","arguments":{"key":"value"}}</tool_call>',
+        "Arguments must be valid JSON matching the selected schema.",
+        "Available tools:",
+    ]
+    available_names = {
+        str(tool.get("function", {}).get("name", "")).casefold()
+        for tool in tools
+        if isinstance(tool.get("function"), Mapping)
+    }
+    if "webfetch" in available_names:
+        lines.insert(
+            5,
+            "Call only a tool listed below. Follow every tool schema exactly. "
+            "WebFetch requires both url and prompt; never omit required fields.",
+        )
+    else:
+        lines.insert(
+            5,
+            "Call only a tool listed below. Deferred tools are unavailable in "
+            "this backend; explain when a needed capability is not listed "
+            "instead of invoking an unlisted tool.",
+        )
+    first_example: tuple[str, dict[str, Any]] | None = None
+    for tool in tools:
+        function = tool.get("function")
+        if not isinstance(function, Mapping):
+            continue
+        name = function.get("name")
+        if not isinstance(name, str) or not name:
+            continue
+        parameters = function.get("parameters")
+        lines.append(
+            json.dumps(
+                {
+                    "name": name,
+                    "description": str(function.get("description") or ""),
+                    "parameters": (
+                        dict(parameters)
+                        if isinstance(parameters, Mapping)
+                        else EMPTY_PARAMETERS
+                    ),
+                },
+                ensure_ascii=False,
+                separators=(",", ":"),
+            )
+        )
+        if first_example is None:
+            first_example = (name, _schema_example(parameters))
+    if first_example:
+        lines.append(
+            "Example syntax: <tool_call>"
+            + json.dumps(
+                {
+                    "name": first_example[0],
+                    "arguments": first_example[1],
+                },
+                ensure_ascii=False,
+                separators=(",", ":"),
+            )
+            + "</tool_call>"
+        )
+    return "\n".join(lines)
+
+
+def text_content(content: Any) -> str:
+    """Convert text-only OpenAI message blocks into chat-template text."""
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        return "\n".join(
+            block.get("text", "")
+            for block in content
+            if isinstance(block, Mapping)
+            and block.get("type") in {"text", "input_text"}
+        )
+    return "" if content is None else str(content)
+
+
+def normalized_tool_calls(raw_calls: object) -> list[dict[str, Any]]:
+    """Keep valid OpenAI calls in the shape Qwen's template understands."""
+    if not isinstance(raw_calls, list):
+        return []
+
+    calls: list[dict[str, Any]] = []
+    for raw_call in raw_calls:
+        if not isinstance(raw_call, Mapping):
+            continue
+        function = raw_call.get("function")
+        if not isinstance(function, Mapping):
+            continue
+        name = function.get("name")
+        if not isinstance(name, str) or not name:
+            continue
+        call: dict[str, Any] = {
+            "type": "function",
+            "function": {
+                "name": name,
+                "arguments": normalize_openai_tool_arguments(
+                    function.get("arguments", {})
+                ),
+            },
+        }
+        if isinstance(raw_call.get("id"), str) and raw_call["id"]:
+            call["id"] = raw_call["id"]
+        calls.append(call)
+    return calls
+
+
+def normalize_messages(
+    messages: list[dict[str, Any]],
+    extra_system_instruction: str | None = None,
+) -> list[dict[str, Any]]:
+    """Normalize multimodal content while preserving native tool history."""
+    normalized: list[dict[str, Any]] = []
+    for message in messages:
+        raw_role = str(message.get("role", "user")).lower()
+        if raw_role in {"system", "developer"}:
+            role = "system"
+        elif raw_role in {"assistant", "tool"}:
+            role = raw_role
+        else:
+            role = "user"
+
+        entry: dict[str, Any] = {
+            "role": role,
+            "content": text_content(message.get("content")),
+        }
+        if role == "assistant":
+            calls = normalized_tool_calls(message.get("tool_calls"))
+            if calls:
+                entry["tool_calls"] = calls
+        if role == "tool" and isinstance(message.get("tool_call_id"), str):
+            entry["tool_call_id"] = message["tool_call_id"]
+        normalized.append(entry)
+
+    if extra_system_instruction:
+        if normalized and normalized[0]["role"] == "system":
+            normalized[0]["content"] = (
+                f"{normalized[0]['content']}\n\n{extra_system_instruction}"
+            ).strip()
+        else:
+            normalized.insert(
+                0, {"role": "system", "content": extra_system_instruction}
+            )
+    return normalized

requirements_final.txt

@@ -0,0 +1,26 @@
+# PyTorch CUDA 13.0 wheels. ZeroGPU log confirmed torch 2.11.0+cu130.
+--extra-index-url https://download.pytorch.org/whl/cu130
+
+# Space/API runtime
+fastapi>=0.115,<1
+pydantic>=2.10,<3
+httpx>=0.27,<1
+gradio==6.22.0
+
+# Pin the exact ZeroGPU-compatible PyTorch pair.
+torch==2.11.0
+torchvision==0.26.0
+
+# GPTQModel 7.3.2 declares accelerate>=1.13.0, torch>=2.8,
+# safetensors>=0.7 and transformers>=5.4.
+accelerate>=1.13.0,<2
+safetensors>=0.7.0
+pillow>=11.3.0
+
+# AWQ runtime verified by the Space startup log.
+transformers==5.14.1
+gptqmodel==7.3.2
+
+# Dependencies used by helper modules in this repository.
+tiktoken>=0.9
+sentencepiece>=0.2