from __future__ import annotations from typing import Any import msgspec from plyngent.agent.responses_bridge import ( chat_messages_to_responses_input, chat_param_to_responses_kwargs, reasoning_summary_text, response_to_assistant_message, response_to_chat_completion, responses_status_to_finish_reason, tool_items_to_response_tools, ) from plyngent.lmproto.openai.model import Response, ResponseFunctionToolCallOutput from plyngent.lmproto.openai_compatible.model import ( AssistantChatMessage, AssistantFunctionTool, AssistantFunctionToolCall, ChatCompletionsParam, SystemChatMessage, ToolChatMessage, ToolFunction, ToolFunctionItem, UserChatMessage, ) def test_tool_items_to_response_tools() -> None: items = [ ToolFunctionItem( function=ToolFunction( name="read_file", description="Read a file", parameters={"type": "object", "properties": {"path": {"type": "string"}}}, ) ) ] tools = tool_items_to_response_tools(items) assert len(tools) == 1 assert tools[0].name == "read_file" def test_chat_messages_to_input_and_instructions() -> None: messages = [ SystemChatMessage(content="You are helpful."), UserChatMessage(content="hi"), AssistantChatMessage( content=None, tool_calls=[ AssistantFunctionToolCall( id="call_1", function=AssistantFunctionTool(name="read_file", arguments='{"path":"a"}'), ) ], ), ToolChatMessage(content="file body", tool_call_id="call_1"), ] instructions, items = chat_messages_to_responses_input(messages) assert instructions == "You are helpful." assert len(items) == 3 # user, function_call, function_call_output def test_parallel_tool_outputs_interleaved_after_calls() -> None: """Regression: DeepSeek Responses rejects call, call, output, output batches. Each ``function_call_output`` must directly follow its ``function_call`` so strict/adjacency-matching servers (DeepSeek) accept parallel tool rounds. """ messages = [ UserChatMessage(content="run both"), AssistantChatMessage( content=None, tool_calls=[ AssistantFunctionToolCall( id="call_01_LPBuUG3KTaMDIwUoLwTE9127", function=AssistantFunctionTool(name="read_file", arguments='{"path":"a"}'), ), AssistantFunctionToolCall( id="call_02_ABCD", function=AssistantFunctionTool(name="regex_files", arguments='{"pattern":"todo"}'), ), ], ), ToolChatMessage(content="file body", tool_call_id="call_01_LPBuUG3KTaMDIwUoLwTE9127"), ToolChatMessage(content="grep hit", tool_call_id="call_02_ABCD"), ] _, items = chat_messages_to_responses_input(messages) types = [item.get("type") if isinstance(item, dict) else item.__struct_config__.tag for item in items] assert types == [ "message", "function_call", "function_call_output", "function_call", "function_call_output", ] # Each output is paired with its own call, in stream order. call_a = items[1] out_a = items[2] call_b = items[3] out_b = items[4] assert isinstance(call_a, dict) and isinstance(call_b, dict) assert isinstance(out_a, ResponseFunctionToolCallOutput) assert isinstance(out_b, ResponseFunctionToolCallOutput) assert out_a.call_id == call_a["call_id"] assert out_b.call_id == call_b["call_id"] def test_response_to_assistant_with_tools() -> None: raw = { "id": "resp_1", "object": "response", "created_at": 1, "model": "gpt-test", "status": "completed", "output": [ { "id": "msg_1", "type": "message", "role": "assistant", "status": "completed", "content": [{"type": "output_text", "text": "done", "annotations": []}], }, { "type": "function_call", "call_id": "call_9", "name": "add", "arguments": '{"a":1}', "status": "completed", }, ], "usage": {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15}, } response = msgspec.convert(raw, Response) from msgspec import UNSET assistant = response_to_assistant_message(response) assert assistant.content == "done" assert assistant.tool_calls is not UNSET assert isinstance(assistant.tool_calls, list) call0 = assistant.tool_calls[0] assert isinstance(call0, AssistantFunctionToolCall) assert call0.id == "call_9" assert call0.function.name == "add" completion = response_to_chat_completion(response) assert completion.choices[0].message.content == "done" assert isinstance(completion.usage, dict) assert completion.usage["input_tokens"] == 10 assert completion.choices[0].finish_reason == "tool_calls" def _reasoning_response( *, output_reasoning: bool = False, top_level: dict[str, Any] | None = None, ) -> Response: output: list[dict[str, Any]] = [] if output_reasoning: output.append( { "id": "reason_1", "type": "reasoning", "summary": [{"type": "summary_text", "text": "openai-style reasoning"}], } ) return Response( id="resp_r", created_at=1, model="deepseek-flash", output=output, reasoning=top_level, ) def test_reasoning_summary_text_deepseek_top_level() -> None: """DeepSeek Responses returns reasoning in the top-level ``reasoning`` field.""" response = _reasoning_response(top_level={"content": [{"type": "reasoning_content", "text": "think step by step"}]}) assert reasoning_summary_text(response) == "think step by step" assistant = response_to_assistant_message(response) assert assistant.reasoning_content == "think step by step" def test_reasoning_summary_text_prefers_top_level() -> None: response = _reasoning_response( output_reasoning=True, top_level={"summary": [{"type": "summary_text", "text": "top-level reasoning"}]}, ) assert reasoning_summary_text(response) == "top-level reasoning" def test_reasoning_summary_text_falls_back_to_output_items() -> None: response = _reasoning_response(output_reasoning=True) assert reasoning_summary_text(response) == "openai-style reasoning" def test_reasoning_summary_text_empty_when_absent() -> None: assert reasoning_summary_text(_reasoning_response()) == "" def test_responses_status_to_finish_reason_incomplete() -> None: body = { "id": "resp_i", "object": "response", "created_at": 1, "model": "gpt-test", "status": "incomplete", "incomplete_details": {"reason": "max_output_tokens"}, "output": [ { "id": "msg_1", "type": "message", "role": "assistant", "status": "incomplete", "content": [{"type": "output_text", "text": "partial", "annotations": []}], } ], } resp = msgspec.convert(body, Response) assert responses_status_to_finish_reason(resp, has_tool_calls=False) == "length" completion = response_to_chat_completion(resp) assert completion.choices[0].finish_reason == "length" def test_chat_param_to_responses_kwargs() -> None: param = ChatCompletionsParam( model="gpt-test", messages=[SystemChatMessage(content="sys"), UserChatMessage(content="hi")], tools=[ToolFunctionItem(function=ToolFunction(name="t", parameters={"type": "object"}))], temperature=0.2, ) kwargs = chat_param_to_responses_kwargs(param) assert kwargs["model"] == "gpt-test" assert kwargs["instructions"] == "sys" assert kwargs["store"] is False assert kwargs["temperature"] == 0.2 assert len(kwargs["tools"]) == 1 # No effort configured → no reasoning block. assert "reasoning" not in kwargs def test_chat_param_to_responses_kwargs_reasoning_effort() -> None: param = ChatCompletionsParam( model="gpt-test", messages=[UserChatMessage(content="hi")], reasoning_effort="xhigh", ) kwargs = chat_param_to_responses_kwargs(param) assert kwargs["reasoning"].effort == "xhigh" def test_provider_tools_merged_after_local_functions() -> None: param = ChatCompletionsParam( model="gpt-test", messages=[UserChatMessage(content="hi")], tools=[ToolFunctionItem(function=ToolFunction(name="local_tool", parameters={"type": "object"}))], ) kwargs = chat_param_to_responses_kwargs( param, provider_tools=[{"type": "web_search"}, {"type": "file_search", "vector_store_ids": ["vs_1"]}], ) tools = kwargs["tools"] assert len(tools) == 3 assert tools[0].name == "local_tool" assert tools[1]["type"] == "web_search" assert tools[2]["vector_store_ids"] == ["vs_1"]