Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
118 changes: 118 additions & 0 deletions tests/sdk/llm/test_responses_parsing_and_kwargs.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,118 @@
from unittest.mock import patch

from litellm.types.llms.openai import ResponseAPIUsage, ResponsesAPIResponse
from openai.types.responses.response_function_tool_call import ResponseFunctionToolCall
from openai.types.responses.response_output_message import ResponseOutputMessage
from openai.types.responses.response_output_text import ResponseOutputText
from openai.types.responses.response_reasoning_item import (
ResponseReasoningItem,
Summary,
)

from openhands.sdk.llm.llm import LLM
from openhands.sdk.llm.message import Message, ReasoningItemModel, TextContent


def build_responses_message_output(texts: list[str]) -> ResponseOutputMessage:
parts = [
ResponseOutputText(type="output_text", text=t, annotations=[]) for t in texts
]
# Bypass stricter static type expectations in test context; runtime is fine
return ResponseOutputMessage.model_construct(
id="m1",
type="message",
role="assistant",
status="completed",
content=parts, # type: ignore[arg-type]
)


def test_from_llm_responses_output_parsing():
# Build typed Responses output: assistant message text + function call + reasoning
msg = build_responses_message_output(["Hello", "World"]) # concatenated
fc = ResponseFunctionToolCall(
type="function_call", name="do", arguments="{}", call_id="fc_1", id="fc_1"
)
reasoning = ResponseReasoningItem(
id="rid",
type="reasoning",
summary=[
Summary(type="summary_text", text="sum1"),
Summary(type="summary_text", text="sum2"),
],
content=None,
encrypted_content=None,
status="completed",
)

m = Message.from_llm_responses_output([msg, fc, reasoning])
# Assistant text joined
assert m.role == "assistant"
assert [c.text for c in m.content if isinstance(c, TextContent)] == ["Hello\nWorld"]
# Tool call normalized
assert m.tool_calls and m.tool_calls[0].name == "do"
# Reasoning mapped
assert isinstance(m.responses_reasoning_item, ReasoningItemModel)
assert m.responses_reasoning_item.summary == ["sum1", "sum2"]


def test_normalize_responses_kwargs_policy():
llm = LLM(model="gpt-5-mini")
# Use a model that is explicitly Responses-capable per model_features

# enable encrypted reasoning and set max_output_tokens to test passthrough
llm.enable_encrypted_reasoning = True
llm.max_output_tokens = 128

out = llm._normalize_responses_kwargs(
{"temperature": 0.3}, include=["text.output_text"], store=None
)
# Temperature forced to 1.0 for Responses path
assert out["temperature"] == 1.0
assert out["tool_choice"] == "auto"
# include should contain original and encrypted_content
assert set(out["include"]) >= {"text.output_text", "reasoning.encrypted_content"}
# store default to False when None passed
assert out["store"] is False
# reasoning config defaulted
r = out["reasoning"]
assert r["effort"] in {"low", "medium", "high", "none"}
assert r["summary"] == "detailed"
# max_output_tokens preserved
assert out["max_output_tokens"] == 128


@patch("openhands.sdk.llm.llm.litellm_responses")
def test_llm_responses_end_to_end(mock_responses_call):
# Configure LLM
llm = LLM(model="gpt-5-mini")
# messages: system + user
sys = Message(role="system", content=[TextContent(text="inst")])
user = Message(role="user", content=[TextContent(text="hi")])

# Build typed ResponsesAPIResponse with usage
msg = build_responses_message_output(["ok"])
usage = ResponseAPIUsage(input_tokens=10, output_tokens=5, total_tokens=15)
resp = ResponsesAPIResponse(
id="r1",
created_at=0,
output=[msg],
parallel_tool_calls=False,
tool_choice="auto",
top_p=None,
tools=[],
usage=usage,
instructions="inst",
status="completed",
)

mock_responses_call.return_value = resp

result = llm.responses([sys, user])
# Returned message is assistant with text
assert result.message.role == "assistant"
assert [c.text for c in result.message.content if isinstance(c, TextContent)] == [
"ok"
]
# Telemetry should have recorded usage (one entry)
assert len(llm._telemetry.metrics.token_usages) == 1 # type: ignore[attr-defined]
100 changes: 100 additions & 0 deletions tests/sdk/llm/test_responses_serialization.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,100 @@
from openhands.sdk.llm.message import (
ImageContent,
Message,
MessageToolCall,
ReasoningItemModel,
TextContent,
)


def test_system_to_responses_value_instructions_concat():
m1 = Message(role="system", content=[TextContent(text="A"), TextContent(text="B")])
m2 = Message(role="system", content=[TextContent(text="C")])

# system messages become instructions string, concatenated with separators
from openhands.sdk.llm.llm import LLM

llm = LLM(model="gpt-5-mini")
instr, inputs = llm.format_messages_for_responses([m1, m2])
assert instr == "A\nB\n\n---\n\nC"
assert inputs == []


def test_user_to_responses_dict_with_and_without_vision():
m = Message(
role="user",
content=[
TextContent(text="hello"),
ImageContent(image_urls=["http://x/y.png"]),
],
)

# without vision: only input_text
items = m.to_responses_dict(vision_enabled=False)
assert len(items) == 1 and items[0]["type"] == "message"
content = items[0]["content"]
assert {c["type"] for c in content} == {"input_text"}

# with vision: include input_image
items_v = m.to_responses_dict(vision_enabled=True)
types = [c["type"] for c in items_v[0]["content"]]
assert "input_text" in types and "input_image" in types


assistant_text = "Here is the result"


def test_assistant_to_responses_dict_with_text_and_tool_calls():
# assistant prior text becomes output_text in message item
tc = MessageToolCall(id="123", name="foo", arguments="{}", origin="responses")
m = Message(
role="assistant", content=[TextContent(text=assistant_text)], tool_calls=[tc]
)

out = m.to_responses_dict(vision_enabled=False)
# Should include a message item with output_text, then function_call item
assert any(item["type"] == "message" for item in out)
msg_item = next(item for item in out if item["type"] == "message")
assert msg_item["role"] == "assistant"
assert {p["type"] for p in msg_item["content"]} == {"output_text"}

fc_items = [item for item in out if item["type"] == "function_call"]
assert len(fc_items) == 1
assert fc_items[0]["id"].startswith("fc_") and fc_items[0]["call_id"].startswith(
"fc_"
)


def test_tool_to_responses_emits_function_call_output_with_fc_prefix():
# tool result requires tool_call_id and outputs function_call_output entries
m = Message(
role="tool",
tool_call_id="abc",
name="foo",
content=[TextContent(text="result1"), TextContent(text="result2")],
)
out = m.to_responses_dict(vision_enabled=False)
assert all(item["type"] == "function_call_output" for item in out)
assert all(item["call_id"].startswith("fc_") for item in out)


def test_assistant_includes_reasoning_passthrough():
ri = ReasoningItemModel(
id="rid1",
summary=["s1", "s2"],
content=["c1"],
encrypted_content="enc",
status="completed",
)
m = Message(role="assistant", content=[], responses_reasoning_item=ri)
out = m.to_responses_dict(vision_enabled=False)

# Contains a reasoning item with exact passthrough fields
r_items = [it for it in out if it["type"] == "reasoning"]
assert len(r_items) == 1
r = r_items[0]
assert r["id"] == "rid1"
assert [s["text"] for s in r["summary"]] == ["s1", "s2"]
assert [c["text"] for c in r.get("content", [])] == ["c1"]
assert r.get("encrypted_content") == "enc"
assert r.get("status") == "completed"
31 changes: 31 additions & 0 deletions tests/sdk/tool/test_to_responses_tool.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
from openhands.sdk.tool.schema import Action, Observation
from openhands.sdk.tool.tool import ToolBase


class A(Action):
x: int


class Obs(Observation):
def to_llm_content(self): # type: ignore[override]
from openhands.sdk.llm import TextContent

return [TextContent(text="ok")]


class T(ToolBase[A, Obs]):
@classmethod
def create(cls, *args, **kwargs): # pragma: no cover
raise NotImplementedError


def test_to_responses_tool_includes_strict_and_params():
out = T(
name="t", description="d", action_type=A, observation_type=Obs
).to_responses_tool()
assert out["type"] == "function"
assert out["name"] == "t"
# description is optional in the TypedDict; access via get for type safety
assert out.get("description") in {"d", None}
assert out["strict"] is True
assert "parameters" in out and isinstance(out["parameters"], dict)
Loading