Spaces:

Erinaldorodrigues
/

qwen-coder-api

Running on Zero

File size: 16,961 Bytes

"""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"<(?P<tag>tool_call|function_call)>\s*(?P<payload>\{.*?\})\s*</(?P=tag)>",
    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"<(?P<tag>tool_call|function_call)>\s*(?P<body>.*?)\s*</(?P=tag)>",
    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|function_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,
    allowed_names: set[str] | None = None,
) -> bool:
    """Return true once generation ended a valid supported tool-call form.

    When ``allowed_names`` is supplied, a syntactically complete hallucinated
    call to an unadvertised function must *not* stop generation. This matters
    for OpenClaude because the final parser rejects unadvertised names.
    """
    if allowed_names is not None:
        calls, _ = extract_tool_calls(text, allowed_names)
        return bool(calls)

    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 Qwen2.5'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 recover_forced_tool_call(text: str, tool_name: str) -> dict[str, Any] | None:
    """Recover argument-only JSON when exactly one tool is mandated.

    Some OpenAI-compatible coder models occasionally emit only the function
    argument object when the caller has already forced a single tool. The
    normal parser correctly refuses to guess a function name from that object.
    In the *single forced-tool* case, however, the name is unambiguous and the
    structured OpenAI call can be reconstructed safely without executing or
    evaluating arbitrary text.
    """
    raw = text.strip()
    fenced = FENCED_JSON_RE.fullmatch(raw)
    if fenced:
        raw = fenced.group("payload")

    payload = _mapping_literal(raw)
    if not isinstance(payload, Mapping):
        return None
    if _looks_like_tool_payload(payload):
        return None

    arguments: Mapping[str, Any] = payload
    nested_arguments = payload.get("arguments")
    if len(payload) == 1 and isinstance(nested_arguments, Mapping):
        arguments = nested_arguments

    return _openai_call(tool_name, arguments, {tool_name})


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, ""