litellm/tests/unit/sandbox/test_opensandbox_sandbox.py
rj-cairn d06b889d41
fix(streaming): keep merged think tags balanced and split SSE lines only on CR/LF (#48755)
* fix(vertex_ai): split Gemini SSE lines on CR/LF only so U+2028 in text keeps the stream

The Gemini and Vertex AI streaming path read the SSE body with httpx
iter_lines()/aiter_lines(). Those follow str.splitlines(), which also breaks
on U+0085, U+2028, U+2029, \x0b, \x0c and \x1c-\x1e. JSON leaves U+0085,
U+2028 and U+2029 unescaped inside strings, so a data: line whose text holds
one of them was cut in two. The tail is not a data: line and the head never
parses, and the accumulate-and-retry fallback then carries the unparsed
prefix forward, so every later event is appended to it and the stream ends
with no more content and no usage.

SSE lines end only at CRLF, LF or CR. The new iter_sse_lines/aiter_sse_lines
split decoded text on exactly those, treating a CR at the end of one chunk and
an LF at the start of the next as one break. The first request and every
long-decode hop now read through them.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(streaming): keep merged think tags balanced and split shared SSE lines only on CR/LF

Close an open <think> block on the first content or tool-call delta and at stream end, reopen it when reasoning returns, and route the shared chat completions stream through the SSE line splitter so U+2028 and U+2029 inside JSON no longer drop chunks.

* fix(streaming): exact-match the SSE [DONE] marker in BaseModelResponseIterator

* fix(streaming): split every SSE reader on CR/LF only and keep the logged think block closed

Switch the remaining httpx stream readers (Anthropic chat, Google GenAI,
Interactions, Codestral, openai_like, Predibase, LangGraph, Vertex Agent
Engine, A2A providers, Azure AI agents, sandboxes, the A2A endpoint, and
others) from iter_lines/aiter_lines to iter_sse_lines/aiter_sse_lines, so a
raw U+2028 or U+2029 inside a JSON string no longer splits an event.

Keep the chunk that closes a think block in the logged and cached response
when a stream ends without a finish_reason, close the block before a legacy
function_call, and make sync Codestral FIM streaming send stream=true and use
the text-completion-codestral stream parser.

Test fakes that only mocked iter_lines now stream real httpx bytes and
assert the content they feed in.

* fix(streaming): split buffered SSE parsers only on CR and LF, shard the codestral tests

Buffered SSE and JSONL parsers used str.splitlines(), which also breaks on
U+2028, U+2029 and U+0085 that JSON allows raw inside strings. They now use
split_sse_lines from sse_lines.py, which also gains a str.split fast path.
Adds tests/unit/llms/codestral to the unit workflow shard.

* fix(streaming): keep U+2028 in restamped Anthropic frames and fold CRLF before splitting SSE lines

* test: type the new stream tests, drop a comment, and benchmark the SSE line splitter

* test: type the iterator class parameters in the line-separator stream tests

* fix(responses): stream tool call arguments opened in a text chunk

* fix(responses): hold a tool call that shares a chunk with text until the text ends

* test: type the a2a SSE handler helper and the sandbox fake responses

---------

Co-authored-by: Tim Vink <vinktim@gmail.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Co-authored-by: michelligabriele <gabriele.michelli@icloud.com>
2026-10-28 17:00:29 -07:00

646 lines
20 KiB
Python

import json
import httpx
import pytest
import litellm
from litellm.llms.base_llm.sandbox.transformation import ContainerHandle
from litellm.llms.opensandbox.sandbox.transformation import (
MAX_OUTPUT_BYTES,
OPEN_SANDBOX_DEFAULT_TEMPLATE,
OpenSandboxSandboxConfig,
)
from litellm.utils import ProviderConfigManager
TEST_API_BASE = "https://sandbox.test/v1"
def http_status_error(status_code, url="http://test"):
return httpx.HTTPStatusError(
f"status {status_code}",
request=httpx.Request("GET", url),
response=httpx.Response(status_code),
)
def sse(data):
return f"data: {json.dumps(data)}"
def _line_stream(lines: list[str]) -> httpx.Response:
return httpx.Response(200, stream=httpx.ByteStream("".join(f"{line}\n" for line in lines).encode()))
class FakeResponse:
def __init__(self, *, json_data: object = None, status_code: int = 200) -> None:
self._json = json_data
self.status_code = status_code
def json(self):
return self._json
def raise_for_status(self):
if self.status_code >= 400:
raise http_status_error(self.status_code)
class FakeHTTPClient:
def __init__(
self,
*,
create_json=None,
sandbox_states=None,
endpoint_json=None,
endpoint_responses=None,
execute_lines=None,
delete_status=204,
execute_raises=None,
):
self.create_json = create_json or {
"id": "osb_123",
"status": {"state": "Running"},
"createdAt": "2026-01-01T00:00:00Z",
"entrypoint": ["/opt/code-interpreter/code-interpreter.sh"],
}
self.sandbox_states = list(
sandbox_states
or [
{
"id": "osb_123",
"status": {"state": "Running"},
"createdAt": "2026-01-01T00:00:00Z",
"entrypoint": ["/opt/code-interpreter/code-interpreter.sh"],
}
]
)
self.endpoint_json = endpoint_json or {
"endpoint": "execd.local:44772",
"headers": {"X-EXECD-ACCESS-TOKEN": "execd-token"},
}
self.endpoint_responses = (
list(endpoint_responses) if endpoint_responses is not None else None
)
self.execute_lines = execute_lines or []
self.delete_status = delete_status
self.execute_raises = execute_raises
self.calls = []
async def post(self, url, headers=None, json=None, stream=False, **kwargs):
self.calls.append(("POST", url, headers, json, {"stream": stream}))
if url.endswith("/sandboxes"):
return FakeResponse(json_data=self.create_json)
if url.endswith("/code"):
if self.execute_raises is not None:
raise self.execute_raises
return _line_stream(self.execute_lines)
raise AssertionError(f"unexpected POST {url}")
async def get(self, url, headers=None, params=None, **kwargs):
self.calls.append(("GET", url, headers, None, params))
if "/endpoints/44772" in url:
if self.endpoint_responses is not None and self.endpoint_responses:
response = self.endpoint_responses.pop(0)
if isinstance(response, Exception):
raise response
if isinstance(response, FakeResponse):
return response
return FakeResponse(json_data=response)
return FakeResponse(json_data=self.endpoint_json)
if "/sandboxes/" in url:
state = self.sandbox_states.pop(0)
return FakeResponse(json_data=state)
raise AssertionError(f"unexpected GET {url}")
async def delete(self, url, headers=None, **kwargs):
self.calls.append(("DELETE", url, headers, None, None))
if not (200 <= self.delete_status < 300):
raise http_status_error(self.delete_status, url)
return FakeResponse(status_code=self.delete_status)
def test_parse_sse_lines_maps_output_result_count_and_error():
lines = [
sse({"type": "stdout", "text": "hello\n"}),
sse({"type": "stderr", "text": "warn\n"}),
sse({"type": "result", "results": {"text/plain": "4"}}),
sse({"type": "execution_count", "execution_count": 7}),
sse(
{
"type": "error",
"error": {
"ename": "ValueError",
"evalue": "bad",
"traceback": ["Traceback"],
},
}
),
]
result = OpenSandboxSandboxConfig._parse_lines(lines)
assert result.stdout == "hello\n"
assert result.stderr == "warn\n"
assert result.results == [{"text/plain": "4"}]
assert result.execution_count == 7
assert result.error == {
"name": "ValueError",
"value": "bad",
"traceback": ["Traceback"],
}
def test_parse_sse_lines_skips_non_json_and_control_lines():
lines = [
"event: message",
"not-json",
"",
sse({"type": "stdout", "text": "ok\n"}),
]
result = OpenSandboxSandboxConfig._parse_lines(lines)
assert result.stdout == "ok\n"
assert result.error is None
def test_parse_sse_lines_maps_fallback_shapes():
lines = [
"data:",
sse(["not-a-dict"]),
sse({"code": "BadRequest", "message": "nope"}),
sse({"type": "result", "text/plain": "4"}),
sse({"type": "error", "name": "RuntimeError", "text": "boom"}),
sse({"type": "execution_count", "execution_count": "8"}),
]
result = OpenSandboxSandboxConfig._parse_lines(lines)
assert result.results == [{"text/plain": "4"}]
assert result.execution_count == 8
assert result.error == {
"name": "BadRequest",
"value": "nope",
"traceback": [],
}
fallback_error = OpenSandboxSandboxConfig._parse_lines(
[sse({"type": "error", "name": "RuntimeError", "text": "boom"})]
)
assert fallback_error.error == {
"name": "RuntimeError",
"value": "boom",
"traceback": [],
}
empty_string_error = OpenSandboxSandboxConfig._parse_lines(
[
sse(
{
"type": "error",
"error": {
"ename": "",
"name": "FallbackName",
"evalue": "",
"value": "fallback value",
"traceback": [],
},
}
)
]
)
assert empty_string_error.error == {
"name": "",
"value": "",
"traceback": [],
}
def test_static_helpers_cover_defaults_and_fallbacks(monkeypatch):
def fake_secret(key):
if key == "OPEN_SANDBOX_API_KEY":
return "env-key"
if key == "OPEN_SANDBOX_API_BASE":
return TEST_API_BASE
return None
monkeypatch.setattr(
"litellm.llms.opensandbox.sandbox.transformation.get_secret_str",
fake_secret,
)
config = OpenSandboxSandboxConfig()
handle = ContainerHandle(id="osb", provider="opensandbox", domain="http://x/v1")
assert config.validate_environment() == "env-key"
assert config.validate_environment(api_key="") == ""
assert config._api_key(api_key=None, handle=handle) == "env-key"
handle._hidden_params = {"api_key": "stored-key"}
assert config._api_key(api_key=None, handle=handle) == "stored-key"
assert config._http(None) is not None
body = config._create_body(
template=None,
timeout=None,
allow_internet_access=False,
metadata=None,
env_vars=None,
resource_limits=None,
resource_requests=None,
entrypoint=None,
network_policy={"egress": [{"domain": "example.com"}]},
secure_access=True,
)
assert body["networkPolicy"] == {"egress": [{"domain": "example.com"}]}
assert body["secureAccess"] is True
other_body = config._create_body(
template=None,
timeout=None,
allow_internet_access=False,
metadata=None,
env_vars=None,
resource_limits=None,
resource_requests=None,
entrypoint=None,
network_policy=None,
secure_access=False,
)
assert body["resourceLimits"] is not other_body["resourceLimits"]
assert config._sandbox_state(None) is None
assert config._sandbox_state({"status": "Running"}) is None
assert config._as_str_dict(None) == {}
assert config._endpoint_base_url("http://execd.local", "https://api/v1") == (
"http://execd.local"
)
assert config._api_base(None) == TEST_API_BASE
assert config._api_base("https://direct.test/v1/") == "https://direct.test/v1"
assert config._as_int("9") == 9
assert config._as_int("nope") is None
assert config._as_int(None) is None
assert isinstance(
ProviderConfigManager.get_provider_sandbox_config("opensandbox"),
OpenSandboxSandboxConfig,
)
def test_api_base_requires_kwarg_or_env(monkeypatch):
monkeypatch.setattr(
"litellm.llms.opensandbox.sandbox.transformation.get_secret_str",
lambda key: None,
)
with pytest.raises(ValueError, match="api_base is required"):
OpenSandboxSandboxConfig._api_base(None)
@pytest.mark.asyncio
async def test_create_posts_default_body_and_omits_empty_api_key():
client = FakeHTTPClient()
handle = await OpenSandboxSandboxConfig().acreate_sandbox(
api_key="", api_base=TEST_API_BASE, client=client
)
method, url, headers, body, _ = client.calls[0]
assert method == "POST"
assert url == f"{TEST_API_BASE}/sandboxes"
assert "OPEN-SANDBOX-API-KEY" not in headers
assert body["image"] == {"uri": OPEN_SANDBOX_DEFAULT_TEMPLATE}
assert body["entrypoint"] == ["/opt/code-interpreter/code-interpreter.sh"]
assert body["timeout"] == 300
assert body["resourceLimits"] == {"cpu": "1", "memory": "2Gi"}
assert body["networkPolicy"] == {"defaultAction": "deny", "egress": []}
assert handle.id == "osb_123"
assert handle._hidden_params["execd_endpoint"] == "execd.local:44772"
@pytest.mark.asyncio
async def test_create_can_opt_into_internet_access():
client = FakeHTTPClient()
await OpenSandboxSandboxConfig().acreate_sandbox(
api_key="",
api_base=TEST_API_BASE,
allow_internet_access=True,
client=client,
)
_, _, _, body, _ = client.calls[0]
assert "networkPolicy" not in body
@pytest.mark.asyncio
async def test_create_custom_options_poll_and_endpoint_resolution():
client = FakeHTTPClient(
create_json={
"id": "osb_pending",
"status": {"state": "Pending"},
"createdAt": "2026-01-01T00:00:00Z",
"entrypoint": ["/bin/sh"],
},
sandbox_states=[
{
"id": "osb_pending",
"status": {"state": "Running"},
"createdAt": "2026-01-01T00:00:00Z",
"entrypoint": ["/bin/sh"],
}
],
)
handle = await OpenSandboxSandboxConfig().acreate_sandbox(
template="custom/image:latest",
timeout=600,
allow_internet_access=False,
api_key="osb-key",
api_base="https://sandbox.example/v1",
metadata={"suite": "unit"},
env_vars={"PYTHONUNBUFFERED": "1"},
resource_limits={"cpu": "500m", "memory": "512Mi"},
resource_requests={"cpu": "250m", "memory": "256Mi"},
entrypoint=["/bin/sh", "-lc", "sleep 3600"],
use_server_proxy=True,
client=client,
)
_, create_url, create_headers, body, _ = client.calls[0]
_, poll_url, poll_headers, _, _ = client.calls[1]
_, endpoint_url, endpoint_headers, _, endpoint_params = client.calls[2]
assert create_url == "https://sandbox.example/v1/sandboxes"
assert create_headers["OPEN-SANDBOX-API-KEY"] == "osb-key"
assert body["image"] == {"uri": "custom/image:latest"}
assert body["entrypoint"] == ["/bin/sh", "-lc", "sleep 3600"]
assert body["metadata"] == {"suite": "unit"}
assert body["env"] == {"PYTHONUNBUFFERED": "1"}
assert body["resourceLimits"] == {"cpu": "500m", "memory": "512Mi"}
assert body["resourceRequests"] == {"cpu": "250m", "memory": "256Mi"}
assert body["networkPolicy"] == {"defaultAction": "deny", "egress": []}
assert poll_url == "https://sandbox.example/v1/sandboxes/osb_pending"
assert poll_headers["OPEN-SANDBOX-API-KEY"] == "osb-key"
assert endpoint_url.endswith("/sandboxes/osb_pending/endpoints/44772")
assert endpoint_headers["OPEN-SANDBOX-API-KEY"] == "osb-key"
assert endpoint_params == {"use_server_proxy": True}
assert handle.id == "osb_pending"
@pytest.mark.asyncio
async def test_create_waits_across_pending_state(monkeypatch):
client = FakeHTTPClient(
create_json={
"id": "osb_pending",
"status": {"state": "Pending"},
"createdAt": "2026-01-01T00:00:00Z",
},
sandbox_states=[
{"id": "osb_pending", "status": {"state": "Pending"}},
{"id": "osb_pending", "status": {"state": "Running"}},
],
)
sleeps = []
async def fake_sleep(interval):
sleeps.append(interval)
monkeypatch.setattr(
"litellm.llms.opensandbox.sandbox.transformation.asyncio.sleep", fake_sleep
)
handle = await OpenSandboxSandboxConfig().acreate_sandbox(
api_key="",
api_base=TEST_API_BASE,
ready_timeout=1,
poll_interval=0.01,
client=client,
)
assert handle.id == "osb_pending"
assert sleeps == [0.01]
@pytest.mark.asyncio
async def test_create_raises_for_terminal_state():
client = FakeHTTPClient(
create_json={"id": "osb_failed", "status": {"state": "Pending"}},
sandbox_states=[
{"id": "osb_failed", "status": {"state": "Failed"}},
],
)
with pytest.raises(ValueError, match="entered Failed"):
await OpenSandboxSandboxConfig().acreate_sandbox(
api_key="", api_base=TEST_API_BASE, client=client
)
@pytest.mark.asyncio
async def test_create_times_out_waiting_for_running():
client = FakeHTTPClient(
create_json={"id": "osb_slow", "status": {"state": "Pending"}},
sandbox_states=[
{"id": "osb_slow", "status": {"state": "Pending"}},
],
)
with pytest.raises(TimeoutError, match="was not Running"):
await OpenSandboxSandboxConfig().acreate_sandbox(
api_key="",
api_base=TEST_API_BASE,
ready_timeout=0,
poll_interval=0,
client=client,
)
@pytest.mark.asyncio
async def test_create_waits_for_endpoint_resolution(monkeypatch):
client = FakeHTTPClient(
endpoint_responses=[
http_status_error(404, f"{TEST_API_BASE}/sandboxes/osb_123"),
{
"endpoint": "execd.local:44772",
"headers": {"X-EXECD-ACCESS-TOKEN": "execd-token"},
},
],
)
sleeps = []
async def fake_sleep(interval):
sleeps.append(interval)
monkeypatch.setattr(
"litellm.llms.opensandbox.sandbox.transformation.asyncio.sleep", fake_sleep
)
handle = await OpenSandboxSandboxConfig().acreate_sandbox(
api_key="",
api_base=TEST_API_BASE,
ready_timeout=1,
poll_interval=0.01,
client=client,
)
endpoint_calls = [call for call in client.calls if "/endpoints/44772" in call[1]]
assert handle._hidden_params["execd_endpoint"] == "execd.local:44772"
assert len(endpoint_calls) == 2
assert sleeps == [0.01]
@pytest.mark.asyncio
async def test_create_raises_when_endpoint_is_missing():
client = FakeHTTPClient(endpoint_json={"headers": {"X": "y"}})
with pytest.raises(TimeoutError, match=r"execd endpoint.*not ready"):
await OpenSandboxSandboxConfig().acreate_sandbox(
api_key="", api_base=TEST_API_BASE, ready_timeout=0, client=client
)
@pytest.mark.asyncio
async def test_create_reraises_non_404_endpoint_error():
client = FakeHTTPClient(endpoint_responses=[http_status_error(500)])
with pytest.raises(httpx.HTTPStatusError):
await OpenSandboxSandboxConfig().acreate_sandbox(
api_key="", api_base=TEST_API_BASE, client=client
)
@pytest.mark.asyncio
async def test_run_code_resolves_bare_id_and_posts_sse_request():
client = FakeHTTPClient(
execute_lines=[
sse({"type": "stdout", "text": "42\n"}),
]
)
result = await OpenSandboxSandboxConfig().arun_code(
container="osb_bare",
code="print(6*7)",
language="python",
api_key="",
api_base="http://sandbox.local/v1",
client=client,
)
endpoint_call = client.calls[0]
run_call = client.calls[1]
assert endpoint_call[0] == "GET"
assert (
endpoint_call[1] == "http://sandbox.local/v1/sandboxes/osb_bare/endpoints/44772"
)
assert run_call[0] == "POST"
assert run_call[1] == "http://execd.local:44772/code"
assert run_call[2]["X-EXECD-ACCESS-TOKEN"] == "execd-token"
assert run_call[3] == {
"code": "print(6*7)",
"context": {"language": "python"},
}
assert run_call[4] == {"stream": True}
assert result.stdout == "42\n"
@pytest.mark.asyncio
async def test_run_code_uses_https_for_scheme_less_endpoint_when_api_base_is_https():
client = FakeHTTPClient()
handle = ContainerHandle(
id="osb_https", provider="opensandbox", domain="https://sandbox.example/v1"
)
handle._hidden_params = {
"execd_endpoint": "execd.example/route/44772",
"execd_headers": {},
}
await OpenSandboxSandboxConfig().arun_code(
container=handle, code="print(1)", client=client
)
assert client.calls[0][1] == "https://execd.example/route/44772/code"
@pytest.mark.asyncio
async def test_run_code_aborts_on_output_over_cap():
client = FakeHTTPClient(execute_lines=["x" * (MAX_OUTPUT_BYTES + 1)])
handle = ContainerHandle(id="osb_big", provider="opensandbox", domain="http://x/v1")
handle._hidden_params = {"execd_endpoint": "execd.local:44772", "execd_headers": {}}
with pytest.raises(ValueError, match="exceeded"):
await OpenSandboxSandboxConfig().arun_code(
container=handle, code="print('x')", client=client
)
@pytest.mark.asyncio
async def test_delete_returns_false_on_404():
client = FakeHTTPClient(delete_status=404)
ok = await OpenSandboxSandboxConfig().adelete_sandbox(
container="osb_gone",
api_key="",
api_base="http://sandbox.local/v1",
client=client,
)
assert ok is False
@pytest.mark.asyncio
async def test_delete_reraises_non_404_http_error():
client = FakeHTTPClient(delete_status=500)
with pytest.raises(httpx.HTTPStatusError):
await OpenSandboxSandboxConfig().adelete_sandbox(
container="osb_err",
api_key="",
api_base="http://sandbox.local/v1",
client=client,
)
@pytest.mark.asyncio
async def test_public_lifecycle_create_run_delete():
client = FakeHTTPClient(
execute_lines=[
sse({"type": "stdout", "text": "42\n"}),
]
)
container = await litellm.acreate_sandbox(
provider="opensandbox", api_key="", api_base=TEST_API_BASE, client=client
)
result = await litellm.arun_code(
provider="opensandbox",
container=container,
code="print(6*7)",
api_key="",
client=client,
)
ok = await litellm.adelete_sandbox(
provider="opensandbox",
container=container,
api_key="",
client=client,
)
assert container.id == "osb_123"
assert result.stdout == "42\n"
assert ok is True
@pytest.mark.asyncio
async def test_code_interpreter_tool_deletes_even_when_run_raises():
client = FakeHTTPClient(execute_raises=RuntimeError("boom"))
with pytest.raises(RuntimeError, match="boom"):
await litellm.acode_interpreter_tool(
provider="opensandbox",
code="1/0",
api_key="",
api_base=TEST_API_BASE,
client=client,
)
assert [call[0] for call in client.calls] == ["POST", "GET", "POST", "DELETE"]
assert client.calls[0][1].endswith("/sandboxes")
assert client.calls[1][1].endswith("/endpoints/44772")
assert client.calls[2][1].endswith("/code")
assert client.calls[3][1].endswith("/sandboxes/osb_123")