Harden _enables_code_execution_tool against non-dict tool entries

Verify each entry and its function field are dicts before reading the tool
name, matching the defensive isinstance checks the payload.tools validation
already uses. The callers pass resolved internal tool specs, so this is
robustness rather than a reachable bug.

studio/backend/core/inference/llama_cpp.py

@@ -8446,6 +8446,7 @@ class LlamaCppBackend:
         seed: Optional[int] = None,
         disable_parallel_tool_use: bool = False,
         confirm_tool_calls: bool = False,
+        confirm_code_execution: bool = False,
         bypass_permissions: bool = False,
     ) -> Generator[dict, None, None]:
         """
@@ -8456,7 +8457,11 @@ class LlamaCppBackend:
           {"type": "content", "text": "token"}            -- streamed content tokens (cumulative)
           {"type": "reasoning", "text": "token"}          -- streamed reasoning tokens (cumulative)
         """
-        from core.inference.tools import build_rag_autoinject, execute_tool
+        from core.inference.tools import (
+            CODE_EXECUTION_TOOL_NAMES,
+            build_rag_autoinject,
+            execute_tool,
+        )
 
         if not self.is_loaded:
             raise RuntimeError("llama-server is not loaded")
@@ -8813,9 +8818,17 @@ class LlamaCppBackend:
                                             in provisional_started_tool_calls.values()
                                         )
                                         # Later parallel cards only reconcile when parallel use is enabled.
+                                        # Suppress the early card whenever this call will be
+                                        # gated for confirmation, so a python/terminal call under
+                                        # confirm_code_execution never flashes "running" before the
+                                        # approve/deny prompt.
                                         _confirm_gated = (
-                                            confirm_tool_calls and not bypass_permissions
+                                            confirm_tool_calls
-                                        )
+                                            or (
+                                                confirm_code_execution
+                                                and current_name in CODE_EXECUTION_TOOL_NAMES
+                                            )
+                                        ) and not bypass_permissions
                                         # Keep small-argument tools on the normal path.
                                         _args_len = len(
                                             tool_calls_acc[idx]["function"].get("arguments", "")
@@ -9373,7 +9386,16 @@ class LlamaCppBackend:
 
                     # Bypass wins over the confirm gate at the loop level too,
                     # so a direct internal caller with both flags never prompts.
-                    needs_confirm = bool(confirm_tool_calls) and not bypass_permissions
+                    # confirm_code_execution narrows the gate to python/terminal so
+                    # a code-execution call pauses for approval even when
+                    # confirm_tool_calls is off (search/render tools stay instant).
+                    needs_confirm = (
+                        bool(confirm_tool_calls)
+                        or (
+                            bool(confirm_code_execution)
+                            and decision.tool_name in CODE_EXECUTION_TOOL_NAMES
+                        )
+                    ) and not bypass_permissions
                     approval_id = new_approval_id() if needs_confirm else ""
                     decision_slot = (
                         begin_tool_decision(session_id, approval_id) if needs_confirm else None

studio/backend/core/inference/orchestrator.py

@@ -1225,6 +1225,7 @@ class InferenceOrchestrator:
         session_id: Optional[str] = None,
         rag_scope: Optional[dict] = None,
         confirm_tool_calls: bool = False,
+        confirm_code_execution: bool = False,
         bypass_permissions: bool = False,
         use_adapter: Optional[Union[bool, str]] = None,
         stats_holder: Optional[dict] = None,
@@ -1291,6 +1292,7 @@ class InferenceOrchestrator:
             session_id = session_id,
             rag_scope = rag_scope,
             confirm_tool_calls = confirm_tool_calls,
+            confirm_code_execution = confirm_code_execution,
             bypass_permissions = bypass_permissions,
         )
 

studio/backend/core/inference/safetensors_agentic.py

@@ -426,6 +426,7 @@ def run_safetensors_tool_loop(
     session_id: Optional[str] = None,
     rag_scope: Optional[dict] = None,
     confirm_tool_calls: bool = False,
+    confirm_code_execution: bool = False,
     bypass_permissions: bool = False,
 ) -> Generator[dict, None, None]:
     """Drive an agentic tool loop on top of a cumulative-text generator.
@@ -453,7 +454,7 @@ def run_safetensors_tool_loop(
     conversation = list(messages)
 
     # Forced first-pass RAG (mirrors the GGUF loop) so doc Qs don't lose to web_search.
-    from core.inference.tools import build_rag_autoinject
+    from core.inference.tools import CODE_EXECUTION_TOOL_NAMES, build_rag_autoinject
 
     _auto = None if confirm_tool_calls else build_rag_autoinject(conversation, rag_scope)
     if _auto:
@@ -1056,7 +1057,15 @@ def run_safetensors_tool_loop(
 
             # Bypass wins over the confirm gate at the loop level too, so a
             # direct internal caller passing both flags never prompts.
-            needs_confirm = bool(confirm_tool_calls) and not bypass_permissions
+            # confirm_code_execution narrows the gate to python/terminal so a
+            # code-execution call pauses for approval even when confirm_tool_calls
+            # is off (search/render tools still run without a prompt).
+            needs_confirm = (
+                bool(confirm_tool_calls)
+                or (
+                    bool(confirm_code_execution) and decision.tool_name in CODE_EXECUTION_TOOL_NAMES
+                )
+            ) and not bypass_permissions
             approval_id = new_approval_id() if needs_confirm else ""
             decision_slot = begin_tool_decision(session_id, approval_id) if needs_confirm else None
             start_event = decision.tool_start_event()

studio/backend/core/inference/tools.py

@@ -832,6 +832,12 @@ ALL_TOOLS = [
     SEARCH_KNOWLEDGE_BASE_TOOL,
 ]
 
+# Built-in tools that run arbitrary model-authored code/commands (through the
+# sandbox unless bypassed). The confirmation gate can target just these
+# (confirm_code_execution) so a code-execution call pauses for approval while
+# search/render tools stay instant.
+CODE_EXECUTION_TOOL_NAMES = frozenset({"python", "terminal"})
+
 
 # OpenAI's function.name regex ^[a-zA-Z0-9_-]{1,64}$, enforced before streaming.
 # MCP tool names with '.', '/', spaces, etc. would 400 the whole request, so we

studio/backend/models/inference.py

@@ -773,6 +773,10 @@ class ChatCompletionRequest(BaseModel):
         None,
         description = "[x-unsloth] When true, pause before each tool call and wait for the user to allow/deny it via POST /api/inference/tool-confirm.",
     )
+    confirm_code_execution: Optional[bool] = Field(
+        None,
+        description = "[x-unsloth] When true, pause only before local code-execution tool calls (python/terminal) and wait for the user to allow/deny each via POST /api/inference/tool-confirm; other tools (web_search, render_html, ...) still run without a prompt. Supported on the OpenAI-compatible local endpoints (/v1/chat/completions, /v1/responses). It is ignored for external providers (their hosted code runs in the provider's sandbox, not locally); on Anthropic /v1/messages it is rejected when a local code-execution tool is selected, since that path does not wire the confirmation prompt. Independent of confirm_tool_calls; requires stream=true when a local code-execution tool is enabled; bypass_permissions still takes precedence.",
+    )
     bypass_permissions: Optional[bool] = Field(
         False,
         description = "[x-unsloth] Bypass Permissions: when true, skip the tool-call confirmation gate AND disable the python/terminal execution sandbox (safety checks, command blocklist, resource limits). Secret env vars are still stripped. Takes precedence over confirm_tool_calls.",

studio/backend/routes/inference.py

@@ -1763,6 +1763,34 @@ async def _select_request_tools(
     return tools
 
 
+def _enables_code_execution_tool(tools: list[dict]) -> bool:
+    """True when a resolved tool list includes a local code-execution tool
+    (python/terminal). ``confirm_code_execution`` only gates those, so the
+    streaming requirement is scoped to requests that actually expose one."""
+    from core.inference.tools import CODE_EXECUTION_TOOL_NAMES
+    return any(
+        isinstance(t, dict)
+        and isinstance(t.get("function"), dict)
+        and t["function"].get("name") in CODE_EXECUTION_TOOL_NAMES
+        for t in (tools or [])
+    )
+
+
+def _payload_may_enable_code_execution(payload) -> bool:
+    """True when a request could resolve a local code-execution tool before the
+    tool list is built (used by the pre-switch guard). Mirrors
+    ``_select_request_tools``: built-ins are off unless the tool loop is enabled,
+    and an explicit ``enabled_tools`` filter must then list python/terminal
+    (MCP/client tools are never code execution)."""
+    from core.inference.tools import CODE_EXECUTION_TOOL_NAMES
+
+    if not _effective_enable_tools(payload):
+        return False
+    if payload.enabled_tools is not None:
+        return bool(set(payload.enabled_tools) & CODE_EXECUTION_TOOL_NAMES)
+    return True
+
+
 def _apply_rag_nudge(nudge: str, tools: list[dict], *, rag_scope) -> str:
     """Append the RAG grounding nudge to ``nudge`` when the knowledge-base tool
     is active (search_knowledge_base present and a retrieval scope is set). The
@@ -5673,6 +5701,9 @@ async def openai_chat_completions(
                     param = "confirm_tool_calls",
                 ),
             )
+        # confirm_code_execution guards only local python/terminal; an external
+        # provider runs any hosted code_execution in its own sandbox, so the flag
+        # simply does not apply here and is ignored (documented on the field).
         if _wants_multiple_choices(payload):
             _raise_unsupported_n("external provider chat completions")
         return await _proxy_to_external_provider(payload, request, current_subject)
@@ -5749,6 +5780,28 @@ async def openai_chat_completions(
                     param = "confirm_tool_calls",
                 ),
             )
+        # Same pre-switch guard for confirm_code_execution, but scoped to requests
+        # that could actually run a local code-execution tool: the gate only
+        # applies to python/terminal, so a non-code tool request (e.g. web_search)
+        # must not be rejected, and a code-execution one must not evict the
+        # resident model only to 400 after the swap.
+        if (
+            payload.confirm_code_execution
+            and not payload.bypass_permissions
+            and not payload.stream
+            and payload.max_tool_calls_per_message != 0
+            and _payload_may_enable_code_execution(payload)
+        ):
+            raise HTTPException(
+                status_code = 400,
+                detail = openai_error_body(
+                    "confirm_code_execution requires stream=true when a "
+                    "code-execution tool (python/terminal) is enabled.",
+                    status = 400,
+                    code = "invalid_request_error",
+                    param = "confirm_code_execution",
+                ),
+            )
         # Reject a malformed tool_choice forcing object before the switch: a
         # {"type": "function", "function": {}} with no name would otherwise be
         # forwarded to llama-server and rejected only after the model swapped.
@@ -6221,6 +6274,23 @@ async def openai_chat_completions(
                         param = "confirm_tool_calls",
                     ),
                 )
+            if (
+                payload.confirm_code_execution
+                and not payload.bypass_permissions
+                and not payload.stream
+                and payload.max_tool_calls_per_message != 0
+                and _enables_code_execution_tool(tools_to_use)
+            ):
+                raise _reject(
+                    400,
+                    openai_error_body(
+                        "confirm_code_execution requires stream=true when a "
+                        "code-execution tool (python/terminal) is enabled.",
+                        status = 400,
+                        code = "invalid_request_error",
+                        param = "confirm_code_execution",
+                    ),
+                )
             if _wants_multiple_choices(payload):
                 raise _reject_unsupported_n("GGUF tool chat completions")
             # ── Tool-use system prompt nudge ──────────────────────
@@ -6289,6 +6359,8 @@ async def openai_chat_completions(
                     # never prompt while bypassing.
                     confirm_tool_calls = bool(payload.confirm_tool_calls)
                     and not bool(payload.bypass_permissions),
+                    confirm_code_execution = bool(payload.confirm_code_execution)
+                    and not bool(payload.bypass_permissions),
                     bypass_permissions = bool(payload.bypass_permissions),
                 )
 
@@ -6945,6 +7017,22 @@ async def openai_chat_completions(
                     param = "confirm_tool_calls",
                 ),
             )
+        if (
+            payload.confirm_code_execution
+            and not payload.bypass_permissions
+            and not payload.stream
+            and _enables_code_execution_tool(_sf_tools_to_use)
+        ):
+            raise _reject(
+                400,
+                openai_error_body(
+                    "confirm_code_execution requires stream=true when a "
+                    "code-execution tool (python/terminal) is enabled.",
+                    status = 400,
+                    code = "invalid_request_error",
+                    param = "confirm_code_execution",
+                ),
+            )
         _sf_nudge = _build_tool_action_nudge(
             tools = _sf_tools_to_use,
             model_name = model_name,
@@ -7014,6 +7102,8 @@ async def openai_chat_completions(
                 # never prompt while bypassing.
                 confirm_tool_calls = bool(payload.confirm_tool_calls)
                 and not bool(payload.bypass_permissions),
+                confirm_code_execution = bool(payload.confirm_code_execution)
+                and not bool(payload.bypass_permissions),
                 bypass_permissions = bool(payload.bypass_permissions),
                 use_adapter = payload.use_adapter,
                 stats_holder = _sf_stats_holder,
@@ -10140,6 +10230,11 @@ async def anthropic_messages(
             ),
         )
 
+    # confirm_code_execution is handled below inside the server-tool branch (it is
+    # rejected when a local python/terminal tool is actually selected, mirroring
+    # confirm_tool_calls), so nothing to do pre-switch here: a non-code request is
+    # unaffected and a disabled request never enters that branch.
+
     # require_vision rejects a swap to a text-only target before it runs, so an
     # image request can't evict the resident vision model only to hit the vision
     # guard (_normalize_anthropic_openai_images) below after the load.
@@ -10340,6 +10435,36 @@ async def anthropic_messages(
             payload.enabled_tools,
         )
 
+        # confirm_code_execution guards local python/terminal execution. On this
+        # path a Studio tool alias like {"type":"python"} maps to the local tool
+        # loop and runs code on this host -- but the Anthropic Messages SSE
+        # translation does not wire the confirmation prompt (which is why
+        # confirm_tool_calls is rejected above). Silently ignoring the flag would
+        # run python/terminal without the promised prompt, so reject it when a
+        # local code-execution tool is actually selected. A non-code selection
+        # (e.g. web_search) is unaffected, and bypass_permissions suppresses the
+        # gate. Gated on server_tools above, so a disabled request never reaches
+        # here.
+        if (
+            bool(getattr(payload, "confirm_code_execution", False))
+            and not bool(getattr(payload, "bypass_permissions", False))
+            and _enables_code_execution_tool(openai_tools)
+        ):
+            api_monitor.fail(
+                monitor_id,
+                "confirm_code_execution is not supported for Anthropic Messages server tools.",
+            )
+            raise HTTPException(
+                status_code = 400,
+                detail = anthropic_error_body(
+                    "confirm_code_execution is not supported for Anthropic Messages "
+                    "server tools; it only guards local python/terminal execution on "
+                    "the OpenAI-compatible endpoints (/v1/chat/completions).",
+                    status = 400,
+                    err_type = "invalid_request_error",
+                ),
+            )
+
         # Build tool-use system prompt nudge (same logic as /chat/completions)
         _nudge = _build_tool_action_nudge(
             tools = openai_tools,

studio/backend/tests/test_anthropic_messages.py

@@ -1718,6 +1718,35 @@ class TestAnthropicMessagesToolRouting:
         assert "confirm_tool_calls is not supported" in exc.value.detail["error"]["message"]
         assert backend.calls == []
 
+    def test_confirm_code_execution_rejected_for_code_server_tools(self, monkeypatch):
+        # A Studio {"type":"python"} alias runs the local python executor via the
+        # tool loop, but this path does not wire the confirmation prompt (like
+        # confirm_tool_calls above). Ignoring the flag would run code without the
+        # prompt, so it is rejected when a local code-execution tool is selected.
+        backend = _mock_backend(monkeypatch)
+        payload = _basic_payload(
+            confirm_code_execution = True,
+            tools = [{"type": "python", "name": "python"}],
+        )
+
+        with pytest.raises(HTTPException) as exc:
+            _drive(anthropic_messages(payload, request = None, current_subject = "t"))
+        assert exc.value.status_code == 400
+        assert "confirm_code_execution is not supported" in exc.value.detail["error"]["message"]
+        assert backend.calls == []
+
+    def test_confirm_code_execution_ignored_for_non_code_server_tools(self, monkeypatch):
+        # web_search is not code execution, so the flag does not apply and the
+        # request proceeds normally rather than being rejected.
+        backend = _mock_backend(monkeypatch)
+        payload = _basic_payload(
+            confirm_code_execution = True,
+            tools = [{"type": "web_search_20250305", "name": "web_search"}],
+        )
+
+        _drive(anthropic_messages(payload, request = None, current_subject = "t"))
+        assert backend.calls[0][0] == "tools"
+
     def test_per_request_enable_tools_false_blocks_server_tool_alias(self, monkeypatch):
         backend = _mock_backend(monkeypatch)
         payload = _basic_payload(

studio/backend/tests/test_confirm_code_execution.py

@@ -0,0 +1,234 @@
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
+
+"""Tests for ``confirm_code_execution``: a narrower confirmation gate that
+pauses only before local code-execution tools (python/terminal) while other
+tools (web_search, render_html, ...) still run without a prompt.
+
+These drive the real ``run_safetensors_tool_loop`` with hand-crafted fake
+generators (no model), mirroring ``test_tool_confirm_loop.py``, and cover the
+scoping predicate the route layer uses to require streaming.
+"""
+
+import pytest
+
+from core.inference.safetensors_agentic import run_safetensors_tool_loop
+from state import tool_approvals
+from state.tool_approvals import resolve_tool_decision
+
+_SESSION = "code-exec-session"
+
+_TOOLS = [
+    {"type": "function", "function": {"name": "python"}},
+    {"type": "function", "function": {"name": "terminal"}},
+    {"type": "function", "function": {"name": "web_search"}},
+]
+
+
+@pytest.fixture(autouse = True)
+def _clear_pending():
+    with tool_approvals._lock:
+        tool_approvals._pending.clear()
+    yield
+    with tool_approvals._lock:
+        tool_approvals._pending.clear()
+
+
+class _FakeExecuteTool:
+    def __init__(self):
+        self.calls = []
+
+    def __call__(
+        self,
+        name,
+        arguments,
+        *,
+        cancel_event = None,
+        timeout = None,
+        session_id = None,
+        rag_scope = None,
+        disable_sandbox = False,
+    ):
+        self.calls.append((name, arguments))
+        return f"RESULT[{name}]"
+
+
+def _tool_call(name, args_json):
+    return f'<tool_call>{{"name": "{name}", "arguments": {args_json}}}</tool_call>'
+
+
+def _multi_turn(turns):
+    turn_iter = iter(turns)
+
+    def _gen(_messages):
+        try:
+            yield next(turn_iter)
+        except StopIteration:
+            return
+
+    return _gen
+
+
+def _drive(turns, decisions, **loop_kwargs):
+    """Run the loop, resolving each gated tool_start with the next decision.
+
+    Non-gated calls (awaiting_confirmation False) execute without consuming a
+    decision. Returns (events, execute_calls).
+    """
+    decision_iter = iter(decisions)
+    exec_fn = _FakeExecuteTool()
+    gen = run_safetensors_tool_loop(
+        single_turn = _multi_turn(turns),
+        messages = [{"role": "user", "content": "hi"}],
+        tools = _TOOLS,
+        execute_tool = exec_fn,
+        session_id = _SESSION,
+        **loop_kwargs,
+    )
+    events = []
+    for ev in gen:
+        events.append(ev)
+        if ev["type"] == "tool_start" and ev.get("awaiting_confirmation"):
+            resolve_tool_decision(ev["approval_id"], next(decision_iter), session_id = _SESSION)
+    return events, exec_fn.calls
+
+
+def _starts(events):
+    return [e for e in events if e["type"] == "tool_start"]
+
+
+def _ends(events):
+    return [e for e in events if e["type"] == "tool_end"]
+
+
+# ── confirm_code_execution gates only python/terminal ────────────────────────
+
+
+def test_python_call_is_gated_and_executes_on_allow():
+    events, calls = _drive(
+        [_tool_call("python", '{"code": "print(1)"}'), "done"],
+        ["allow"],
+        confirm_code_execution = True,
+    )
+    starts = _starts(events)
+    assert len(starts) == 1
+    assert starts[0]["awaiting_confirmation"] is True
+    assert starts[0]["approval_id"]
+    assert calls == [("python", {"code": "print(1)"})]
+    assert _ends(events)[0]["result"] == "RESULT[python]"
+
+
+def test_terminal_call_is_gated_and_skipped_on_deny():
+    events, calls = _drive(
+        [_tool_call("terminal", '{"command": "ls"}'), "done"],
+        ["deny"],
+        confirm_code_execution = True,
+    )
+    starts = _starts(events)
+    assert len(starts) == 1
+    assert starts[0]["awaiting_confirmation"] is True
+    # Denied: the tool never runs.
+    assert calls == []
+
+
+def test_web_search_is_not_gated_by_confirm_code_execution():
+    # No decision is supplied: a gated call would block waiting for one.
+    events, calls = _drive(
+        [_tool_call("web_search", '{"query": "cats"}'), "done"],
+        [],
+        confirm_code_execution = True,
+    )
+    starts = _starts(events)
+    assert len(starts) == 1
+    assert starts[0]["awaiting_confirmation"] is False
+    assert not starts[0]["approval_id"]
+    assert calls == [("web_search", {"query": "cats"})]
+
+
+def test_bypass_permissions_overrides_confirm_code_execution():
+    events, calls = _drive(
+        [_tool_call("python", '{"code": "print(1)"}'), "done"],
+        [],
+        confirm_code_execution = True,
+        bypass_permissions = True,
+    )
+    starts = _starts(events)
+    assert starts[0]["awaiting_confirmation"] is False
+    assert calls == [("python", {"code": "print(1)"})]
+
+
+def test_default_off_does_not_gate_code_execution():
+    events, calls = _drive(
+        [_tool_call("python", '{"code": "print(1)"}'), "done"],
+        [],
+        # Neither flag set: unchanged legacy behavior, python runs immediately.
+    )
+    starts = _starts(events)
+    assert starts[0]["awaiting_confirmation"] is False
+    assert calls == [("python", {"code": "print(1)"})]
+
+
+def test_confirm_tool_calls_still_gates_every_tool():
+    # confirm_tool_calls is the broad gate; web_search is prompted under it even
+    # though confirm_code_execution would not touch it.
+    events, calls = _drive(
+        [_tool_call("web_search", '{"query": "cats"}'), "done"],
+        ["allow"],
+        confirm_tool_calls = True,
+    )
+    starts = _starts(events)
+    assert starts[0]["awaiting_confirmation"] is True
+    assert calls == [("web_search", {"query": "cats"})]
+
+
+# ── scoping predicates used by the route streaming requirement ───────────────
+
+
+def _spec(name):
+    return {"type": "function", "function": {"name": name}}
+
+
+def test_enables_code_execution_tool_predicate():
+    from routes.inference import _enables_code_execution_tool
+
+    assert _enables_code_execution_tool([_spec("python")]) is True
+    assert _enables_code_execution_tool([_spec("terminal")]) is True
+    assert _enables_code_execution_tool([_spec("web_search"), _spec("python")]) is True
+    # A non-code tool list must not trip the streaming requirement.
+    assert _enables_code_execution_tool([_spec("web_search"), _spec("render_html")]) is False
+    assert _enables_code_execution_tool([]) is False
+    assert _enables_code_execution_tool(None) is False
+
+
+def test_payload_may_enable_code_execution_predicate(monkeypatch):
+    import routes.inference as inf
+
+    monkeypatch.setattr("state.tool_policy.get_tool_policy", lambda: None)
+
+    class _P:
+        def __init__(
+            self,
+            enable_tools = None,
+            enabled_tools = None,
+        ):
+            self.enable_tools = enable_tools
+            self.enabled_tools = enabled_tools
+
+    # Built-ins off -> never code execution, even if enabled_tools lists python.
+    assert inf._payload_may_enable_code_execution(_P(enable_tools = None)) is False
+    assert (
+        inf._payload_may_enable_code_execution(_P(enable_tools = False, enabled_tools = ["python"]))
+        is False
+    )
+    # Built-ins on, explicit filter without a code tool -> not code execution (the fix).
+    assert (
+        inf._payload_may_enable_code_execution(_P(enable_tools = True, enabled_tools = ["web_search"]))
+        is False
+    )
+    # Built-ins on, code tool in the filter -> code execution.
+    assert (
+        inf._payload_may_enable_code_execution(_P(enable_tools = True, enabled_tools = ["terminal"]))
+        is True
+    )
+    # Built-ins on, no filter -> all built-ins including python/terminal.
+    assert inf._payload_may_enable_code_execution(_P(enable_tools = True)) is True

studio/backend/tests/test_openai_tool_passthrough.py

@@ -462,6 +462,39 @@ class TestChatCompletionRequestToolFields:
         assert body["error"]["param"] == "confirm_tool_calls"
         assert "only supported for local streaming tools" in body["error"]["message"]
 
+    def test_confirm_code_execution_ignored_for_provider_tools(self, monkeypatch):
+        # confirm_code_execution guards only local python/terminal. An external
+        # provider runs any hosted code_execution in its own sandbox, so the flag
+        # is ignored (not rejected) -- even when provider code_execution is enabled.
+        import routes.inference as inference_route
+
+        called = {"proxied": False}
+
+        async def _fake_proxy(payload, request, current_subject):
+            called["proxied"] = True
+            return {"ok": True}
+
+        monkeypatch.setattr(inference_route, "_proxy_to_external_provider", _fake_proxy)
+
+        class _UnusedBackend:
+            is_loaded = False
+
+        client = self._v1_client(monkeypatch, _UnusedBackend())
+        resp = client.post(
+            "/v1/chat/completions",
+            json = {
+                "messages": [{"role": "user", "content": "hi"}],
+                "provider_type": "openai",
+                "external_model": "gpt-4.1",
+                "enable_tools": True,
+                "enabled_tools": ["code_execution"],
+                "confirm_code_execution": True,
+            },
+        )
+
+        assert resp.status_code != 400
+        assert called["proxied"] is True
+
     def test_logprobs_rejected_until_supported(self, monkeypatch):
         class _UnusedBackend:
             is_loaded = False
@@ -612,6 +645,51 @@ class TestChatCompletionRequestToolFields:
         assert "confirm_tool_calls requires stream=true" in entry["error"]
         assert monitor.active_count() == 0
 
+    def test_confirm_code_execution_requires_streaming_for_safetensors_tools(self, monkeypatch):
+        import routes.inference as inference_route
+
+        class _NoGGUFBackend:
+            is_loaded = False
+            supports_tools = False
+
+        class _InferenceBackend:
+            active_model_name = "test-model"
+            models = {"test-model": {"chat_template_info": {"template": "chatml"}}}
+
+            def generate_chat_completion_with_tools(self, **kwargs):
+                raise AssertionError("tool loop should be rejected before starting")
+
+            def generate_chat_completion(self, **kwargs):
+                raise AssertionError("plain path should not be used")
+
+        monkeypatch.setattr(
+            inference_route,
+            "_detect_safetensors_features",
+            lambda backend, chat_template: {"supports_tools": True},
+        )
+        monitor = ApiMonitor(max_entries = 3)
+        monkeypatch.setattr(inference_route, "api_monitor", monitor)
+        client = self._v1_client(monkeypatch, _NoGGUFBackend(), _InferenceBackend())
+        resp = client.post(
+            "/v1/chat/completions",
+            json = {
+                "messages": [{"role": "user", "content": "run code"}],
+                "enable_tools": True,
+                "enabled_tools": ["python"],
+                "confirm_code_execution": True,
+                "stream": False,
+            },
+        )
+
+        assert resp.status_code == 400
+        body = resp.json()
+        assert body["error"]["param"] == "confirm_code_execution"
+        assert "requires stream=true" in body["error"]["message"]
+        [entry] = monitor.snapshot()
+        assert entry["status"] == "error"
+        assert "confirm_code_execution requires stream=true" in entry["error"]
+        assert monitor.active_count() == 0
+
     def test_multiturn_tool_loop_messages(self):
         req = ChatCompletionRequest(
             messages = [
@@ -1493,6 +1571,105 @@ class TestGgufVisionToolRouting:
         assert "confirm_tool_calls requires stream=true" in entry["error"]
         assert monitor.active_count() == 0
 
+    def test_confirm_code_execution_requires_streaming_for_gguf_tools(self, monkeypatch):
+        import routes.inference as inf_mod
+
+        def _plain(**kwargs):
+            raise AssertionError("plain GGUF path should not be used")
+
+        def _tools(**kwargs):
+            raise AssertionError("tool loop should be rejected before starting")
+
+        backend = SimpleNamespace(
+            is_loaded = True,
+            is_vision = False,
+            supports_tools = True,
+            model_identifier = "test-gguf",
+            context_length = 4096,
+            generate_chat_completion = _plain,
+            generate_chat_completion_with_tools = _tools,
+        )
+        monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
+        monitor = ApiMonitor(max_entries = 3)
+        monkeypatch.setattr(inf_mod, "api_monitor", monitor)
+
+        payload = ChatCompletionRequest(
+            model = "default",
+            enable_tools = True,
+            enabled_tools = ["python"],
+            confirm_code_execution = True,
+            stream = False,
+            messages = [{"role": "user", "content": "run code"}],
+        )
+
+        with pytest.raises(HTTPException) as exc:
+            self._drive(
+                openai_chat_completions(
+                    payload,
+                    request = self._Request(),
+                    current_subject = "test",
+                )
+            )
+        assert exc.value.status_code == 400
+        assert exc.value.detail["error"]["param"] == "confirm_code_execution"
+        assert "requires stream=true" in exc.value.detail["error"]["message"]
+        [entry] = monitor.snapshot()
+        assert entry["status"] == "error"
+        assert "confirm_code_execution requires stream=true" in entry["error"]
+        assert monitor.active_count() == 0
+
+    def test_confirm_code_execution_budget_zero_not_rejected_gguf(self, monkeypatch):
+        # max_tool_calls_per_message=0 disables tool execution (max_tool_iterations=0),
+        # so no code-execution tool can run and the stream requirement must not reject.
+        import routes.inference as inf_mod
+
+        reached = {"tools": False}
+
+        def _plain(**kwargs):
+            raise AssertionError("plain GGUF path should not be used")
+
+        def _tools(**kwargs):
+            # Reaching here proves the request cleared both the pre-switch and the
+            # per-handler confirm_code_execution guards instead of being rejected.
+            reached["tools"] = True
+            raise RuntimeError("reached tool backend")
+
+        backend = SimpleNamespace(
+            is_loaded = True,
+            is_vision = False,
+            supports_tools = True,
+            model_identifier = "test-gguf",
+            context_length = 4096,
+            generate_chat_completion = _plain,
+            generate_chat_completion_with_tools = _tools,
+            _maybe_recover_from_mtp_crash = lambda e: None,
+        )
+        monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
+        monkeypatch.setattr(inf_mod, "api_monitor", ApiMonitor(max_entries = 3))
+
+        payload = ChatCompletionRequest(
+            model = "default",
+            enable_tools = True,
+            enabled_tools = ["python"],
+            confirm_code_execution = True,
+            stream = False,
+            max_tool_calls_per_message = 0,
+            messages = [{"role": "user", "content": "run code"}],
+        )
+
+        with pytest.raises(HTTPException) as exc:
+            self._drive(
+                openai_chat_completions(
+                    payload,
+                    request = self._Request(),
+                    current_subject = "test",
+                )
+            )
+        # Reaching the tool backend (and not a confirm_code_execution 400) proves the
+        # budget-zero request was not rejected by the stream requirement.
+        assert reached["tools"] is True
+        assert exc.value.status_code != 400
+
     def test_standard_gguf_stream_splits_reasoning_content(self, monkeypatch):
         def _generate(**_kwargs):
             yield "<thi"