from __future__ import annotations import asyncio import unittest from datetime import UTC, datetime from types import SimpleNamespace from unittest.mock import AsyncMock, patch from models import AssistantConfig from pipecat.frames.frames import ( LLMContextFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMTextFrame, MetricsFrame, ) from pipecat.metrics.metrics import LLMTokenUsage, LLMUsageMetricsData from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from services.brains.base import BaseBrain, BrainSpec from services.openai_responses.protocol import ResponsesAPIError, parse_request from services.openai_responses.runner import ( ResponsesClientToolPort, ResponsesTextRunner, _context_snapshot, _usage_payload, ) from services.openai_responses.state import ( _client_tool_results, _validate_tool_outputs, items_to_messages, ) from routes.openai_responses import _stream_response class ResponsesProtocolTests(unittest.TestCase): def test_first_turn_requires_model_and_normalizes_text(self): request = parse_request( { "model": "assistant:asst_1", "input": "你好", "x_interactive_media": {"variables": {"customer": "王女士"}}, } ) self.assertEqual(request.model, "assistant:asst_1") self.assertFalse(request.stream) self.assertEqual(request.dynamic_variables, {"customer": "王女士"}) self.assertEqual(request.input_items[0]["role"], "user") self.assertEqual(request.input_items[0]["content"][0]["text"], "你好") def test_continuation_can_inherit_model(self): request = parse_request( { "previous_response_id": "resp_parent", "input": [ { "type": "function_call_output", "call_id": "call_1", "output": {"ok": True}, } ], "stream": True, } ) self.assertIsNone(request.model) self.assertTrue(request.stream) self.assertEqual(request.input_items[0]["output"], '{"ok": true}') def test_rejects_stateless_and_assistant_owned_options(self): with self.assertRaisesRegex(ResponsesAPIError, "store must be true"): parse_request( {"model": "assistant:asst_1", "input": "hi", "store": False} ) with self.assertRaisesRegex(ResponsesAPIError, "store must be true"): parse_request( {"model": "assistant:asst_1", "input": "hi", "store": None} ) with self.assertRaisesRegex(ResponsesAPIError, "owned"): parse_request( { "model": "assistant:asst_1", "input": "hi", "instructions": "override", } ) def test_rejects_unknown_top_level_parameters(self): with self.assertRaises(ResponsesAPIError) as raised: parse_request( {"model": "assistant:asst_1", "input": "hi", "metadata": {}} ) self.assertEqual(raised.exception.code, "unsupported_parameter") self.assertEqual(raised.exception.param, "metadata") def test_rejects_images_and_unknown_items(self): with self.assertRaises(ResponsesAPIError) as raised: parse_request( { "model": "assistant:asst_1", "input": [ { "type": "message", "role": "user", "content": [{"type": "input_image", "image_url": "x"}], } ], } ) self.assertEqual(raised.exception.code, "invalid_request_error") with self.assertRaises(ResponsesAPIError) as raised: parse_request( { "model": "assistant:asst_1", "input": [ { "type": "message", "id": "caller_owned_id", "role": "user", "content": "hello", } ], } ) self.assertEqual(raised.exception.code, "unsupported_parameter") self.assertEqual(raised.exception.param, "input[0].id") class ResponsesStateTests(unittest.TestCase): def test_context_items_become_llm_messages(self): messages = items_to_messages( [ { "type": "message", "role": "user", "content": [{"type": "input_text", "text": "查订单"}], }, { "type": "function_call", "call_id": "call_1", "name": "lookup_order", "arguments": '{"id":"A1"}', }, { "type": "function_call_output", "call_id": "call_1", "output": '{"status":"ok"}', }, ] ) self.assertEqual(messages[0], {"role": "user", "content": "查订单"}) self.assertEqual(messages[1]["tool_calls"][0]["id"], "call_1") self.assertEqual(messages[2]["role"], "tool") def test_tool_output_must_resolve_parent_pending_call(self): parent = SimpleNamespace( output_items=[{"type": "function_call", "call_id": "call_1"}], input_items=[], ) _validate_tool_outputs( [parent], [ { "type": "function_call_output", "call_id": "call_1", "output": "ok", } ], ) with self.assertRaises(ResponsesAPIError) as raised: _validate_tool_outputs( [parent], [ { "type": "function_call_output", "call_id": "call_other", "output": "ok", } ], ) self.assertEqual(raised.exception.code, "invalid_tool_output") def test_provider_usage_is_combined_without_estimation(self): usage = _usage_payload( [ LLMTokenUsage( prompt_tokens=10, completion_tokens=4, total_tokens=14, cache_read_input_tokens=3, reasoning_tokens=2, ), LLMTokenUsage( prompt_tokens=5, completion_tokens=2, total_tokens=7, ), ] ) self.assertEqual( usage, { "input_tokens": 15, "input_tokens_details": {"cached_tokens": 3}, "output_tokens": 6, "output_tokens_details": {"reasoning_tokens": 2}, "total_tokens": 21, }, ) self.assertIsNone(_usage_payload([])) def test_context_snapshot_aligns_provider_and_public_call_ids(self): context = SimpleNamespace( get_messages=lambda: [ { "role": "assistant", "tool_calls": [ { "id": "provider_call", "type": "function", "function": {"name": "show_message", "arguments": "{}"}, } ], } ] ) snapshot = _context_snapshot( context, function_calls=[ { "call_id": "call_public", "name": "show_message", "arguments": "{}", } ], ) self.assertEqual(snapshot[0]["tool_calls"][0]["id"], "call_public") def test_context_snapshot_aligns_repeated_tool_names_by_arguments(self): context = SimpleNamespace( get_messages=lambda: [ { "role": "assistant", "tool_calls": [ { "id": "provider_a", "type": "function", "function": { "name": "lookup_order", "arguments": '{"id":"A"}', }, }, { "id": "provider_b", "type": "function", "function": { "name": "lookup_order", "arguments": '{"id":"B"}', }, }, ], } ] ) snapshot = _context_snapshot( context, function_calls=[ { "call_id": "call_a", "name": "lookup_order", "arguments": '{"id": "A"}', }, { "call_id": "call_b", "name": "lookup_order", "arguments": '{"id": "B"}', }, ], ) self.assertEqual( [call["id"] for call in snapshot[0]["tool_calls"]], ["call_a", "call_b"], ) def test_tool_output_cannot_be_replayed(self): resolved = SimpleNamespace( output_items=[{"type": "function_call", "call_id": "call_1"}], input_items=[ { "type": "function_call_output", "call_id": "call_1", "output": "ok", } ], ) with self.assertRaises(ResponsesAPIError) as raised: _validate_tool_outputs( [resolved], [ { "type": "function_call_output", "call_id": "call_1", "output": "again", } ], ) self.assertEqual(raised.exception.code, "invalid_tool_output") def test_tool_results_keep_original_call_order(self): parent = SimpleNamespace( output_items=[ { "type": "function_call", "call_id": "call_a", "name": "lookup_order", "arguments": '{"id":"A"}', }, { "type": "function_call", "call_id": "call_b", "name": "lookup_order", "arguments": '{"id":"B"}', }, ], input_items=[], ) results = _client_tool_results( [parent], [ { "type": "function_call_output", "call_id": "call_b", "output": '{"status":"ok","data":"B"}', }, { "type": "function_call_output", "call_id": "call_a", "output": '{"status":"ok","data":"A"}', }, ], ) self.assertEqual( [result["call_id"] for result in results], ["call_a", "call_b"], ) self.assertEqual( [result["arguments"] for result in results], ['{"id":"A"}', '{"id":"B"}'], ) class ResponsesClientToolTests(unittest.IsolatedAsyncioTestCase): async def test_client_tool_is_exposed_without_waiting_for_browser(self): port = ResponsesClientToolPort() task = asyncio.create_task( port.call("show_message", {"message": "确认"}, timeout_seconds=3) ) await port.called.wait() self.assertEqual(port.calls[0]["type"], "function_call") self.assertEqual(port.calls[0]["name"], "show_message") self.assertIn('"message": "确认"', port.calls[0]["arguments"]) task.cancel() await asyncio.gather(task, return_exceptions=True) async def test_persisted_result_resumes_deterministic_client_tool(self): port = ResponsesClientToolPort( [ { "name": "show_message", "result": {"status": "ok", "data": {"action": "confirmed"}}, } ] ) result = await port.call( "show_message", {"message": "确认"}, timeout_seconds=3, ) self.assertEqual(result["data"]["action"], "confirmed") self.assertFalse(port.called.is_set()) async def test_same_name_results_are_matched_by_arguments(self): port = ResponsesClientToolPort( [ { "name": "lookup_order", "arguments": '{"id":"A"}', "result": {"status": "ok", "data": "A"}, }, { "name": "lookup_order", "arguments": '{"id":"B"}', "result": {"status": "ok", "data": "B"}, }, ] ) second = await port.call( "lookup_order", {"id": "B"}, timeout_seconds=3, ) first = await port.call( "lookup_order", {"id": "A"}, timeout_seconds=3, ) self.assertEqual(second["data"], "B") self.assertEqual(first["data"], "A") class _FakeTextLLM(FrameProcessor): async def process_frame(self, frame, direction: FrameDirection): await super().process_frame(frame, direction) if not isinstance(frame, LLMContextFrame): await self.push_frame(frame, direction) return await self.push_frame(LLMFullResponseStartFrame()) await self.push_frame(LLMTextFrame("状态已恢复")) await self.push_frame( MetricsFrame( data=[ LLMUsageMetricsData( processor="fake-llm", model="fake-model", value=LLMTokenUsage( prompt_tokens=8, completion_tokens=3, total_tokens=11, ), ) ] ) ) await self.push_frame(LLMFullResponseEndFrame()) class _FakeBrain(BaseBrain): spec = BrainSpec( type="dify", supported_runtime_modes=frozenset({"pipeline"}), owns_context=False, ) def build_llm(self, _cfg, _context): return _FakeTextLLM() class ResponsesTextRunnerTests(unittest.IsolatedAsyncioTestCase): async def test_transportless_runner_streams_and_finishes_text(self): deltas: list[str] = [] async def on_delta(value: str) -> None: deltas.append(value) with patch( "services.openai_responses.runner.build_brain", return_value=_FakeBrain(), ): result = await asyncio.wait_for( ResponsesTextRunner().run( AssistantConfig( type="dify", runtimeMode="pipeline", greeting="", ), history_messages=[{"role": "user", "content": "上一轮"}], current_messages=[{"role": "user", "content": "继续"}], on_delta=on_delta, ), timeout=3, ) self.assertEqual(result.text, "状态已恢复") self.assertEqual(deltas, ["状态已恢复"]) self.assertEqual(result.usage["total_tokens"], 11) class ResponsesStreamingTests(unittest.IsolatedAsyncioTestCase): async def test_stream_uses_responses_event_lifecycle(self): now = datetime.now(UTC) row = SimpleNamespace( id="resp_1", api_key_id="rtkey_1", assistant_id="asst_1", previous_response_id=None, status="completed", error=None, output_items=[ { "id": "msg_1", "type": "message", "status": "completed", "role": "assistant", "content": [ { "type": "output_text", "text": "完成", "annotations": [], } ], } ], created_at=now, completed_at=now, ) prepared = SimpleNamespace(row=row) async def run_prepared(_prepared, *, on_delta, text_item_id): self.assertEqual(text_item_id, "msg_1") await on_delta("完") await on_delta("成") return row with patch( "routes.openai_responses._run_prepared", new=AsyncMock(side_effect=run_prepared), ): chunks = [ chunk.decode("utf-8") async for chunk in _stream_response(prepared, text_item_id="msg_1") ] body = "".join(chunks) self.assertIn("event: response.created", body) self.assertIn("event: response.output_text.delta", body) self.assertIn("event: response.output_text.done", body) self.assertIn("event: response.completed", body) if __name__ == "__main__": unittest.main()