"""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"|" 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"(?Pbash|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"<(?Pavailable-deferred-tools|system-reminder)\b[^>]*>.*?", 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 ( "]*>.*?", "", 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: {"name":"TOOL_NAME","arguments":{"key":"value"}}', "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: " + json.dumps( { "name": first_example[0], "arguments": first_example[1], }, ensure_ascii=False, separators=(",", ":"), ) + "" ) 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