1bomopenai_compat.py

"""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:
    """Override only auto/default choices; explicit client choices win."""
    # Se o cliente (OpenClaude) enviou 'required' ou um objeto de função específico,
    # devemos honrar isso independente da nossa análise de fluxo, a menos que
    # ferramentas estejam explicitamente desabilitadas no sistema.
    if isinstance(requested_choice, (dict, Mapping)) or (
        isinstance(requested_choice, str) and requested_choice.casefold() == "required"
    ):
        return requested_choice

    if (
        (state.can_finalize or state.reason == "no concrete tool action was requested")
        and (
            requested_choice is None
            or (
                isinstance(requested_choice, str)
                and requested_choice.casefold() == "auto"
            )
        )
    ):
        # Apenas para 'auto' ou nulo em saudações simples, forçamos 'none' para economizar GPU.
        # Mas se for 'required', o bloco acima já terá retornado.
        return "none"
    # OpenClaude commonly sends ``required`` after naming a concrete tool in
    # the user request. Restrict that choice to the detected function so small
    # models do not have to guess among tools and return prose instead.
    if (
        state.requires_tool
        and state.forced_tool
        and isinstance(requested_choice, str)
        and requested_choice.casefold() == "required"
    ):
        return {
            "type": "function",
            "function": {"name": state.forced_tool},
        }
    is_auto = requested_choice is None or (
        isinstance(requested_choice, str)
        and requested_choice.casefold() == "auto"
    )
    if not is_auto:
        return requested_choice
    if state.requires_tool:
        if state.forced_tool:
            return {
                "type": "function",
                "function": {"name": state.forced_tool},
            }
        return "required"
    if state.can_finalize:
        return "none"
    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]]) -> 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 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