Upload 2 files

openai_compat_loopfix.py

@@ -0,0 +1,1155 @@
+"""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*(?:sudo\s+)?(?:"
+    r"cd\b[^;&|]*(?:&&|;)\s*)?"
+    r"(?:"
+    r"ls|pwd|find|rg|grep|cat|sed|head|tail|wc|stat|tree|git|cd|"
+    r"apt(?:-get)?\s+(?:list|show|policy)|"
+    r"apt-cache\s+(?:policy|show|search)|"
+    r"dpkg\s+(?:-l|--list|--status)|"
+    r"command\s+-v|which|type"
+    r")\b"
+)
+SYSTEM_MAINTENANCE_REQUEST_RE = re.compile(
+    r"(?i)\b(?:"
+    r"atualize|atualizar|atualiza[cç][aã]o|upgrade|update|"
+    r"instale|instalar|instala[cç][aã]o|install|"
+    r"remova|remover|remo[cç][aã]o|remove|"
+    r"desinstale|desinstalar|desinstala[cç][aã]o|uninstall"
+    r")\b"
+)
+SYSTEM_MUTATION_COMMAND_RE = re.compile(
+    r"(?i)^\s*(?:sudo\s+)?(?:"
+    r"apt(?:-get)?\s+(?:update|upgrade|full-upgrade|dist-upgrade|install|remove|purge)|"
+    r"dnf\s+(?:upgrade|update|install|remove)|"
+    r"yum\s+(?:update|upgrade|install|remove)|"
+    r"pacman\s+-S|zypper\s+(?:update|install|remove)"
+    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"pacotes?|packages?|depend[eê]ncias?|dependencies"
+    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,
+        )
+
+    # A successful evidence-producing tool must be able to terminate a turn.
+    # OpenClaude repeats ``tool_choice=required`` across tool-result requests;
+    # leaving the state neutral therefore forces another call even when the
+    # previous action already answered the user's request.
+    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."
+            ),
+        )
+
+    if last_event.name.casefold() == "bash" and not last_event.is_error:
+        command = _bash_command(last_event)
+        maintenance_requested = bool(SYSTEM_MAINTENANCE_REQUEST_RE.search(request))
+        inspection_completed = bool(INSPECTION_COMMAND_RE.search(command))
+        maintenance_completed = bool(SYSTEM_MUTATION_COMMAND_RE.search(command))
+
+        # Read-only inspection is terminal unless the same user request also
+        # asks for a package/system mutation. A successful mutation command is
+        # terminal by itself. This prevents required -> Bash -> required loops
+        # such as apt-list / apt-upgrade / apt-list repeating forever.
+        if maintenance_completed or (inspection_completed and not maintenance_requested):
+            return ToolFlowState(
+                active=True,
+                can_finalize=True,
+                reason=(
+                    "the requested system maintenance command completed"
+                    if maintenance_completed
+                    else "the requested Bash inspection returned usable evidence"
+                ),
+                instruction=(
+                    "OPENCLAUDE FLOW STATE: the latest Bash result is sufficient "
+                    "for this request. Do not repeat Bash or print another tool "
+                    "call as prose. Summarize the actual tool result now in "
+                    "Brazilian Portuguese."
+                ),
+            )
+
+    # Hard loop guard: if the same successful tool call has already executed
+    # more than once in the current turn, never ask the model to repeat it.
+    if not last_event.is_error:
+        last_signature = (
+            last_event.name.casefold(),
+            json.dumps(last_event.arguments, ensure_ascii=False, sort_keys=True),
+        )
+        repeated = sum(
+            1
+            for event in events
+            if not event.is_error
+            and (
+                event.name.casefold(),
+                json.dumps(event.arguments, ensure_ascii=False, sort_keys=True),
+            )
+            == last_signature
+        )
+        if repeated > 1:
+            return ToolFlowState(
+                active=True,
+                can_finalize=True,
+                reason="the same successful tool call was already repeated",
+                instruction=(
+                    "OPENCLAUDE FLOW STATE: stop the tool loop now. The same "
+                    "successful call has already run more than once. Do not "
+                    "repeat it; summarize the available results."
+                ),
+            )
+
+    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.",
+        "Never repeat a tool call whose successful result is already present in "
+        "the conversation; use that result and continue or finalize instead.",
+        "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

tool_calls_loopfix.py

@@ -0,0 +1,462 @@
+"""Translate common Qwen/OpenClaude textual tool calls to OpenAI payloads."""
+
+from __future__ import annotations
+
+import ast
+import html
+import json
+import re
+import shlex
+import uuid
+from collections.abc import Mapping
+from typing import Any
+
+
+JSON_TOOL_CALL_RE = re.compile(
+    r"<tool_call>\s*(?P<payload>\{.*?\})\s*</tool_call>",
+    re.DOTALL | re.IGNORECASE,
+)
+XML_JSON_TOOL_CALL_RE = re.compile(
+    r"<xml>\s*(?P<payload>\{.*?\})\s*</xml>",
+    re.DOTALL | re.IGNORECASE,
+)
+STANDARD_XML_TOOL_CALL_RE = re.compile(
+    r"<tool_call>\s*(?P<body>.*?)\s*</tool_call>",
+    re.DOTALL | re.IGNORECASE,
+)
+FUNCTION_NAME_RE = re.compile(
+    r"<function\s*=\s*(?P<name>[A-Za-z_][\w.-]*)\s*>",
+    re.IGNORECASE,
+)
+PARAMETER_RE = re.compile(
+    r"<parameter\s*=\s*(?P<key>[A-Za-z_][\w.-]*)\s*>"
+    r"(?P<value>.*?)</parameter\s*>",
+    re.DOTALL | re.IGNORECASE,
+)
+DASHED_XML_TOOL_CALL_RE = re.compile(
+    r"<tool-call>\s*<name>\s*(?P<name>[A-Za-z_][\w.-]*)\s*</name>\s*"
+    r"<arguments>\s*(?P<arguments>.*?)\s*</arguments>\s*</tool-call>",
+    re.DOTALL | re.IGNORECASE,
+)
+NAMED_ARGUMENT_RE = re.compile(
+    r"<argument\s+name\s*=\s*(?P<quote>[\"'])"
+    r"(?P<key>[A-Za-z_][\w.-]*)(?P=quote)\s*>"
+    r"(?P<value>.*?)</argument\s*>",
+    re.DOTALL | re.IGNORECASE,
+)
+ELEMENT_ARGUMENT_RE = re.compile(
+    r"<(?P<key>[A-Za-z_][\w.-]*)\s*>(?P<value>.*?)</(?P=key)\s*>",
+    re.DOTALL | re.IGNORECASE,
+)
+SELF_CLOSING_TOOL_RE = re.compile(
+    r"<(?P<name>[A-Za-z_][\w.-]*)\b(?P<attributes>[^<>]*?)/\s*>",
+    re.DOTALL,
+)
+ATTRIBUTE_RE = re.compile(
+    r'''(?P<key>[A-Za-z_][\w.-]*)\s*=\s*(?:
+        "(?P<double>(?:\\.|[^"\\])*)"
+        |'(?P<single>(?:\\.|[^'\\])*)'
+    )''',
+    re.DOTALL | re.VERBOSE,
+)
+HTML_ENTITY_RE = re.compile(
+    r"&(?:#[0-9]+|#[xX][0-9A-Fa-f]+|[A-Za-z][A-Za-z0-9]+);"
+)
+ASSISTANT_CALLED_TOOL_RE = re.compile(
+    r"^\s*\[Assistant called tool (?P<name>[A-Za-z_][\w.-]*) "
+    r"with arguments (?P<arguments>\{.*\})\]\s*$",
+    re.DOTALL,
+)
+TEXTUAL_TOOL_CALL_RE = re.compile(
+    r"^[ \t]*(?P<name>[A-Za-z_][\w.-]*)[ \t]+"
+    r"(?:with|using)[ \t]+(?P<arguments>.+?)[ \t]*$",
+    re.MULTILINE | re.IGNORECASE,
+)
+FENCED_JSON_RE = re.compile(
+    r"^\s*```(?:json)?\s*(?P<payload>\{.*\})\s*```\s*$",
+    re.DOTALL | re.IGNORECASE,
+)
+TOOL_CALL_CLOSE_RE = re.compile(r"</tool_call>\s*$", re.IGNORECASE)
+XML_JSON_TOOL_CALL_CLOSE_RE = re.compile(
+    r"<xml>\s*\{.*?\}\s*</xml>\s*$",
+    re.DOTALL | re.IGNORECASE,
+)
+SELF_CLOSING_TOOL_AT_END_RE = re.compile(
+    r"<(?:tool\b|[A-Z][A-Za-z0-9_.-]*)\b[^<>]*/\s*>\s*(?:```)?\s*$",
+    re.DOTALL,
+)
+KNOWN_TEXTUAL_TOOL_NAMES = frozenset(
+    {
+        "agent",
+        "askuserquestion",
+        "bash",
+        "edit",
+        "enterplanmode",
+        "glob",
+        "grep",
+        "lsp",
+        "notebookedit",
+        "read",
+        "skill",
+        "task",
+        "todowrite",
+        "webfetch",
+        "websearch",
+        "write",
+    }
+)
+
+
+def _looks_like_tool_payload(payload: object) -> bool:
+    """Return whether a mapping has the OpenAI/Qwen tool-call shape."""
+    if not isinstance(payload, Mapping):
+        return False
+    function = payload.get("function")
+    if isinstance(function, Mapping):
+        return isinstance(function.get("name"), str) and bool(function.get("name"))
+    return isinstance(payload.get("name"), str) and bool(payload.get("name"))
+
+
+def _terminal_json_tool_payload(text: str) -> tuple[int, int, Mapping[str, Any]] | None:
+    """Recover a complete bare JSON tool call at the end of model output.
+
+    Small coder models sometimes obey the JSON schema but omit the surrounding
+    ``<tool_call>`` tags, occasionally after a short explanatory prefix.  The
+    normal extractor can parse a *pure* JSON response, but the generation
+    stopping criterion previously failed to stop there, allowing the model to
+    continue with prose and additional simulated calls.  Scan JSON-object starts
+    and accept only a terminal mapping with a tool-call shape.
+    """
+    candidate_text = text.rstrip()
+    decoder = json.JSONDecoder()
+    for start, char in enumerate(candidate_text):
+        if char != "{":
+            continue
+        try:
+            payload, consumed = decoder.raw_decode(candidate_text[start:])
+        except json.JSONDecodeError:
+            continue
+        end = start + consumed
+        if candidate_text[end:].strip():
+            continue
+        if _looks_like_tool_payload(payload):
+            return start, len(candidate_text), payload
+    return None
+
+
+def has_complete_tool_call(text: str) -> bool:
+    """Return true once generation has ended a supported tool-call form."""
+    fenced_json = FENCED_JSON_RE.fullmatch(text)
+    return bool(
+        TOOL_CALL_CLOSE_RE.search(text)
+        or XML_JSON_TOOL_CALL_CLOSE_RE.search(text)
+        or SELF_CLOSING_TOOL_AT_END_RE.search(text)
+        or (
+            fenced_json
+            and _looks_like_tool_payload(_mapping_literal(fenced_json.group("payload")))
+        )
+        or _terminal_json_tool_payload(text)
+        or any(
+            match.group("name").casefold() in KNOWN_TEXTUAL_TOOL_NAMES
+            for match in TEXTUAL_TOOL_CALL_RE.finditer(text)
+        )
+    )
+
+
+def _canonical_name(name: Any, allowed_names: set[str] | None) -> str | None:
+    if not isinstance(name, str) or not name:
+        return None
+    if not allowed_names:
+        return name
+    by_casefold = {candidate.casefold(): candidate for candidate in allowed_names}
+    normalized = name.casefold()
+    canonical = by_casefold.get(normalized)
+    if canonical is not None:
+        return canonical
+    # OpenClaude exposes the legacy Agent executor as Task. Accept both names
+    # in textual generations while returning the advertised catalog name.
+    alias = {"agent": "task", "task": "agent"}.get(normalized)
+    return by_casefold.get(alias) if alias else None
+
+
+def _coerce_value(value: str) -> Any:
+    value = _unescape_entities(value.strip())
+    try:
+        return json.loads(value)
+    except json.JSONDecodeError:
+        return value
+
+
+def _decode_attribute(value: str) -> str:
+    try:
+        value = json.loads(f'"{value}"')
+    except json.JSONDecodeError:
+        pass
+    return _unescape_entities(value)
+
+
+def _unescape_entities(value: str) -> str:
+    """Decode explicit entities without treating a URL's bare ``&`` as HTML.
+
+    ``html.unescape`` accepts legacy semicolon-less names such as ``&curren``.
+    That turns a query key like ``&current_weather`` into ``¤t_weather``.
+    XML entities are terminated with a semicolon, so only decode that form.
+    """
+    return HTML_ENTITY_RE.sub(lambda match: html.unescape(match.group(0)), value)
+
+
+def _attributes(raw: str) -> dict[str, str]:
+    values: dict[str, str] = {}
+    for match in ATTRIBUTE_RE.finditer(raw):
+        value = match.group("double")
+        if value is None:
+            value = match.group("single")
+        if value is not None:
+            values[match.group("key")] = _decode_attribute(value)
+    return values
+
+
+def _arguments(value: Any) -> dict[str, Any] | None:
+    if isinstance(value, Mapping):
+        return dict(value)
+    if not isinstance(value, str):
+        return None
+    parsed = _mapping_literal(_unescape_entities(value))
+    return dict(parsed) if isinstance(parsed, Mapping) else None
+
+
+def _mapping_literal(value: str) -> Mapping[str, Any] | None:
+    """Parse JSON or a Python-style mapping without evaluating expressions."""
+    try:
+        parsed = json.loads(value)
+    except json.JSONDecodeError:
+        try:
+            parsed = ast.literal_eval(value)
+        except (SyntaxError, ValueError):
+            return None
+    return parsed if isinstance(parsed, Mapping) else None
+
+
+def normalize_openai_tool_arguments(value: Any) -> dict[str, Any]:
+    """Return the mapping required by Qwen3's chat-template ``items`` filter.
+
+    OpenAI serializes function arguments as a JSON string, while Qwen3's
+    official template iterates them as a mapping when replaying tool history.
+    Accept both representations so a completed tool call can be followed by a
+    tool result without raising a template ``TypeError``.
+    """
+    parsed = _arguments(value)
+    return parsed if parsed is not None else {}
+
+
+def _openai_call(
+    name: Any,
+    arguments: Any,
+    allowed_names: set[str] | None,
+) -> dict[str, Any] | None:
+    canonical_name = _canonical_name(name, allowed_names)
+    if canonical_name is None:
+        return None
+    if isinstance(arguments, str):
+        parsed = _arguments(arguments)
+        arguments = parsed if parsed is not None else {}
+    if not isinstance(arguments, Mapping):
+        arguments = {}
+    return {
+        "id": f"call_{uuid.uuid4().hex[:24]}",
+        "type": "function",
+        "function": {
+            "name": canonical_name,
+            "arguments": json.dumps(
+                dict(arguments), ensure_ascii=False, separators=(",", ":")
+            ),
+        },
+    }
+
+
+def _payload_call(
+    payload: Any,
+    allowed_names: set[str] | None,
+) -> dict[str, Any] | None:
+    if not isinstance(payload, Mapping):
+        return None
+    function = payload.get("function")
+    if isinstance(function, Mapping):
+        return _openai_call(
+            function.get("name"),
+            function.get("arguments", {}),
+            allowed_names,
+        )
+    return _openai_call(
+        payload.get("name"), payload.get("arguments", {}), allowed_names
+    )
+
+
+def _xml_arguments(arguments: str) -> dict[str, Any]:
+    named = {
+        match.group("key"): _coerce_value(match.group("value"))
+        for match in NAMED_ARGUMENT_RE.finditer(arguments)
+    }
+    if named:
+        return named
+    return {
+        match.group("key"): _coerce_value(match.group("value"))
+        for match in ELEMENT_ARGUMENT_RE.finditer(arguments)
+    }
+
+
+def _textual_arguments(value: str) -> dict[str, Any] | None:
+    """Parse Qwen's compact ``tool with key=value`` representation."""
+    raw = value.strip().rstrip(";").strip()
+    mapping = _mapping_literal(raw)
+    if isinstance(mapping, Mapping):
+        return dict(mapping)
+
+    # Normalize optional whitespace around '=' before shlex handles quoted
+    # values containing spaces. No expressions are evaluated here.
+    raw = re.sub(
+        r"(?P<key>[A-Za-z_][\w.-]*)\s*=\s*",
+        r"\g<key>=",
+        raw,
+    )
+    try:
+        tokens = shlex.split(raw, posix=True)
+    except ValueError:
+        return None
+
+    arguments: dict[str, Any] = {}
+    for token in tokens:
+        if "=" not in token:
+            continue
+        key, item = token.split("=", 1)
+        if not re.fullmatch(r"[A-Za-z_][\w.-]*", key):
+            continue
+        arguments[key] = _coerce_value(item)
+    return arguments or None
+
+
+def extract_tool_call(
+    text: str,
+    allowed_names: set[str] | None = None,
+) -> tuple[dict[str, Any] | None, str]:
+    """Extract the first supported tool call for backward compatibility."""
+    calls, visible = extract_tool_calls(text, allowed_names)
+    return (calls[0] if calls else None), visible
+
+
+def extract_tool_calls(
+    text: str,
+    allowed_names: set[str] | None = None,
+) -> tuple[list[dict[str, Any]], str]:
+    """Extract all tool calls while accepting Qwen's common XML variations.
+
+    Matching calls deliberately clear visible content. Agent clients should
+    receive structured OpenAI calls rather than Markdown/XML renditions of the
+    same calls before they execute the tools.
+    """
+    candidates: list[tuple[int, int, dict[str, Any]]] = []
+
+    for match in JSON_TOOL_CALL_RE.finditer(text):
+        call = _payload_call(
+            _mapping_literal(match.group("payload")),
+            allowed_names,
+        )
+        if call:
+            candidates.append((match.start(), match.end(), call))
+
+    for match in XML_JSON_TOOL_CALL_RE.finditer(text):
+        call = _payload_call(
+            _mapping_literal(match.group("payload")),
+            allowed_names,
+        )
+        if call:
+            candidates.append((match.start(), match.end(), call))
+
+    for match in STANDARD_XML_TOOL_CALL_RE.finditer(text):
+        body = match.group("body")
+        function = FUNCTION_NAME_RE.search(body)
+        if function:
+            call = _openai_call(
+                function.group("name"),
+                {
+                    parameter.group("key"): _coerce_value(parameter.group("value"))
+                    for parameter in PARAMETER_RE.finditer(body)
+                },
+                allowed_names,
+            )
+            if call:
+                candidates.append((match.start(), match.end(), call))
+
+    for match in DASHED_XML_TOOL_CALL_RE.finditer(text):
+        call = _openai_call(
+            match.group("name"),
+            _xml_arguments(match.group("arguments")),
+            allowed_names,
+        )
+        if call:
+            candidates.append((match.start(), match.end(), call))
+
+    for match in SELF_CLOSING_TOOL_RE.finditer(text):
+        tag_name = match.group("name")
+        attributes = _attributes(match.group("attributes"))
+        if tag_name.casefold() == "tool":
+            tool_name = attributes.pop("name", None)
+            arguments = _arguments(
+                attributes.pop("arguments", attributes.pop("args", ""))
+            )
+            if arguments is None:
+                arguments = attributes
+        else:
+            tool_name = tag_name
+            arguments = attributes
+        call = _openai_call(tool_name, arguments, allowed_names)
+        if call:
+            candidates.append((match.start(), match.end(), call))
+
+    for match in TEXTUAL_TOOL_CALL_RE.finditer(text):
+        arguments = _textual_arguments(match.group("arguments"))
+        if arguments is None:
+            continue
+        call = _openai_call(match.group("name"), arguments, allowed_names)
+        if call:
+            candidates.append((match.start(), match.end(), call))
+
+    assistant_called = ASSISTANT_CALLED_TOOL_RE.fullmatch(text)
+    if assistant_called:
+        call = _openai_call(
+            assistant_called.group("name"),
+            _arguments(assistant_called.group("arguments")),
+            allowed_names,
+        )
+        if call:
+            candidates.append((assistant_called.start(), assistant_called.end(), call))
+
+    if not candidates:
+        fenced_json = FENCED_JSON_RE.fullmatch(text)
+        raw_json = fenced_json.group("payload") if fenced_json else text.strip()
+        call = _payload_call(_mapping_literal(raw_json), allowed_names)
+        if call:
+            candidates.append((0, len(text), call))
+
+    if not candidates:
+        terminal_json = _terminal_json_tool_payload(text)
+        if terminal_json is not None:
+            start, end, payload = terminal_json
+            call = _payload_call(payload, allowed_names)
+            if call:
+                candidates.append((start, end, call))
+
+    if not candidates:
+        return [], text
+
+    # Different parsers can recognize the same outer wrapper. Keep one result
+    # per source span while preserving the order produced by the model.
+    unique: list[dict[str, Any]] = []
+    seen_spans: set[tuple[int, int]] = set()
+    for start, end, call in sorted(candidates, key=lambda item: (item[0], item[1])):
+        span = (start, end)
+        if span in seen_spans:
+            continue
+        seen_spans.add(span)
+        unique.append(call)
+    return unique, ""