scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-12 03:48:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 57
litellm/tests/unit/integrations/code_interpreter_interception/test_handler.py
devin-ai-integration[bot] 13928bdabe
fix(websearch_interception): drop stream_options when forcing stream=False and replay it on the final stream (#45433) 
* fix(websearch_interception): drop stream_options when forcing stream=False and replay it on the final stream

* test(websearch_interception): type the conversion helpers and drive the replay test through mock_response

* test(websearch_interception): pin the replay test's values as Final and record follow-ups immutably

* fix(proxy): strip every agentic-loop control key from client bodies, the web search keys included

* fix(interception): move stream_options off forced non-stream calls for code interpreter and headroom, and keep adapters from replaying nested calls

* test: annotate new interception test locals with Final

* fix(responses): drop params a deployment hook removed on sync Responses calls

* test: drive the sync Responses stream_options regression through a mock transport and real Azure config

* test: type the agentic loop recorder and capture the sent request on an object

* test(integration): cover interception stream_options on chat, responses, messages and the SDK

* test(integration): assert the follow-up error text and release a held burst on failure

* test(integration): pin the follow-up failure contract and wait for the replacement worker before teardown

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
2026-10-09 22:16:02 -07:00

1450 lines
53 KiB
Python
Raw Blame History

"""
Unit tests for CodeInterpreterInterceptionLogger.
All sandbox dependencies are injected: a FakeSandbox stands in for the real e2b
config and records how it is called.
"""
import time
from typing import Final
import pytest
from litellm.integrations.code_interpreter_interception.handler import (
_INTERCEPTION_ACTIVE_KEY as _ACTIVE_KEY,
)
from litellm.integrations.code_interpreter_interception.handler import (
_SANDBOX_KEY,
_SESSION_SCOPED_KEY,
LITELLM_CODE_EXECUTION_TOOL_NAME,
CodeInterpreterInterceptionLogger,
)
from litellm.llms.base_llm.sandbox.transformation import CodeExecutionResult
from litellm.types.integrations.custom_logger import (
CHAT_COMPLETION_AGENTIC_SURFACE,
CODE_INTERPRETER_STREAM_OPTIONS_KEY,
NON_CODE_INTERPRETER_INTERCEPTION_INTERNAL_PREFIXES,
is_interception_internal_key,
)
from litellm.types.utils import CallTypes
class FakeHandle:
def __init__(self, sandbox_id="sbx_fake"):
self.id = sandbox_id
class FakeSandbox:
"""Records acreate_sandbox / arun_code / adelete_sandbox calls."""
def __init__(self, stdout="42"):
self.stdout = stdout
self.create_calls = []
self.run_calls = []
self.delete_calls = []
async def acreate_sandbox(self, **kwargs):
self.create_calls.append(kwargs)
return FakeHandle()
async def arun_code(self, *, container, code, **kwargs):
self.run_calls.append({"container": container, "code": code})
return CodeExecutionResult(stdout=self.stdout)
async def adelete_sandbox(self, *, container, **kwargs):
self.delete_calls.append({"container": container})
return True
class FakeLogging:
def __init__(self, litellm_call_id="k1"):
self.litellm_call_id = litellm_call_id
self.model_call_details = {}
self.dynamic_success_callbacks = []
def pre_call(self, *args, **kwargs):
return None
def post_call(self, *args, **kwargs):
return None
def _function_call_item(call_id="c1", name=LITELLM_CODE_EXECUTION_TOOL_NAME):
return {
"type": "function_call",
"call_id": call_id,
"name": name,
"arguments": '{"code":"print(40 + 2)"}',
}
def _chat_function_call_item(call_id="call_1", name=LITELLM_CODE_EXECUTION_TOOL_NAME):
return {
"id": call_id,
"type": "function",
"function": {
"name": name,
"arguments": '{"code":"print(40 + 2)"}',
},
}
class FakeResponse:
def __init__(self, output):
self.output = output
def _iter_messages(plan):
patch = plan.request_patch
assert patch is not None, "plan.request_patch must be set"
assert patch.messages is not None, "plan.request_patch.messages must be set"
return patch.messages
def test_interception_internal_key_prefix_sets_preserve_code_interpreter_state():
assert is_interception_internal_key("_code_interpreter_interception_active")
assert not is_interception_internal_key(
"_code_interpreter_interception_active",
prefixes=NON_CODE_INTERPRETER_INTERCEPTION_INTERNAL_PREFIXES,
)
assert is_interception_internal_key(
"_websearch_interception_converted_stream",
prefixes=NON_CODE_INTERPRETER_INTERCEPTION_INTERNAL_PREFIXES,
)
@pytest.mark.asyncio
async def test_build_plan_runs_code_and_feeds_output_back():
sandbox = FakeSandbox(stdout="42")
logger = CodeInterpreterInterceptionLogger(sandbox_config=sandbox)
response = FakeResponse(output=[_function_call_item()])
plan = await logger.async_build_agentic_loop_plan(
tools={
"tool_calls": [
{
"call_id": "c1",
"name": LITELLM_CODE_EXECUTION_TOOL_NAME,
"arguments": '{"code":"print(40 + 2)"}',
}
]
},
model="gpt-5",
messages=[{"role": "user", "content": "x"}],
response=response,
anthropic_messages_provider_config=None,
anthropic_messages_optional_request_params={"tools": []},
logging_obj=FakeLogging(litellm_call_id="k1"),
stream=False,
kwargs={"litellm_call_id": "k1", _SANDBOX_KEY: "sbxkey1"},
)
assert sandbox.run_calls, "sandbox.arun_code must be invoked"
assert sandbox.run_calls[0]["code"] == "print(40 + 2)"
messages = _iter_messages(plan)
outputs = [m for m in messages if isinstance(m, dict) and m.get("type") == "function_call_output"]
assert outputs, "expected a function_call_output item appended"
output_item = next(m for m in outputs if m.get("call_id") == "c1")
assert "42" in str(output_item["output"])
@pytest.mark.asyncio
async def test_pre_call_converts_code_interpreter_tool():
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
kwargs = {
"tools": [{"type": "code_interpreter", "container": {"type": "auto"}}],
"custom_llm_provider": "openai",
}
result = await logger.async_pre_call_deployment_hook(kwargs, CallTypes.aresponses)
assert result is not None
tools = result["tools"]
assert not any(t.get("type") == "code_interpreter" for t in tools), "code_interpreter tool must be removed"
names = [t.get("name") or (t.get("function") or {}).get("name") for t in tools]
assert LITELLM_CODE_EXECUTION_TOOL_NAME in names
@pytest.mark.asyncio
async def test_pre_call_converts_code_interpreter_tool_for_chat_completions():
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
kwargs = {
"tools": [{"type": "code_interpreter", "container": {"type": "auto"}}],
"tool_choice": {"type": "code_interpreter"},
"custom_llm_provider": "openai",
}
result = await logger.async_pre_call_deployment_hook(kwargs, CallTypes.acompletion)
assert result is not None
tool = result["tools"][0]
assert tool["type"] == "function"
assert tool["function"]["name"] == LITELLM_CODE_EXECUTION_TOOL_NAME
assert tool["function"]["parameters"]["required"] == ["code"]
assert result["tool_choice"] == {
"type": "function",
"function": {"name": LITELLM_CODE_EXECUTION_TOOL_NAME},
}
assert result["litellm_metadata"][_ACTIVE_KEY] is True
assert result["litellm_metadata"][_SANDBOX_KEY] == result[_SANDBOX_KEY]
@pytest.mark.asyncio
@pytest.mark.parametrize(
"tool_choice",
[
{"type": "code_interpreter"},
{"type": "hosted_tool", "name": "code_interpreter"},
],
)
async def test_pre_call_rewrites_forced_code_interpreter_tool_choice(tool_choice):
"""A forced tool_choice targeting the native code_interpreter tool must be
rewritten to the generated function tool; otherwise the outbound request
references a tool that no longer exists and the provider rejects it."""
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
kwargs = {
"tools": [{"type": "code_interpreter", "container": {"type": "auto"}}],
"tool_choice": tool_choice,
"custom_llm_provider": "openai",
}
result = await logger.async_pre_call_deployment_hook(kwargs, CallTypes.aresponses)
assert result is not None
assert result["tool_choice"] == {
"type": "function",
"name": LITELLM_CODE_EXECUTION_TOOL_NAME,
}
@pytest.mark.asyncio
async def test_pre_call_leaves_unrelated_tool_choice_untouched():
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
kwargs = {
"tools": [{"type": "code_interpreter", "container": {"type": "auto"}}],
"tool_choice": "auto",
"custom_llm_provider": "openai",
}
result = await logger.async_pre_call_deployment_hook(kwargs, CallTypes.aresponses)
assert result is not None
assert result["tool_choice"] == "auto"
@pytest.mark.asyncio
async def test_pre_call_noop_on_non_responses():
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
kwargs = {
"tools": [{"type": "code_interpreter", "container": {"type": "auto"}}],
"custom_llm_provider": "openai",
}
result = await logger.async_pre_call_deployment_hook(kwargs, CallTypes.aembedding)
assert result is None
@pytest.mark.asyncio
async def test_pre_call_noop_on_chat_completion_without_code_interpreter_tool():
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
kwargs = {
"tools": [{"type": "web_search"}],
"custom_llm_provider": "openai",
}
result = await logger.async_pre_call_deployment_hook(kwargs, CallTypes.acompletion)
assert result is None
assert _ACTIVE_KEY not in kwargs
assert _SANDBOX_KEY not in kwargs
@pytest.mark.asyncio
async def test_should_run_detects_only_matching_function_call():
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
active_kwargs = {"_code_interpreter_interception_active": True}
match = FakeResponse(output=[_function_call_item(name=LITELLM_CODE_EXECUTION_TOOL_NAME)])
should_run, payload = await logger.async_should_run_agentic_loop(
response=match,
model="gpt-5",
messages=[{"role": "user", "content": "x"}],
tools=[],
stream=False,
custom_llm_provider="openai",
kwargs=active_kwargs,
)
assert should_run is True
assert payload.get("tool_calls")
no_match = FakeResponse(output=[_function_call_item(name="something_else")])
should_run2, payload2 = await logger.async_should_run_agentic_loop(
response=no_match,
model="gpt-5",
messages=[{"role": "user", "content": "x"}],
tools=[],
stream=False,
custom_llm_provider="openai",
kwargs=active_kwargs,
)
assert should_run2 is False
assert payload2 == {}
@pytest.mark.asyncio
async def test_container_reused_within_request_via_server_sandbox_key():
sandbox = FakeSandbox(stdout="42")
logger = CodeInterpreterInterceptionLogger(sandbox_config=sandbox)
response = FakeResponse(output=[_function_call_item()])
common = dict(
tools={
"tool_calls": [
{
"call_id": "c1",
"name": LITELLM_CODE_EXECUTION_TOOL_NAME,
"arguments": '{"code":"print(40 + 2)"}',
}
]
},
model="gpt-5",
messages=[{"role": "user", "content": "x"}],
response=response,
anthropic_messages_provider_config=None,
anthropic_messages_optional_request_params={"tools": []},
stream=False,
)
await logger.async_build_agentic_loop_plan(
logging_obj=FakeLogging(litellm_call_id="k1"),
kwargs={"litellm_call_id": "k1", _SANDBOX_KEY: "server-nonce-1"},
**common,
)
await logger.async_build_agentic_loop_plan(
logging_obj=FakeLogging(litellm_call_id="k1"),
kwargs={"litellm_call_id": "k1", _SANDBOX_KEY: "server-nonce-1"},
**common,
)
assert len(sandbox.create_calls) == 1, "the sandbox is reused across loop iterations sharing one server sandbox key"
@pytest.mark.asyncio
async def test_colliding_caller_call_id_does_not_share_sandbox():
"""Two requests with the same caller-controlled litellm_call_id but distinct
server-minted sandbox keys must NOT share a container; otherwise one user's
code could read another in-flight request's sandbox state."""
sandbox = FakeSandbox(stdout="42")
logger = CodeInterpreterInterceptionLogger(sandbox_config=sandbox)
common = dict(
tools={
"tool_calls": [
{
"call_id": "c1",
"name": LITELLM_CODE_EXECUTION_TOOL_NAME,
"arguments": '{"code":"print(1)"}',
}
]
},
model="gpt-5",
messages=[{"role": "user", "content": "x"}],
response=FakeResponse(output=[_function_call_item()]),
anthropic_messages_provider_config=None,
anthropic_messages_optional_request_params={"tools": []},
stream=False,
)
await logger.async_build_agentic_loop_plan(
logging_obj=FakeLogging(litellm_call_id="shared"),
kwargs={"litellm_call_id": "shared", _SANDBOX_KEY: "nonce-A"},
**common,
)
await logger.async_build_agentic_loop_plan(
logging_obj=FakeLogging(litellm_call_id="shared"),
kwargs={"litellm_call_id": "shared", _SANDBOX_KEY: "nonce-B"},
**common,
)
assert len(sandbox.create_calls) == 2, (
"distinct server sandbox keys must isolate sandboxes despite a colliding call id"
)
@pytest.mark.asyncio
async def test_pre_call_mints_server_sandbox_key():
"""The interceptor mints a server-side sandbox key (not derived from the
caller-controlled call id) when it activates."""
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
kwargs = {
"tools": [{"type": "code_interpreter", "container": {"type": "auto"}}],
"custom_llm_provider": "openai",
"litellm_call_id": "caller-supplied",
_SANDBOX_KEY: "caller-forged",
}
result = await logger.async_pre_call_deployment_hook(kwargs, CallTypes.aresponses)
assert result is not None
assert result[_SANDBOX_KEY] not in ("caller-forged", "caller-supplied")
assert len(result[_SANDBOX_KEY]) >= 16
@pytest.mark.asyncio
async def test_build_plan_records_code_interpreter_call_metadata():
sandbox = FakeSandbox(stdout="42")
logger = CodeInterpreterInterceptionLogger(sandbox_config=sandbox)
plan = await logger.async_build_agentic_loop_plan(
tools={
"tool_calls": [
{
"call_id": "c1",
"name": LITELLM_CODE_EXECUTION_TOOL_NAME,
"arguments": '{"code":"print(40 + 2)"}',
}
]
},
model="gpt-5",
messages=[{"role": "user", "content": "x"}],
response=FakeResponse(output=[_function_call_item()]),
anthropic_messages_provider_config=None,
anthropic_messages_optional_request_params={"tools": []},
logging_obj=FakeLogging(litellm_call_id="k1"),
stream=False,
kwargs={"litellm_call_id": "k1", _SANDBOX_KEY: "sbxkey1"},
)
calls = plan.metadata["code_interpreter_calls"]
assert calls, "build_plan must record a code_interpreter_call for re-injection"
assert calls[0]["code"] == "print(40 + 2)"
assert calls[0]["container_id"] == "sbx_fake"
assert calls[0]["type"] == "code_interpreter_call"
assert calls[0]["status"] == "completed"
assert calls[0]["outputs"] == [{"type": "logs", "logs": "42"}], (
"outputs must be an OpenAI-shaped logs array (not None) so clients that "
"iterate over code_interpreter_call.outputs do not break"
)
@pytest.mark.asyncio
async def test_build_plan_outputs_empty_array_when_no_stdout():
"""No stdout must still yield an iteration-safe empty array, never None."""
sandbox = FakeSandbox(stdout="")
logger = CodeInterpreterInterceptionLogger(sandbox_config=sandbox)
plan = await logger.async_build_agentic_loop_plan(
tools={
"tool_calls": [
{
"call_id": "c1",
"name": LITELLM_CODE_EXECUTION_TOOL_NAME,
"arguments": '{"code":"pass"}',
}
]
},
model="gpt-5",
messages=[{"role": "user", "content": "x"}],
response=FakeResponse(output=[_function_call_item()]),
anthropic_messages_provider_config=None,
anthropic_messages_optional_request_params={"tools": []},
logging_obj=FakeLogging(litellm_call_id="k1"),
stream=False,
kwargs={"litellm_call_id": "k1", _SANDBOX_KEY: "sbxkey1"},
)
assert plan.metadata["code_interpreter_calls"][0]["outputs"] == []
@pytest.mark.asyncio
async def test_post_hook_injects_code_interpreter_call_matching_openai_shape():
from litellm.types.integrations.custom_logger import AgenticLoopPlan
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
ci_item = {
"id": "ci_x",
"type": "code_interpreter_call",
"status": "completed",
"code": "print(1)",
"container_id": "sbx_fake",
"outputs": [{"type": "logs", "logs": "1"}],
}
plan = AgenticLoopPlan(
run_agentic_loop=True,
metadata={"code_interpreter_calls": [ci_item]},
)
response = FakeResponse(output=[{"type": "message", "content": []}])
out = await logger.async_post_agentic_loop_response_hook(response=response, plan=plan, kwargs={})
types = [item.get("type") for item in out.output]
assert types == ["code_interpreter_call", "message"], (
"code_interpreter_call must be re-injected before the message, matching OpenAI's native output ordering"
)
assert set(out.output[0].keys()) == {
"id",
"type",
"status",
"code",
"container_id",
"outputs",
}, "injected item must match OpenAI's code_interpreter_call keys exactly"
@pytest.mark.asyncio
async def test_post_hook_noop_without_recorded_calls():
from litellm.types.integrations.custom_logger import AgenticLoopPlan
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
response = FakeResponse(output=[{"type": "message", "content": []}])
out = await logger.async_post_agentic_loop_response_hook(
response=response, plan=AgenticLoopPlan(run_agentic_loop=True), kwargs={}
)
assert [item.get("type") for item in out.output] == ["message"]
@pytest.mark.asyncio
async def test_pre_call_forces_non_stream_for_loop():
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
kwargs = {
"tools": [{"type": "code_interpreter", "container": {"type": "auto"}}],
"custom_llm_provider": "openai",
"stream": True,
}
out = await logger.async_pre_call_deployment_hook(kwargs, CallTypes.aresponses)
assert out is not None
assert out["stream"] is False, "loop requires a non-streaming upstream call"
assert out["_code_interpreter_interception_converted_stream"] is True, (
"the converted-stream flag must be set so the final response is wrapped back into a stream for the caller"
)
@pytest.mark.asyncio
@pytest.mark.parametrize("call_type", (CallTypes.acompletion, CallTypes.aresponses))
async def test_pre_call_moves_stream_options_off_the_forced_non_stream_call(call_type: CallTypes):
logger: Final = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
stream_options: Final = {"include_usage": True}
kwargs: Final[dict[str, object]] = {
"tools": [{"type": "code_interpreter", "container": {"type": "auto"}}],
"custom_llm_provider": "azure",
"stream": True,
"stream_options": stream_options,
}
out: Final = await logger.async_pre_call_deployment_hook(kwargs, call_type)
assert out is not None
assert out["stream"] is False
assert "stream_options" not in out
assert out[CODE_INTERPRETER_STREAM_OPTIONS_KEY] == stream_options
async def _build_plan(logger, sandbox, call_id="k1", provider="openai"):
return await logger.async_build_agentic_loop_plan(
tools={
"tool_calls": [
{
"call_id": "c1",
"name": LITELLM_CODE_EXECUTION_TOOL_NAME,
"arguments": '{"code":"print(40 + 2)"}',
}
]
},
model="gpt-5",
messages=[{"role": "user", "content": "x"}],
response=FakeResponse(output=[_function_call_item()]),
anthropic_messages_provider_config=None,
anthropic_messages_optional_request_params={"tools": []},
logging_obj=FakeLogging(litellm_call_id=call_id),
stream=False,
kwargs={"litellm_call_id": call_id, _SANDBOX_KEY: "sbxkey1"},
)
@pytest.mark.asyncio
async def test_gate_refuses_without_server_active_marker():
"""A forged litellm_code_execution call must not trigger the loop unless the
pre-call hook actually converted a native code_interpreter tool."""
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
forged = FakeResponse(output=[_function_call_item(name=LITELLM_CODE_EXECUTION_TOOL_NAME)])
should_run, payload = await logger.async_should_run_agentic_loop(
response=forged,
model="gpt-5",
messages=[{"role": "user", "content": "x"}],
tools=[],
stream=False,
custom_llm_provider="openai",
kwargs={},
)
assert should_run is False
assert payload == {}
@pytest.mark.asyncio
async def test_gate_rechecks_provider_scope():
"""enabled_providers must be re-enforced at the gate, not only in pre-call."""
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox(), enabled_providers=["openai"])
response = FakeResponse(output=[_function_call_item(name=LITELLM_CODE_EXECUTION_TOOL_NAME)])
should_run, _ = await logger.async_should_run_agentic_loop(
response=response,
model="claude-x",
messages=[{"role": "user", "content": "x"}],
tools=[],
stream=False,
custom_llm_provider="anthropic",
kwargs={_ACTIVE_KEY: True},
)
assert should_run is False
@pytest.mark.asyncio
async def test_chat_completion_gate_detects_code_execution_tool_call():
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
response = {"choices": [{"message": {"tool_calls": [_chat_function_call_item(call_id="call_123")]}}]}
should_run, payload = await logger.async_should_run_agentic_loop(
response=response,
model="gpt-5",
messages=[{"role": "user", "content": "x"}],
tools=[],
stream=False,
custom_llm_provider="openai",
kwargs={
_ACTIVE_KEY: True,
"_agentic_loop_api_surface": CHAT_COMPLETION_AGENTIC_SURFACE,
},
)
assert should_run is True
assert payload["tool_calls"][0]["id"] == "call_123"
assert payload["tool_calls"][0]["arguments"] == '{"code":"print(40 + 2)"}'
@pytest.mark.asyncio
async def test_chat_completion_gate_refuses_without_server_active_marker():
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
response = {"choices": [{"message": {"tool_calls": [_chat_function_call_item()]}}]}
should_run, payload = await logger.async_should_run_agentic_loop(
response=response,
model="gpt-5",
messages=[{"role": "user", "content": "x"}],
tools=[],
stream=False,
custom_llm_provider="openai",
kwargs={"_agentic_loop_api_surface": CHAT_COMPLETION_AGENTIC_SURFACE},
)
assert should_run is False
assert payload == {}
@pytest.mark.asyncio
async def test_chat_completion_build_plan_runs_code_and_appends_tool_message():
sandbox = FakeSandbox(stdout="42")
logger = CodeInterpreterInterceptionLogger(sandbox_config=sandbox)
native_chat_tool = {"type": "code_interpreter", "container": {"type": "auto"}}
plan = await logger.async_build_agentic_loop_plan(
tools={
"tool_calls": [
{
"id": "call_1",
"name": LITELLM_CODE_EXECUTION_TOOL_NAME,
"arguments": '{"code":"print(40 + 2)"}',
}
]
},
model="gpt-5",
messages=[{"role": "user", "content": "x"}],
response={"choices": [{"message": {"tool_calls": [_chat_function_call_item()]}}]},
anthropic_messages_provider_config=None,
anthropic_messages_optional_request_params={
"tools": [native_chat_tool],
"tool_choice": {"type": "code_interpreter", "container": {"type": "auto"}},
"temperature": 0,
},
logging_obj=FakeLogging(litellm_call_id="k1"),
stream=False,
kwargs={
"acompletion": True,
"litellm_call_id": "k1",
_ACTIVE_KEY: True,
_SANDBOX_KEY: "sbxkey1",
"_code_interpreter_interception_converted_stream": True,
"_agentic_loop_api_surface": CHAT_COMPLETION_AGENTIC_SURFACE,
},
)
assert sandbox.run_calls[0]["code"] == "print(40 + 2)"
patch = plan.request_patch
assert patch is not None
assert patch.tools == [
{
"type": "function",
"function": {
"name": LITELLM_CODE_EXECUTION_TOOL_NAME,
"description": "Execute python code in a sandbox and return stdout.",
"parameters": {
"type": "object",
"properties": {"code": {"type": "string"}},
"required": ["code"],
},
},
}
]
assert patch.optional_params == {"temperature": 0}
assert patch.kwargs == {
"litellm_call_id": "k1",
_ACTIVE_KEY: True,
_SANDBOX_KEY: "sbxkey1",
"_code_interpreter_interception_converted_stream": True,
"_agentic_loop_api_surface": CHAT_COMPLETION_AGENTIC_SURFACE,
}
assert patch.messages is not None
assert patch.messages[-2]["role"] == "assistant"
assert patch.messages[-2]["tool_calls"][0]["id"] == "call_1"
assert patch.messages[-1] == {
"role": "tool",
"tool_call_id": "call_1",
"content": "42",
}
assert plan.metadata["code_interpreter_calls"][0]["code"] == "print(40 + 2)"
@pytest.mark.asyncio
async def test_pre_call_strips_client_forged_marker_on_initial_request():
"""A client cannot pre-set the active marker on the original request: with no
native code_interpreter tool, any client-supplied interception markers in
litellm_metadata are scrubbed and the active flag in kwargs is cleared."""
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
kwargs = {
"tools": [{"type": "web_search"}],
"custom_llm_provider": "openai",
_ACTIVE_KEY: True,
"litellm_metadata": {
_ACTIVE_KEY: True,
_SANDBOX_KEY: "client-forged",
"safe_user_value": "kept",
},
}
await logger.async_pre_call_deployment_hook(kwargs, CallTypes.aresponses)
assert _ACTIVE_KEY not in kwargs, (
"no native code_interpreter tool was present, so a client-supplied active marker must be cleared"
)
assert kwargs["litellm_metadata"] == {"safe_user_value": "kept"}
@pytest.mark.asyncio
async def test_pre_call_strips_forged_loop_controls_then_mints_own_markers():
"""On an INITIAL request (no server-set _agentic_loop_depth) a client cannot
smuggle loop-control state: forged _agentic_loop_depth / max_agentic_loops and
interception markers in litellm_metadata are stripped before the interceptor
activates, so the only interception markers that survive are the ones the
server mints for the converted code_interpreter tool."""
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
kwargs = {
"tools": [{"type": "code_interpreter", "container": {"type": "auto"}}],
"custom_llm_provider": "openai",
"litellm_metadata": {
_ACTIVE_KEY: True,
_SANDBOX_KEY: "client-forged",
"_agentic_loop_depth": 99,
"max_agentic_loops": 999,
"safe_user_value": "kept",
},
}
result = await logger.async_pre_call_deployment_hook(kwargs, CallTypes.acompletion)
assert result is not None
metadata = result["litellm_metadata"]
assert metadata["safe_user_value"] == "kept"
assert "_agentic_loop_depth" not in metadata, "forged loop depth must be stripped"
assert "max_agentic_loops" not in metadata, "forged loop cap must be stripped"
assert metadata[_ACTIVE_KEY] is True
assert metadata[_SANDBOX_KEY] == result[_SANDBOX_KEY]
assert metadata[_SANDBOX_KEY] != "client-forged", (
"the surviving sandbox key must be the server-minted one, not the forged value the client supplied"
)
@pytest.mark.asyncio
async def test_pre_call_preserves_marker_on_server_followup():
"""On a server-driven followup (depth>0) the marker is trusted and kept."""
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
kwargs = {
"tools": [{"type": "function", "name": LITELLM_CODE_EXECUTION_TOOL_NAME}],
"custom_llm_provider": "openai",
"_agentic_loop_depth": 1,
_ACTIVE_KEY: True,
}
await logger.async_pre_call_deployment_hook(kwargs, CallTypes.aresponses)
assert kwargs.get(_ACTIVE_KEY) is True, (
"the server-set marker must survive followup requests so multi-round code execution keeps working"
)
@pytest.mark.asyncio
async def test_sandbox_deleted_after_loop_completes():
sandbox = FakeSandbox(stdout="42")
logger = CodeInterpreterInterceptionLogger(sandbox_config=sandbox)
plan = await _build_plan(logger, sandbox, call_id="k1")
assert sandbox.create_calls, "sandbox must be created during the loop"
assert not sandbox.delete_calls, "sandbox must outlive the loop until the final hook"
await logger.async_post_agentic_loop_response_hook(
response=FakeResponse(output=[{"type": "message", "content": []}]),
plan=plan,
kwargs={},
)
assert len(sandbox.delete_calls) == 1, (
"the sandbox must be deleted once the final response is assembled, otherwise it keeps running and billing"
)
assert "sbxkey1" not in logger._container_cache
@pytest.mark.asyncio
async def test_post_hook_delete_is_idempotent_across_loop_levels():
sandbox = FakeSandbox(stdout="42")
logger = CodeInterpreterInterceptionLogger(sandbox_config=sandbox)
plan = await _build_plan(logger, sandbox, call_id="k1")
response = FakeResponse(output=[{"type": "message", "content": []}])
await logger.async_post_agentic_loop_response_hook(response=response, plan=plan, kwargs={})
await logger.async_post_agentic_loop_response_hook(response=response, plan=plan, kwargs={})
assert len(sandbox.delete_calls) == 1, (
"deleting an already-removed container must be a no-op so unwinding loop levels do not double-delete"
)
@pytest.mark.asyncio
async def test_build_plan_deletes_sandbox_when_execution_raises():
"""If sandbox execution raises before a plan is built (e.g. E2B aborts
output over its cap), the cached sandbox must be deleted before re-raising,
otherwise a caller can leak paid containers until the prune TTL."""
class RaisingSandbox(FakeSandbox):
async def arun_code(self, *, container, code, **kwargs):
raise ValueError("output exceeded cap")
sandbox = RaisingSandbox()
logger = CodeInterpreterInterceptionLogger(sandbox_config=sandbox)
with pytest.raises(ValueError, match="exceeded cap"):
await _build_plan(logger, sandbox, call_id="k1")
assert len(sandbox.create_calls) == 1, "the sandbox must have been created"
assert len(sandbox.delete_calls) == 1, (
"a build failure must delete the cached sandbox so it does not keep running and billing"
)
assert "sbxkey1" not in logger._container_cache
@pytest.mark.asyncio
async def test_cleanup_hook_deletes_sandbox():
sandbox = FakeSandbox(stdout="42")
logger = CodeInterpreterInterceptionLogger(sandbox_config=sandbox)
plan = await _build_plan(logger, sandbox, call_id="k1")
await logger.async_agentic_loop_cleanup_hook(plan=plan, kwargs={})
assert len(sandbox.delete_calls) == 1, (
"the cleanup hook must delete the sandbox so a rerun failure cannot leak a running container"
)
assert "sbxkey1" not in logger._container_cache
@pytest.mark.asyncio
async def test_cleanup_hook_is_idempotent_with_post_hook():
sandbox = FakeSandbox(stdout="42")
logger = CodeInterpreterInterceptionLogger(sandbox_config=sandbox)
plan = await _build_plan(logger, sandbox, call_id="k1")
await logger.async_post_agentic_loop_response_hook(
response=FakeResponse(output=[{"type": "message", "content": []}]),
plan=plan,
kwargs={},
)
await logger.async_agentic_loop_cleanup_hook(plan=plan, kwargs={})
assert len(sandbox.delete_calls) == 1, (
"cleanup running in finally after the success-path post hook already deleted the sandbox must not double-delete"
)
@pytest.mark.asyncio
async def test_responses_plan_cleans_up_sandbox_when_followup_raises():
"""If the agentic rerun fails, _execute_responses_agentic_plan must still
invoke the cleanup hook so the sandbox is not left running."""
import litellm
from litellm.integrations.custom_logger import CustomLogger
from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
from litellm.types.integrations.custom_logger import (
AgenticLoopPlan,
AgenticLoopRequestPatch,
)
cleanup_calls = []
class CleanupCallback(CustomLogger):
async def async_post_agentic_loop_response_hook(self, response, plan, kwargs):
return response
async def async_agentic_loop_cleanup_hook(self, plan, kwargs):
cleanup_calls.append(plan)
plan = AgenticLoopPlan(
run_agentic_loop=True,
request_patch=AgenticLoopRequestPatch(model="gpt-5", messages=[{"role": "user", "content": "x"}]),
metadata={"sandbox_key": "sbxkey1"},
)
original = litellm.aresponses
async def _boom(*args, **kwargs):
raise RuntimeError("upstream blew up")
litellm.aresponses = _boom
try:
with pytest.raises(RuntimeError, match="upstream blew up"):
await BaseLLMHTTPHandler()._execute_responses_agentic_plan(
plan=plan,
model="gpt-5",
response_api_optional_request_params={},
logging_obj=FakeLogging(litellm_call_id="k1"),
kwargs={},
depth=0,
max_loops=3,
fingerprints=[],
fingerprint="fp",
callback=CleanupCallback(),
)
finally:
litellm.aresponses = original
assert cleanup_calls == [plan], (
"cleanup hook must run in finally even when the rerun raises, otherwise "
"the sandbox keeps running until the prune TTL"
)
@pytest.mark.asyncio
async def test_run_code_does_not_re_resolve_registry(monkeypatch):
"""Params resolved once at create time must be reused for running code, so a
registry clear between create and run cannot turn into a create-then-fail."""
import litellm
from litellm.sandbox import sandbox_tools
sandbox_tools.register_sandbox_tools(
[
{
"sandbox_tool_name": "e2b_default",
"litellm_params": {"sandbox_provider": "e2b", "api_key": "sk-x"},
}
]
)
create_kwargs = {}
run_kwargs = {}
async def fake_acreate_sandbox(**kwargs):
create_kwargs.update(kwargs)
return FakeHandle()
async def fake_arun_code(**kwargs):
run_kwargs.update(kwargs)
return CodeExecutionResult(stdout="ok")
monkeypatch.setattr(litellm, "acreate_sandbox", fake_acreate_sandbox)
monkeypatch.setattr(litellm, "arun_code", fake_arun_code)
logger = CodeInterpreterInterceptionLogger(sandbox_tool_name="e2b_default")
try:
container, params = await logger._get_or_create_container(cache_key="k1")
assert params is not None and params["sandbox_provider"] == "e2b"
sandbox_tools.clear_sandbox_tools()
stdout = await logger._run_tool_call(container=container, params=params, arguments='{"code":"print(1)"}')
finally:
sandbox_tools.clear_sandbox_tools()
assert stdout == "ok", "run must succeed using the params captured at create time"
assert run_kwargs["provider"] == "e2b"
@pytest.mark.asyncio
async def test_run_tool_call_surfaces_execution_error():
"""A sandbox execution error must be fed back to the model as a labelled
string, not raised, so the agentic loop can react to it."""
class ErroringSandbox(FakeSandbox):
async def arun_code(self, *, container, code, **kwargs):
self.run_calls.append({"container": container, "code": code})
return CodeExecutionResult(stdout="", error={"name": "ValueError", "value": "boom"})
sandbox = ErroringSandbox()
logger = CodeInterpreterInterceptionLogger(sandbox_config=sandbox)
container = await logger._create_container()
stdout = await logger._run_tool_call(
container=container[0], params=None, arguments='{"code":"raise ValueError(1)"}'
)
assert stdout == "[execution error] boom"
@pytest.mark.asyncio
async def test_run_tool_call_reports_unparseable_arguments():
"""Malformed tool arguments must produce a parse error string the model can
see rather than crashing the interceptor."""
sandbox = FakeSandbox()
logger = CodeInterpreterInterceptionLogger(sandbox_config=sandbox)
container = await logger._create_container()
stdout = await logger._run_tool_call(container=container[0], params=None, arguments="not-json")
assert stdout == "[invalid tool arguments: could not parse code]"
assert not sandbox.run_calls, "code must not run when arguments cannot be parsed"
@pytest.mark.asyncio
async def test_pre_call_skips_provider_outside_scope():
"""enabled_providers must filter the pre-call conversion so a request to an
out-of-scope provider is left untouched."""
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox(), enabled_providers=["openai"])
kwargs = {
"tools": [{"type": "code_interpreter", "container": {"type": "auto"}}],
"custom_llm_provider": "anthropic",
}
result = await logger.async_pre_call_deployment_hook(kwargs, CallTypes.aresponses)
assert result is None
assert kwargs["tools"][0]["type"] == "code_interpreter", "tool must be untouched"
assert _ACTIVE_KEY not in kwargs
@pytest.mark.asyncio
async def test_resolve_provider_falls_back_to_model_lookup():
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
assert logger._resolve_provider({"custom_llm_provider": "openai"}) == "openai"
assert logger._resolve_provider({"model": "gpt-5"}) == "openai"
assert logger._resolve_provider({"model": 123}) is None
assert logger._resolve_provider({"model": "no-such-provider-xyz"}) is None
@pytest.mark.asyncio
async def test_create_container_without_sandbox_raises():
"""The registry path must raise a clear error when no sandbox is resolvable
instead of silently creating nothing."""
logger = CodeInterpreterInterceptionLogger(sandbox_tool_name="missing")
with pytest.raises(ValueError, match="no sandbox available"):
await logger._create_container()
@pytest.mark.asyncio
async def test_run_code_without_params_raises():
logger = CodeInterpreterInterceptionLogger(sandbox_tool_name="missing")
with pytest.raises(ValueError, match="no sandbox available to run code"):
await logger._run_code(container=FakeHandle(), params=None, code="print(1)")
@pytest.mark.asyncio
async def test_delete_container_swallows_errors():
"""A delete failure must not propagate; the request already succeeded."""
class FailingDeleteSandbox(FakeSandbox):
async def adelete_sandbox(self, *, container, **kwargs):
raise RuntimeError("e2b unreachable")
sandbox = FailingDeleteSandbox()
logger = CodeInterpreterInterceptionLogger(sandbox_config=sandbox)
container, params = await logger._create_container()
await logger._delete_container(container=container, params=params)
@pytest.mark.asyncio
async def test_prune_expired_cache_deletes_underlying_container():
"""Expired cache entries must have their sandbox deleted, not just dropped,
otherwise an orphaned sandbox keeps running."""
import litellm.integrations.code_interpreter_interception.handler as handler_mod
sandbox = FakeSandbox()
logger = CodeInterpreterInterceptionLogger(sandbox_config=sandbox)
container, params = await logger._create_container()
logger._container_cache["old"] = (
container,
params,
time.time() - handler_mod._CACHE_TTL_SECONDS - 1,
None,
)
await logger._prune_expired_cache()
assert "old" not in logger._container_cache
assert len(sandbox.delete_calls) == 1, "expired sandbox must be deleted"
@pytest.mark.asyncio
async def test_normalize_messages_handles_str_and_unknown():
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
assert logger._normalize_messages("hi") == [{"role": "user", "content": "hi"}]
assert logger._normalize_messages([{"role": "user"}]) == [{"role": "user"}]
assert logger._normalize_messages(42) == []
def test_from_config_yaml_reads_fields():
cfg = {
"enabled": False,
"enabled_providers": ["openai"],
"sandbox_tool_name": "e2b_default",
}
logger = CodeInterpreterInterceptionLogger.from_config_yaml(cfg)
assert logger.enabled is False
assert logger.enabled_providers == ["openai"]
assert logger.sandbox_tool_name == "e2b_default"
def test_initialize_from_proxy_config_prefers_litellm_settings():
logger = CodeInterpreterInterceptionLogger.initialize_from_proxy_config(
litellm_settings={
"code_interpreter_interception_params": {
"enabled_providers": ["openai"],
"sandbox_tool_name": "e2b_default",
}
},
callback_specific_params={},
)
assert logger.enabled_providers == ["openai"]
assert logger.sandbox_tool_name == "e2b_default"
@pytest.mark.asyncio
async def test_build_plan_handles_dict_shaped_response():
"""A responses payload delivered as a plain dict (not an object) must flow
through detection, execution, and re-injection the same as the typed form."""
sandbox = FakeSandbox(stdout="42")
logger = CodeInterpreterInterceptionLogger(sandbox_config=sandbox)
dict_response = {"output": [_function_call_item()]}
plan = await logger.async_build_agentic_loop_plan(
tools={"tool_calls": logger._extract_code_execution_tool_calls(dict_response)},
model="gpt-5",
messages=[{"role": "user", "content": "x"}],
response=dict_response,
anthropic_messages_provider_config=None,
anthropic_messages_optional_request_params={"tools": []},
logging_obj=FakeLogging(litellm_call_id="k1"),
stream=False,
kwargs={"litellm_call_id": "k1", _SANDBOX_KEY: "sbxkey1"},
)
assert sandbox.run_calls, "code must run for a dict-shaped response"
assert plan.metadata["code_interpreter_calls"][0]["code"] == "print(40 + 2)"
out = await logger.async_post_agentic_loop_response_hook(
response={"output": [{"type": "message", "content": []}]},
plan=plan,
kwargs={},
)
assert [item.get("type") for item in out["output"]] == [
"code_interpreter_call",
"message",
], "the dict-shaped response must get the code_interpreter_call re-injected"
@pytest.mark.asyncio
async def test_extract_tool_calls_reads_object_attributes():
"""Detection must work when output items are objects with attributes, not
only dicts."""
class Item:
def __init__(self):
self.type = "function_call"
self.name = LITELLM_CODE_EXECUTION_TOOL_NAME
self.call_id = "c9"
self.arguments = '{"code":"print(1)"}'
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
calls = logger._extract_code_execution_tool_calls(FakeResponse(output=[Item()]))
assert len(calls) == 1
assert calls[0]["call_id"] == "c9"
assert calls[0]["arguments"] == '{"code":"print(1)"}'
# ---------------------------------------------------------------------------
# Sticky session tests
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_pre_call_uses_session_id_from_metadata_as_sandbox_key():
"""When session_id is in request metadata, it becomes the sandbox key so the
container is shared across requests in the same session."""
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
session_id = "conv-abc-123"
kwargs = {
"tools": [{"type": "code_interpreter", "container": {"type": "auto"}}],
"custom_llm_provider": "openai",
"metadata": {"session_id": session_id},
}
result = await logger.async_pre_call_deployment_hook(kwargs, CallTypes.acompletion)
assert result is not None
assert result[_SANDBOX_KEY] == session_id
assert result[_SESSION_SCOPED_KEY] is True
assert result["litellm_metadata"][_SANDBOX_KEY] == session_id
assert result["litellm_metadata"][_SESSION_SCOPED_KEY] is True
@pytest.mark.asyncio
async def test_pre_call_uses_session_id_from_litellm_metadata():
"""session_id in litellm_metadata also works as the sticky key."""
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
session_id = "sess-xyz-789"
kwargs = {
"tools": [{"type": "code_interpreter", "container": {"type": "auto"}}],
"custom_llm_provider": "openai",
"litellm_metadata": {"session_id": session_id},
}
result = await logger.async_pre_call_deployment_hook(kwargs, CallTypes.acompletion)
assert result is not None
assert result[_SANDBOX_KEY] == session_id
assert result[_SESSION_SCOPED_KEY] is True
@pytest.mark.asyncio
async def test_pre_call_without_session_id_still_mints_random_key():
"""Requests without a session_id still get a server-minted random sandbox key."""
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
kwargs = {
"tools": [{"type": "code_interpreter", "container": {"type": "auto"}}],
"custom_llm_provider": "openai",
}
result = await logger.async_pre_call_deployment_hook(kwargs, CallTypes.acompletion)
assert result is not None
assert _SESSION_SCOPED_KEY not in result or result[_SESSION_SCOPED_KEY] is False
assert len(result[_SANDBOX_KEY]) >= 16
@pytest.mark.asyncio
async def test_session_scoped_sandbox_survives_agentic_loop_cleanup():
"""A session-scoped sandbox must NOT be deleted by the cleanup or post hooks;
it needs to persist across requests within the same session."""
sandbox = FakeSandbox(stdout="42")
logger = CodeInterpreterInterceptionLogger(sandbox_config=sandbox)
session_id = "conv-persist-me"
plan = await logger.async_build_agentic_loop_plan(
tools={
"tool_calls": [
{
"call_id": "c1",
"name": LITELLM_CODE_EXECUTION_TOOL_NAME,
"arguments": '{"code":"x = 10"}',
}
]
},
model="gpt-4o-mini",
messages=[{"role": "user", "content": "set x"}],
response=FakeResponse(output=[_function_call_item()]),
anthropic_messages_provider_config=None,
anthropic_messages_optional_request_params={"tools": []},
logging_obj=FakeLogging(litellm_call_id="k1"),
stream=False,
kwargs={
"litellm_call_id": "k1",
_SANDBOX_KEY: session_id,
_SESSION_SCOPED_KEY: True,
},
)
assert plan.metadata["is_session_scoped"] is True
await logger.async_post_agentic_loop_response_hook(
response=FakeResponse(output=[{"type": "message", "content": []}]),
plan=plan,
kwargs={},
)
await logger.async_agentic_loop_cleanup_hook(plan=plan, kwargs={})
assert not sandbox.delete_calls, (
"session-scoped sandbox must not be deleted after a single agentic loop; "
"it must persist for the next request in the session"
)
assert session_id in logger._container_cache, "session-scoped container must remain in cache after loop ends"
@pytest.mark.asyncio
async def test_session_scoped_sandbox_reused_across_sequential_requests():
"""Two sequential requests with the same session_id must share one container,
confirming state (e.g. assigned variables) can persist across HTTP requests."""
sandbox = FakeSandbox(stdout="42")
logger = CodeInterpreterInterceptionLogger(sandbox_config=sandbox)
session_id = "conv-reuse-me"
common_plan_args = dict(
tools={
"tool_calls": [
{
"call_id": "c1",
"name": LITELLM_CODE_EXECUTION_TOOL_NAME,
"arguments": '{"code":"print(1)"}',
}
]
},
model="gpt-4o-mini",
messages=[{"role": "user", "content": "x"}],
response=FakeResponse(output=[_function_call_item()]),
anthropic_messages_provider_config=None,
anthropic_messages_optional_request_params={"tools": []},
stream=False,
)
session_kwargs = {_SANDBOX_KEY: session_id, _SESSION_SCOPED_KEY: True}
plan1 = await logger.async_build_agentic_loop_plan(
logging_obj=FakeLogging(litellm_call_id="req1"),
kwargs={"litellm_call_id": "req1", **session_kwargs},
**common_plan_args,
)
await logger.async_post_agentic_loop_response_hook(
response=FakeResponse(output=[{"type": "message", "content": []}]),
plan=plan1,
kwargs={},
)
plan2 = await logger.async_build_agentic_loop_plan(
logging_obj=FakeLogging(litellm_call_id="req2"),
kwargs={"litellm_call_id": "req2", **session_kwargs},
**common_plan_args,
)
await logger.async_post_agentic_loop_response_hook(
response=FakeResponse(output=[{"type": "message", "content": []}]),
plan=plan2,
kwargs={},
)
assert len(sandbox.create_calls) == 1, (
"a single container must serve both requests in the same session; "
"two creates means state cannot persist between requests"
)
assert len(sandbox.delete_calls) == 0, "the session container must still be alive after both requests complete"
@pytest.mark.asyncio
async def test_non_session_sandbox_still_deleted_after_loop():
"""Without a session_id, the existing per-request ephemeral behavior is unchanged."""
sandbox = FakeSandbox(stdout="42")
logger = CodeInterpreterInterceptionLogger(sandbox_config=sandbox)
plan = await _build_plan(logger, sandbox, call_id="k1")
await logger.async_post_agentic_loop_response_hook(
response=FakeResponse(output=[{"type": "message", "content": []}]),
plan=plan,
kwargs={},
)
assert len(sandbox.delete_calls) == 1, "non-session sandbox must still be cleaned up after each request"
@pytest.mark.asyncio
async def test_sandbox_key_scoped_to_api_key_hash_isolates_users():
"""Two callers supplying the same session_id but different API key hashes must
each get their own sandbox; sharing across tenants would let one read or mutate
the other's interpreter state."""
logger = CodeInterpreterInterceptionLogger(sandbox_config=FakeSandbox())
session_id = "same-session-id"
result_a = await logger.async_pre_call_deployment_hook(
{
"tools": [{"type": "code_interpreter"}],
"custom_llm_provider": "openai",
"metadata": {"session_id": session_id},
"user_api_key_hash": "hash-for-tenant-a",
},
CallTypes.acompletion,
)
result_b = await logger.async_pre_call_deployment_hook(
{
"tools": [{"type": "code_interpreter"}],
"custom_llm_provider": "openai",
"metadata": {"session_id": session_id},
"user_api_key_hash": "hash-for-tenant-b",
},
CallTypes.acompletion,
)
assert result_a is not None and result_b is not None
assert result_a[_SANDBOX_KEY] != result_b[_SANDBOX_KEY], (
"same session_id from different API keys must yield different sandbox keys; "
"otherwise tenant A can read tenant B's sandbox state"
)
assert "hash-for-tenant-a" in result_a[_SANDBOX_KEY]
assert "hash-for-tenant-b" in result_b[_SANDBOX_KEY]
@pytest.mark.asyncio
async def test_per_identity_cap_evicts_lru_session():
"""When a single identity holds the cap limit of session sandboxes and opens a
new one, the least-recently-used session is evicted so the allocation stays
bounded. Without this, rotating session IDs is an unbounded sandbox leak."""
from litellm.integrations.code_interpreter_interception.handler import _SESSION_SCOPED_PER_IDENTITY_CAP
sandbox = FakeSandbox(stdout="ok")
logger = CodeInterpreterInterceptionLogger(sandbox_config=sandbox)
identity = "hash-for-identity-x"
for i in range(_SESSION_SCOPED_PER_IDENTITY_CAP):
await logger._get_or_create_container(
cache_key=f"{identity}:session-{i}",
identity=identity,
)
logger._container_cache[f"{identity}:session-{i}"] = (
logger._container_cache[f"{identity}:session-{i}"][0],
logger._container_cache[f"{identity}:session-{i}"][1],
float(i),
identity,
)
assert len(logger._container_cache) == _SESSION_SCOPED_PER_IDENTITY_CAP
await logger._get_or_create_container(
cache_key=f"{identity}:session-new",
identity=identity,
)
assert len(logger._container_cache) == _SESSION_SCOPED_PER_IDENTITY_CAP, (
"adding a new session beyond the cap must evict one entry so total stays bounded"
)
assert f"{identity}:session-0" not in logger._container_cache, (
"the entry with the oldest last_accessed timestamp must be evicted first (LRU)"
)
assert len(sandbox.delete_calls) == 1, "evicted sandbox must be deleted, not just removed from cache"
Reference in a new issue View git blame Copy permalink