From 3294e694cb2276ad81067022624c14ab458b99dd Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Mon, 31 Aug 2026 14:15:03 +0200 Subject: [PATCH 1/4] Python: align DevUI Responses parsing Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../agent_framework_devui/_conversations.py | 218 +++++++++++++++--- .../devui/agent_framework_devui/_executor.py | 187 ++++++++++----- .../devui/agent_framework_devui/_mapper.py | 78 +++---- .../tests/devui/test_approval_validation.py | 14 +- .../devui/tests/devui/test_conversations.py | 62 +++++ .../packages/devui/tests/devui/test_mapper.py | 26 +++ .../tests/devui/test_multimodal_workflow.py | 101 +++++++- 7 files changed, 532 insertions(+), 154 deletions(-) diff --git a/python/packages/devui/agent_framework_devui/_conversations.py b/python/packages/devui/agent_framework_devui/_conversations.py index 9ad8534e184..22a42e05139 100644 --- a/python/packages/devui/agent_framework_devui/_conversations.py +++ b/python/packages/devui/agent_framework_devui/_conversations.py @@ -8,13 +8,15 @@ from __future__ import annotations +import mimetypes import time import uuid from abc import ABC, abstractmethod from collections.abc import MutableSequence from typing import Any, Literal, cast +from urllib.parse import urlparse -from agent_framework import AgentSession, Message +from agent_framework import AgentSession, Content, Message from agent_framework._workflows._checkpoint import InMemoryCheckpointStorage, WorkflowCheckpoint from openai.types.conversations import Conversation, ConversationDeletedResource from openai.types.conversations.conversation_item import ConversationItem @@ -34,6 +36,32 @@ # Checkpoint item type constants CONVERSATION_ITEM_TYPE_CHECKPOINT = "checkpoint" CONVERSATION_TYPE_CHECKPOINT_CONTAINER = "checkpoint_container" +_CANONICAL_MEDIA_TYPES = { + ".jpeg": "image/jpeg", + ".jpg": "image/jpeg", + ".m4a": "audio/mp4", + ".mp3": "audio/mpeg", +} + + +def _infer_media_type(*, filename: str | None = None, uri: str | None = None, default: str) -> str: + """Infer a canonical media type from explicit data or a filename.""" + if uri and uri.startswith("data:"): + media_type = uri[5:].split(";", 1)[0].split(",", 1)[0] + if "/" in media_type: + return media_type + + for candidate in (filename, urlparse(uri).path if uri else None): + if not candidate: + continue + candidate_lower = candidate.lower() + for suffix, canonical_media_type in _CANONICAL_MEDIA_TYPES.items(): + if candidate_lower.endswith(suffix): + return canonical_media_type + if guessed_media_type := mimetypes.guess_type(candidate)[0]: + return guessed_media_type + + return default class ConversationStore(ABC): @@ -304,17 +332,23 @@ async def add_items(self, conversation_id: str, items: list[dict[str, Any]]) -> # Convert items to Messages and add to storage chat_messages: list[Message] = [] for item in items: - # Simple conversion - assume text content for now role = item.get("role", "user") content = item.get("content", []) - first_content = cast( - dict[str, Any], - content[0] if content and isinstance(content, list) and isinstance(content[0], dict) else {}, - ) - text_obj = first_content.get("text", "") - text = text_obj if isinstance(text_obj, str) else str(text_obj) + contents: list[Content] = [] + + if isinstance(content, str): + contents.append(Content.from_text(text=content)) + elif isinstance(content, list): + for content_item in cast(list[object], content): + if isinstance(content_item, dict): + agent_content = self._to_agent_content(cast(dict[str, Any], content_item)) + if agent_content is not None: + contents.append(agent_content) + + if not contents: + raise ValueError("Conversation message did not contain any supported content") - chat_msg = Message(role=role, contents=[text]) + chat_msg = Message(role=role, contents=contents) chat_messages.append(chat_msg) # Add messages to internal storage @@ -328,9 +362,9 @@ async def add_items(self, conversation_id: str, items: list[dict[str, Any]]) -> # Convert Message contents to OpenAI TextContent format message_content: MutableSequence[OpenAIContent] = [] for content_item in msg.contents: - if content_item.type == "text": - # Extract text from TextContent object - message_content.append(TextContent(type="text", text=content_item.text or "")) + openai_content = self._to_openai_content(content_item) + if openai_content is not None: + message_content.append(openai_content) # Create Message object (concrete type from ConversationItem union) message = OpenAIMessage( @@ -390,29 +424,10 @@ async def list_items( for content in msg.contents: content_type = getattr(content, "type", None) + mapped_content = self._to_openai_content(content) - if content_type == "text": - # Text content for Message - text_value = getattr(content, "text", "") - message_contents.append(TextContent(type="text", text=text_value)) - - elif content_type == "data": - # Data content (images, files, PDFs) - uri = getattr(content, "uri", "") - media_type = getattr(content, "media_type", None) - - if media_type and media_type.startswith("image/"): - # Convert to ResponseInputImage - message_contents.append(ResponseInputImage(type="input_image", image_url=uri, detail="auto")) - else: - # Convert to ResponseInputFile - # Extract filename from URI if possible - filename = None - if media_type == "application/pdf": - filename = "document.pdf" - - message_contents.append(ResponseInputFile(type="input_file", file_url=uri, filename=filename)) - + if mapped_content is not None: + message_contents.append(mapped_content) elif content_type == "function_call": # Function call - create separate ConversationItem call_id = getattr(content, "call_id", None) @@ -532,6 +547,141 @@ async def list_items( return paginated_items, has_more + @staticmethod + def _to_agent_content(content: dict[str, Any]) -> Content | None: + """Convert one supported OpenAI conversation message part.""" + content_type = content.get("type") + if content_type in ("text", "input_text"): + text = content.get("text") + return Content.from_text(text=text) if isinstance(text, str) else None + + if content_type == "input_image": + detail = content.get("detail", "auto") + image_properties = { + "openai_content_type": "input_image", + "detail": detail if isinstance(detail, str) else "auto", + } + image_url = content.get("image_url") + if isinstance(image_url, str) and image_url: + return Content.from_uri( + uri=image_url, + media_type=_infer_media_type(uri=image_url, default="image/png"), + additional_properties=image_properties, + ) + file_id = content.get("file_id") + if isinstance(file_id, str) and file_id: + return Content.from_hosted_file( + file_id=file_id, + additional_properties=image_properties, + ) + return None + + if content_type == "input_file": + filename = content.get("filename") + filename = filename if isinstance(filename, str) else "" + detail = content.get("detail") + file_properties: dict[str, Any] = {"openai_content_type": "input_file"} + if filename: + file_properties["filename"] = filename + if isinstance(detail, str): + file_properties["detail"] = detail + + file_data = content.get("file_data") + if isinstance(file_data, str) and file_data: + media_type = _infer_media_type( + filename=filename, + uri=file_data, + default="application/octet-stream", + ) + uri = file_data if file_data.startswith("data:") else f"data:{media_type};base64,{file_data}" + return Content.from_uri( + uri=uri, + media_type=media_type, + additional_properties=file_properties, + ) + + file_url = content.get("file_url") + if isinstance(file_url, str) and file_url: + return Content.from_uri( + uri=file_url, + media_type=_infer_media_type( + filename=filename, + uri=file_url, + default="application/octet-stream", + ), + additional_properties=file_properties, + ) + + file_id = content.get("file_id") + if isinstance(file_id, str) and file_id: + return Content.from_hosted_file( + file_id=file_id, + media_type=_infer_media_type(filename=filename, default="application/octet-stream"), + name=filename or None, + additional_properties=file_properties, + ) + + return None + + @staticmethod + def _to_openai_content(content: Content) -> TextContent | ResponseInputImage | ResponseInputFile | None: + """Convert one supported Agent Framework message part.""" + content_type = content.type + if content_type == "text": + return TextContent(type="text", text=content.text or "") + + additional_properties = content.additional_properties or {} + openai_content_type = additional_properties.get("openai_content_type") + is_image = openai_content_type == "input_image" or ( + openai_content_type is None and bool(content.media_type and content.media_type.startswith("image/")) + ) + + if content_type in ("data", "uri"): + if is_image: + detail = additional_properties.get("detail", "auto") + if detail not in ("low", "high", "auto", "original"): + detail = "auto" + return ResponseInputImage( + type="input_image", + image_url=content.uri, + detail=detail, + ) + + filename = additional_properties.get("filename") + detail = additional_properties.get("detail") + if detail not in ("low", "high", "auto"): + detail = None + return ResponseInputFile( + type="input_file", + file_url=content.uri, + filename=filename if isinstance(filename, str) else None, + detail=detail, + ) + + if content_type == "hosted_file": + if is_image: + detail = additional_properties.get("detail", "auto") + if detail not in ("low", "high", "auto", "original"): + detail = "auto" + return ResponseInputImage( + type="input_image", + file_id=content.file_id, + detail=detail, + ) + + filename = content.name or additional_properties.get("filename") + detail = additional_properties.get("detail") + if detail not in ("low", "high", "auto"): + detail = None + return ResponseInputFile( + type="input_file", + file_id=content.file_id, + filename=filename if isinstance(filename, str) else None, + detail=detail, + ) + + return None + async def get_item(self, conversation_id: str, item_id: str) -> ConversationItem | None: """Get a specific conversation item by ID. diff --git a/python/packages/devui/agent_framework_devui/_executor.py b/python/packages/devui/agent_framework_devui/_executor.py index f6ea02b5cf5..2f9117fb83f 100644 --- a/python/packages/devui/agent_framework_devui/_executor.py +++ b/python/packages/devui/agent_framework_devui/_executor.py @@ -6,8 +6,10 @@ import json import logging +import mimetypes from collections.abc import AsyncGenerator from typing import Any, cast +from urllib.parse import urlparse from agent_framework import Content, SupportsAgentRun, Workflow @@ -20,6 +22,14 @@ logger = logging.getLogger(__name__) +_OPENAI_MESSAGE_ROLES = {"user", "assistant", "system", "developer"} +_CANONICAL_MEDIA_TYPES = { + ".jpeg": "image/jpeg", + ".jpg": "image/jpeg", + ".m4a": "audio/mp4", + ".mp3": "audio/mpeg", +} + def _get_event_type(event: Any) -> str | None: """Safely get the type of an event, handling both objects and dicts.""" @@ -29,12 +39,36 @@ def _get_event_type(event: Any) -> str | None: return getattr(event, "type", None) +def _infer_media_type(*, filename: str | None = None, uri: str | None = None, default: str) -> str: + """Infer a canonical media type from explicit data or a filename.""" + if uri and uri.startswith("data:"): + media_type = uri[5:].split(";", 1)[0].split(",", 1)[0] + if "/" in media_type: + return media_type + + for candidate in (filename, urlparse(uri).path if uri else None): + if not candidate: + continue + candidate_lower = candidate.lower() + for suffix, canonical_media_type in _CANONICAL_MEDIA_TYPES.items(): + if candidate_lower.endswith(suffix): + return canonical_media_type + if guessed_media_type := mimetypes.guess_type(candidate)[0]: + return guessed_media_type + + return default + + class EntityNotFoundError(Exception): """Raised when an entity is not found.""" pass +class InputConversionError(ValueError): + """Raised when OpenAI input contains no supported Agent Framework content.""" + + class AgentFrameworkExecutor: """Executor for Agent Framework entities - agents and workflows.""" @@ -592,6 +626,10 @@ async def _execute_workflow( # The workflow is already paused by ctx.request_info() in the framework # DevUI should continue yielding events even during HIL pause + except InputConversionError as e: + from .models._openai_custom import AgentFailedEvent + + yield AgentFailedEvent(error=e) except Exception as e: logger.error(f"Error in workflow execution: {e}") yield {"type": "error", "message": f"Workflow execution error: {e!s}"} @@ -610,7 +648,7 @@ def _convert_input_to_chat_message(self, input_data: Any) -> Any: """ # Import Agent Framework types try: - from agent_framework import Message, Role + from agent_framework import Message except ImportError: # Fallback to string extraction if Agent Framework not available return self._extract_user_message_fallback(input_data) @@ -622,13 +660,13 @@ def _convert_input_to_chat_message(self, input_data: Any) -> Any: # Handle OpenAI ResponseInputParam (List[ResponseInputItemParam]) if isinstance(input_data, list): input_items: Any = cast(Any, input_data) - return self._convert_openai_input_to_chat_message(input_items, Message, Role) + return self._convert_openai_input_to_chat_message(input_items, Message) # Fallback for other formats return self._extract_user_message_fallback(input_data) - def _convert_openai_input_to_chat_message(self, input_items: list[Any], Message: Any, Role: Any) -> Any: - """Convert OpenAI ResponseInputParam to Agent Framework Message. + def _convert_openai_input_to_chat_message(self, input_items: list[Any], Message: Any) -> Any: + """Convert OpenAI ResponseInputParam to Agent Framework messages. Processes text, images, files, and other content types from OpenAI format to Agent Framework Message with appropriate content objects. @@ -636,12 +674,11 @@ def _convert_openai_input_to_chat_message(self, input_items: list[Any], Message: Args: input_items: List of OpenAI ResponseInputItemParam objects (dicts or objects) Message: Message class for creating chat messages - Role: Role enum for message roles Returns: - Message with converted content + One Message, or a list of Messages when input contains multiple message items """ - contents: list[Content] = [] + messages: list[Any] = [] # Process each input item for item in input_items: @@ -649,9 +686,15 @@ def _convert_openai_input_to_chat_message(self, input_items: list[Any], Message: if isinstance(item, dict): item_dict = cast(dict[str, Any], item) item_type = item_dict.get("type") - if item_type == "message": + is_easy_message = item_type is None and "role" in item_dict and "content" in item_dict + if item_type == "message" or is_easy_message: + role_value = item_dict.get("role", "user") + if not isinstance(role_value, str) or role_value not in _OPENAI_MESSAGE_ROLES: + raise InputConversionError(f"Unsupported OpenAI input message role: {role_value!r}") + # Extract content from OpenAI message message_content = item_dict.get("content", []) + contents: list[Content] = [] # Handle both string content and list content if isinstance(message_content, str): @@ -671,71 +714,90 @@ def _convert_openai_input_to_chat_message(self, input_items: list[Any], Message: elif content_type == "input_image": image_url = content_dict.get("image_url", "") + file_id = content_dict.get("file_id") + detail = content_dict.get("detail", "auto") + image_properties = { + "openai_content_type": "input_image", + "detail": detail if isinstance(detail, str) else "auto", + } if isinstance(image_url, str) and image_url: - # Extract media type from data URI if possible - # Parse media type from data URL, fallback to image/png - if image_url.startswith("data:"): - try: - # Extract media type from data:image/jpeg;base64,... format - media_type = image_url.split(";")[0].split(":")[1] - except (IndexError, AttributeError): - logger.warning( - f"Failed to parse media type from data URL: {image_url[:30]}..." - ) - media_type = "image/png" - else: - media_type = "image/png" - contents.append(Content.from_uri(uri=image_url, media_type=media_type)) + media_type = _infer_media_type(uri=image_url, default="image/png") + contents.append( + Content.from_uri( + uri=image_url, + media_type=media_type, + additional_properties=image_properties, + ) + ) + elif isinstance(file_id, str) and file_id: + contents.append( + Content.from_hosted_file( + file_id=file_id, + additional_properties=image_properties, + ) + ) elif content_type == "input_file": # Handle file input file_data = content_dict.get("file_data") + file_id = content_dict.get("file_id") file_url = content_dict.get("file_url") filename = content_dict.get("filename", "") + detail = content_dict.get("detail") if not isinstance(filename, str): filename = "" - # Determine media type from filename - media_type = "application/octet-stream" # default - if filename: - if filename.lower().endswith(".pdf"): - media_type = "application/pdf" - elif filename.lower().endswith((".png", ".jpg", ".jpeg", ".gif")): - media_type = f"image/{filename.split('.')[-1].lower()}" - elif filename.lower().endswith(( - ".wav", - ".mp3", - ".m4a", - ".ogg", - ".flac", - ".aac", - )): - ext = filename.split(".")[-1].lower() - # Normalize extensions to match audio MIME types - media_type = "audio/mp4" if ext == "m4a" else f"audio/{ext}" - # Use file_data or file_url # Include filename in additional_properties for OpenAI/Azure file handling - additional_props: dict[str, Any] | None = ( - {"filename": filename} if filename else None - ) + file_properties: dict[str, Any] = {"openai_content_type": "input_file"} + if filename: + file_properties["filename"] = filename + if isinstance(detail, str): + file_properties["detail"] = detail + if isinstance(file_data, str) and file_data: - # Assume file_data is base64, create data URI - data_uri = f"data:{media_type};base64,{file_data}" + media_type = _infer_media_type( + filename=filename, + uri=file_data, + default="application/octet-stream", + ) + data_uri = ( + file_data + if file_data.startswith("data:") + else f"data:{media_type};base64,{file_data}" + ) contents.append( Content.from_uri( uri=data_uri, media_type=media_type, - additional_properties=additional_props, + additional_properties=file_properties, ) ) elif isinstance(file_url, str) and file_url: + media_type = _infer_media_type( + filename=filename, + uri=file_url, + default="application/octet-stream", + ) contents.append( Content.from_uri( uri=file_url, media_type=media_type, - additional_properties=additional_props, + additional_properties=file_properties, + ) + ) + elif isinstance(file_id, str) and file_id: + media_type = _infer_media_type( + filename=filename, + default="application/octet-stream", + ) + contents.append( + Content.from_hosted_file( + file_id=file_id, + media_type=media_type, + name=filename or None, + additional_properties=file_properties, ) ) @@ -801,24 +863,18 @@ def _convert_openai_input_to_chat_message(self, input_items: list[Any], Message: except Exception as e: logger.error(f"Failed to process FunctionApprovalResponseContent: {e}") + if contents: + messages.append(Message(role=role_value, contents=contents)) + # Handle other OpenAI input item types as needed # (tool calls, function results, etc.) - # If no contents found, create a simple text message - if not contents: - contents.append(Content.from_text(text="")) + if not messages: + raise InputConversionError("OpenAI input did not contain any supported message content") - chat_message = Message(role="user", contents=contents) - - logger.info(f"Created Message with {len(contents)} contents:") - for idx, content in enumerate(contents): - content_type = content.__class__.__name__ - if hasattr(content, "media_type"): - logger.info(f" [{idx}] {content_type} - media_type: {content.media_type}") - else: - logger.info(f" [{idx}] {content_type}") + logger.info("Created %d Message object(s) from OpenAI input", len(messages)) - return chat_message + return messages[0] if len(messages) == 1 else messages def _extract_user_message_fallback(self, input_data: Any) -> str: """Fallback method to extract user message as string. @@ -857,8 +913,11 @@ def _is_openai_multimodal_format(self, input_data: Any) -> bool: first_item = input_data_items[0] if not isinstance(first_item, dict): return False - first_type = cast(dict[str, Any], first_item).get("type") - return isinstance(first_type, str) and first_type == "message" + first_item_dict = cast(dict[str, Any], first_item) + first_type = first_item_dict.get("type") + return first_type == "message" or ( + first_type is None and "role" in first_item_dict and "content" in first_item_dict + ) async def _parse_workflow_input(self, workflow: Any, raw_input: Any) -> Any: """Parse input based on workflow's expected input type. @@ -893,6 +952,8 @@ async def _parse_workflow_input(self, workflow: Any, raw_input: Any) -> Any: # Handle string input return self._parse_raw_workflow_input(workflow, str(raw_input)) + except InputConversionError: + raise except Exception as e: logger.warning(f"Error parsing workflow input: {e}") return cast(Any, raw_input) diff --git a/python/packages/devui/agent_framework_devui/_mapper.py b/python/packages/devui/agent_framework_devui/_mapper.py index 929f47e349f..b5ca3c34f72 100644 --- a/python/packages/devui/agent_framework_devui/_mapper.py +++ b/python/packages/devui/agent_framework_devui/_mapper.py @@ -280,7 +280,20 @@ async def aggregate_to_response(self, events: Sequence[Any], request: AgentFrame Returns: Final aggregated OpenAI response """ + context = self._get_or_create_context(request) try: + failed_response = next( + ( + event.response + for event in events + if getattr(event, "type", None) == "response.failed" + and isinstance(getattr(event, "response", None), Response) + ), + None, + ) + if failed_response is not None: + return cast(OpenAIResponse, failed_response) + # Collect output items in order output_items: list[Any] = [] @@ -413,22 +426,10 @@ async def aggregate_to_response(self, events: Sequence[Any], request: AgentFrame request_id = str(id(request)) usage_data = self._usage_accumulator.pop(request_id, None) - if usage_data is not None: - usage = _response_usage(usage_data) - else: - # Fallback: estimate if no usage was tracked - input_token_count = len(str(request.input)) // 4 if request.input else 0 - output_token_count = len(full_content) // 4 - usage = _response_usage( - UsageDetails( - input_token_count=input_token_count, - output_token_count=output_token_count, - total_token_count=input_token_count + output_token_count, - ) - ) + usage = _response_usage(usage_data) if usage_data is not None else None return OpenAIResponse( - id=f"resp_{uuid.uuid4().hex[:12]}", + id=context["response_id"], object="response", created_at=datetime.now().timestamp(), model=request.model or "devui", @@ -441,13 +442,14 @@ async def aggregate_to_response(self, events: Sequence[Any], request: AgentFrame except Exception as e: logger.exception(f"Error aggregating response: {e}") - return await self._create_error_response(str(e), request) + return await self._create_error_response(str(e), request, context["response_id"]) finally: # Cleanup: Remove context after aggregation to prevent memory leak # This handles the common case where streaming completes successfully request_key = id(request) if self._conversion_contexts.pop(request_key, None): logger.debug(f"Cleaned up context for request {request_key} after aggregation") + self._usage_accumulator.pop(str(request_key), None) def _get_or_create_context(self, request: AgentFrameworkRequest) -> dict[str, Any]: """Get or create conversion context for this request. @@ -468,8 +470,13 @@ def _get_or_create_context(self, request: AgentFrameworkRequest) -> dict[str, An evicted_key, _ = self._conversion_contexts.popitem(last=False) logger.debug(f"Evicted oldest context (key={evicted_key}) - at max capacity ({self._max_contexts})") + response_id = request.extra_body.get("response_id") if request.extra_body else None + if not isinstance(response_id, str) or not response_id: + response_id = f"resp_{uuid.uuid4().hex[:12]}" + self._conversion_contexts[request_key] = { "sequence_counter": 0, + "response_id": response_id, "item_id": f"msg_{uuid.uuid4().hex[:8]}", "content_index": 0, "output_index": 0, @@ -818,12 +825,9 @@ async def _convert_agent_lifecycle_event(self, event: Any, context: dict[str, An model = request_obj.model if request_obj and request_obj.model else "devui" if isinstance(event, AgentStartedEvent): - execution_id = f"agent_{uuid4().hex[:12]}" - context["execution_id"] = execution_id - # Create Response object response_obj = Response( - id=f"resp_{execution_id}", + id=context["response_id"], object="response", created_at=float(time.time()), model=model, @@ -850,15 +854,13 @@ async def _convert_agent_lifecycle_event(self, event: Any, context: dict[str, An return [] if isinstance(event, AgentFailedEvent): - execution_id = context.get("execution_id", f"agent_{uuid4().hex[:12]}") - # Create error object response_error = ResponseError( message=str(event.error) if event.error else "Unknown error", code="server_error" ) response_obj = Response( - id=f"resp_{execution_id}", + id=context["response_id"], object="response", created_at=float(time.time()), model=model, @@ -899,9 +901,6 @@ async def _convert_workflow_event(self, event: Any, context: dict[str, Any]) -> # Response-level events - construct proper OpenAI objects if event_type == "started": - workflow_id = getattr(event, "workflow_id", str(uuid4())) - context["workflow_id"] = workflow_id - # Import Response type for proper construction from openai.types.responses import Response @@ -914,7 +913,7 @@ async def _convert_workflow_event(self, event: Any, context: dict[str, Any]) -> # Create a full Response object with all required fields response_obj = Response( - id=f"resp_{workflow_id}", + id=context["response_id"], object="response", created_at=float(time.time()), model=model, @@ -1030,7 +1029,6 @@ async def _convert_workflow_event(self, event: Any, context: dict[str, Any]) -> return [] if event_type == "failed": - workflow_id = context.get("workflow_id", str(uuid4())) # failed event (type='failed') uses 'details' field (WorkflowErrorDetails), not 'error' # This matches executor_failed event which also uses 'details' details = getattr(event, "details", None) @@ -1059,7 +1057,7 @@ async def _convert_workflow_event(self, event: Any, context: dict[str, Any]) -> # Create a full Response object for failed state response_obj = Response( - id=f"resp_{workflow_id}", + id=context["response_id"], object="response", created_at=float(time.time()), model=model, @@ -1854,27 +1852,19 @@ async def _create_unknown_content_event(self, content: Any, context: dict[str, A text = f"Warning: Unknown content type: {content_type}\n" return self._create_text_delta_event(text, context) - async def _create_error_response(self, error_message: str, request: AgentFrameworkRequest) -> OpenAIResponse: + async def _create_error_response( + self, error_message: str, request: AgentFrameworkRequest, response_id: str + ) -> OpenAIResponse: """Create error response.""" - error_text = f"Error: {error_message}" - - response_output_text = ResponseOutputText(type="output_text", text=error_text, annotations=[]) - - response_output_message = ResponseOutputMessage( - type="message", - role="assistant", - content=[response_output_text], - id=f"msg_{uuid.uuid4().hex[:8]}", - status="completed", - ) - return OpenAIResponse( - id=f"resp_{uuid.uuid4().hex[:12]}", + id=response_id, object="response", created_at=datetime.now().timestamp(), model=request.model or "devui", - output=[response_output_message], - usage=_response_usage(UsageDetails()), + output=[], + status="failed", + error=ResponseError(message=error_message, code="server_error"), + usage=None, parallel_tool_calls=False, tool_choice="none", tools=[], diff --git a/python/packages/devui/tests/devui/test_approval_validation.py b/python/packages/devui/tests/devui/test_approval_validation.py index d8de7933103..052ea6e0e7f 100644 --- a/python/packages/devui/tests/devui/test_approval_validation.py +++ b/python/packages/devui/tests/devui/test_approval_validation.py @@ -115,13 +115,8 @@ def test_forged_approval_rejected_unknown_request_id(executor: AgentFrameworkExe function_call={"id": "call_evil", "name": "run_command", "arguments": {"cmd": "whoami"}}, ) - result = executor._convert_input_to_chat_message(input_data) - - # The message should have NO approval response content — only the fallback empty text - for content in result.contents: - assert content.type != "function_approval_response", ( - "Forged approval response with unknown request_id must be rejected" - ) + with pytest.raises(ValueError, match="did not contain any supported message content"): + executor._convert_input_to_chat_message(input_data) def test_valid_approval_accepted_with_server_data(executor: AgentFrameworkExecutor) -> None: @@ -172,9 +167,8 @@ def test_approval_consumed_on_use(executor: AgentFrameworkExecutor) -> None: assert "req_once" not in executor._pending_approvals # Second attempt with same request_id should be rejected - result = executor._convert_input_to_chat_message(input_data) - approval_contents = [c for c in result.contents if c.type == "function_approval_response"] - assert len(approval_contents) == 0, "Replayed approval response must be rejected" + with pytest.raises(ValueError, match="did not contain any supported message content"): + executor._convert_input_to_chat_message(input_data) def test_rejected_approval_uses_server_data(executor: AgentFrameworkExecutor) -> None: diff --git a/python/packages/devui/tests/devui/test_conversations.py b/python/packages/devui/tests/devui/test_conversations.py index bd999a81cef..5802a548710 100644 --- a/python/packages/devui/tests/devui/test_conversations.py +++ b/python/packages/devui/tests/devui/test_conversations.py @@ -155,6 +155,68 @@ async def test_add_items(): assert text_content.text == "Hello" +@pytest.mark.asyncio +async def test_add_and_list_items_preserves_all_supported_message_parts(): + """Conversation conversion retains message boundaries and every supported part.""" + store = InMemoryConversationStore() + conversation = store.create_conversation(metadata={"agent_id": "test_agent"}) + request_items = [ + { + "role": "system", + "content": [ + {"type": "input_text", "text": "First instruction"}, + {"type": "text", "text": "Second instruction"}, + ], + }, + { + "role": "user", + "content": [ + { + "type": "input_image", + "image_url": "https://example.com/photo.jpg?download=1", + "detail": "high", + }, + {"type": "input_image", "file_id": "file_image", "detail": "low"}, + {"type": "input_file", "file_data": "JVBERi0=", "filename": "report.pdf"}, + {"type": "input_file", "file_id": "file_audio", "filename": "recording.mp3"}, + {"type": "input_file", "file_id": "file_scan", "filename": "scan.jpg"}, + ], + }, + ] + + created_items = await store.add_items(conversation.id, items=request_items) + listed_items, has_more = await store.list_items(conversation.id) + + assert [item.role for item in created_items] == ["system", "user"] + assert [len(item.content) for item in created_items] == [2, 5] + assert created_items[1].content[0].image_url == "https://example.com/photo.jpg?download=1" + assert created_items[1].content[0].detail == "high" + assert created_items[1].content[1].file_id == "file_image" + assert created_items[1].content[1].detail == "low" + assert created_items[1].content[2].file_url == "data:application/pdf;base64,JVBERi0=" + assert created_items[1].content[2].filename == "report.pdf" + assert created_items[1].content[3].file_id == "file_audio" + assert created_items[1].content[3].filename == "recording.mp3" + assert created_items[1].content[4].type == "input_file" + assert created_items[1].content[4].file_id == "file_scan" + + stored_messages = store._conversations[conversation.id]["messages"] + assert stored_messages[1].contents[0].media_type == "image/jpeg" + assert stored_messages[1].contents[2].media_type == "application/pdf" + assert stored_messages[1].contents[3].media_type == "audio/mpeg" + assert stored_messages[1].contents[4].media_type == "image/jpeg" + + assert has_more is False + assert [item.role for item in listed_items] == ["system", "user"] + assert [len(item.content) for item in listed_items] == [2, 5] + assert listed_items[1].content[0].detail == "high" + assert listed_items[1].content[1].file_id == "file_image" + assert listed_items[1].content[2].file_url == "data:application/pdf;base64,JVBERi0=" + assert listed_items[1].content[3].file_id == "file_audio" + assert listed_items[1].content[4].type == "input_file" + assert listed_items[1].content[4].file_id == "file_scan" + + @pytest.mark.asyncio async def test_list_items(): """Test listing conversation items.""" diff --git a/python/packages/devui/tests/devui/test_mapper.py b/python/packages/devui/tests/devui/test_mapper.py index b37647b05d6..dacea8f8729 100644 --- a/python/packages/devui/tests/devui/test_mapper.py +++ b/python/packages/devui/tests/devui/test_mapper.py @@ -295,6 +295,13 @@ async def test_zero_usage_content_does_not_use_estimate( assert response.usage.total_tokens == 0 +async def test_missing_usage_is_not_estimated(mapper: MessageMapper, test_request: AgentFrameworkRequest) -> None: + """Completed responses omit usage when the framework reported none.""" + response = await mapper.aggregate_to_response([], test_request) + + assert response.usage is None + + # ============================================================================= # Agent Lifecycle Event Tests # ============================================================================= @@ -329,6 +336,25 @@ async def test_agent_lifecycle_events(mapper: MessageMapper, test_request: Agent assert events[0].response.error.message == "Test error" +async def test_response_id_is_stable_across_lifecycle_and_aggregation(mapper: MessageMapper) -> None: + """The server tracking ID remains the OpenAI response ID for the full lifecycle.""" + request = AgentFrameworkRequest( + model="devui", + input="hello", + extra_body={"response_id": "resp_tracking_123"}, + ) + + started_events = await mapper.convert_event(AgentStartedEvent(), request) + failed_events = await mapper.convert_event(AgentFailedEvent(error=RuntimeError("failed")), request) + response = await mapper.aggregate_to_response([*started_events, *failed_events], request) + + assert [event.response.id for event in started_events] == ["resp_tracking_123", "resp_tracking_123"] + assert failed_events[0].response.id == "resp_tracking_123" + assert response.id == "resp_tracking_123" + assert response.status == "failed" + assert response.output == [] + + async def test_agent_run_response_mapping(mapper: MessageMapper, test_request: AgentFrameworkRequest) -> None: """Test that mapper handles complete AgentResponse (non-streaming).""" response = create_agent_run_response("Complete response from run()") diff --git a/python/packages/devui/tests/devui/test_multimodal_workflow.py b/python/packages/devui/tests/devui/test_multimodal_workflow.py index 7af7f3f3082..c582bc37d94 100644 --- a/python/packages/devui/tests/devui/test_multimodal_workflow.py +++ b/python/packages/devui/tests/devui/test_multimodal_workflow.py @@ -9,6 +9,8 @@ import json from unittest.mock import MagicMock +import pytest + from agent_framework_devui._discovery import EntityDiscovery from agent_framework_devui._executor import AgentFrameworkExecutor from agent_framework_devui._mapper import MessageMapper @@ -30,11 +32,10 @@ def test_is_openai_multimodal_format_detects_message_format(self): # Valid OpenAI multimodal format valid_format = [ { - "type": "message", "role": "user", "content": [ {"type": "input_text", "text": "Describe this image"}, - {"type": "input_image", "image_url": TEST_IMAGE_DATA_URI}, + {"type": "input_image", "image_url": TEST_IMAGE_DATA_URI, "detail": "high"}, ], } ] @@ -62,7 +63,7 @@ def test_convert_openai_input_to_chat_message_with_image(self): "role": "user", "content": [ {"type": "input_text", "text": "Describe this image"}, - {"type": "input_image", "image_url": TEST_IMAGE_DATA_URI}, + {"type": "input_image", "image_url": TEST_IMAGE_DATA_URI, "detail": "high"}, ], } ] @@ -85,6 +86,100 @@ def test_convert_openai_input_to_chat_message_with_image(self): assert result.contents[1].type == "data" assert result.contents[1].media_type == "image/png" assert result.contents[1].uri == TEST_IMAGE_DATA_URI + assert result.contents[1].additional_properties["detail"] == "high" + + def test_convert_openai_input_preserves_message_roles_and_boundaries(self): + """Official input message roles remain distinct Agent Framework messages.""" + from agent_framework import Message + + executor = AgentFrameworkExecutor(MagicMock(spec=EntityDiscovery), MagicMock(spec=MessageMapper)) + openai_input = [ + {"role": "system", "content": "System guidance"}, + { + "type": "message", + "role": "developer", + "content": [{"type": "input_text", "text": "Developer guidance"}], + }, + { + "type": "message", + "role": "user", + "content": [ + {"type": "input_text", "text": "First part"}, + {"type": "input_text", "text": "Second part"}, + ], + }, + ] + + result = executor._convert_input_to_chat_message(openai_input) + + assert isinstance(result, list) + assert all(isinstance(message, Message) for message in result) + assert [message.role for message in result] == ["system", "developer", "user"] + assert [[content.text for content in message.contents] for message in result] == [ + ["System guidance"], + ["Developer guidance"], + ["First part", "Second part"], + ] + + def test_convert_openai_input_preserves_file_ids_and_canonical_media_types(self): + """Hosted files and straightforward MIME types survive input conversion.""" + executor = AgentFrameworkExecutor(MagicMock(spec=EntityDiscovery), MagicMock(spec=MessageMapper)) + openai_input = [ + { + "type": "message", + "role": "user", + "content": [ + { + "type": "input_image", + "image_url": "https://example.com/photo.jpg?download=1", + "detail": "low", + }, + {"type": "input_image", "file_id": "file_image", "detail": "high"}, + {"type": "input_file", "file_id": "file_audio", "filename": "recording.mp3"}, + {"type": "input_file", "file_url": "https://example.com/recording.m4a"}, + ], + } + ] + + result = executor._convert_input_to_chat_message(openai_input) + + assert result.contents[0].media_type == "image/jpeg" + assert result.contents[0].additional_properties["detail"] == "low" + assert result.contents[1].type == "hosted_file" + assert result.contents[1].file_id == "file_image" + assert result.contents[1].additional_properties["openai_content_type"] == "input_image" + assert result.contents[1].additional_properties["detail"] == "high" + assert result.contents[2].type == "hosted_file" + assert result.contents[2].file_id == "file_audio" + assert result.contents[2].media_type == "audio/mpeg" + assert result.contents[3].media_type == "audio/mp4" + + def test_convert_openai_input_rejects_unsupported_only_content(self): + """Unsupported input must not become a fabricated empty user message.""" + executor = AgentFrameworkExecutor(MagicMock(spec=EntityDiscovery), MagicMock(spec=MessageMapper)) + openai_input = [ + { + "type": "message", + "role": "user", + "content": [{"type": "unsupported_content", "value": "ignored"}], + } + ] + + with pytest.raises(ValueError, match="did not contain any supported message content"): + executor._convert_input_to_chat_message(openai_input) + + async def test_parse_workflow_input_rejects_unsupported_only_content(self): + """Workflow parsing must not fall back to the raw unsupported payload.""" + executor = AgentFrameworkExecutor(MagicMock(spec=EntityDiscovery), MagicMock(spec=MessageMapper)) + openai_input = [ + { + "role": "user", + "content": [{"type": "unsupported_content", "value": "ignored"}], + } + ] + + with pytest.raises(ValueError, match="did not contain any supported message content"): + await executor._parse_workflow_input(MagicMock(), openai_input) async def test_parse_workflow_input_handles_json_string_with_multimodal(self): """Test that _parse_workflow_input correctly handles JSON string with multimodal content.""" From 5dfaef081c2a971cb0d891b552544a21533287bf Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Mon, 31 Aug 2026 15:14:41 +0200 Subject: [PATCH 2/4] Python: address DevUI parsing review feedback Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../devui/agent_framework_devui/_executor.py | 8 ++- .../devui/agent_framework_devui/_mapper.py | 7 +- .../devui/agent_framework_devui/_server.py | 44 ++++++++---- .../devui/tests/devui/test_conversations.py | 72 +++++++++++++------ .../packages/devui/tests/devui/test_mapper.py | 21 ++++++ .../tests/devui/test_multimodal_workflow.py | 26 ++++++- .../packages/devui/tests/devui/test_server.py | 49 ++++++++++++- 7 files changed, 184 insertions(+), 43 deletions(-) diff --git a/python/packages/devui/agent_framework_devui/_executor.py b/python/packages/devui/agent_framework_devui/_executor.py index 2f9117fb83f..cec18108a0b 100644 --- a/python/packages/devui/agent_framework_devui/_executor.py +++ b/python/packages/devui/agent_framework_devui/_executor.py @@ -707,7 +707,7 @@ def _convert_openai_input_to_chat_message(self, input_items: list[Any], Message: content_dict = cast(dict[str, Any], content_item) content_type = content_dict.get("type") - if content_type == "input_text": + if content_type in ("input_text", "output_text", "text"): text = content_dict.get("text", "") if isinstance(text, str): contents.append(Content.from_text(text=text)) @@ -863,8 +863,10 @@ def _convert_openai_input_to_chat_message(self, input_items: list[Any], Message: except Exception as e: logger.error(f"Failed to process FunctionApprovalResponseContent: {e}") - if contents: - messages.append(Message(role=role_value, contents=contents)) + if not contents: + raise InputConversionError("OpenAI input message did not contain any supported message content") + + messages.append(Message(role=role_value, contents=contents)) # Handle other OpenAI input item types as needed # (tool calls, function results, etc.) diff --git a/python/packages/devui/agent_framework_devui/_mapper.py b/python/packages/devui/agent_framework_devui/_mapper.py index b5ca3c34f72..acaab2e92fb 100644 --- a/python/packages/devui/agent_framework_devui/_mapper.py +++ b/python/packages/devui/agent_framework_devui/_mapper.py @@ -291,9 +291,6 @@ async def aggregate_to_response(self, events: Sequence[Any], request: AgentFrame ), None, ) - if failed_response is not None: - return cast(OpenAIResponse, failed_response) - # Collect output items in order output_items: list[Any] = [] @@ -411,7 +408,7 @@ async def aggregate_to_response(self, events: Sequence[Any], request: AgentFrame output_items.append(response_output_message) # If no output items at all, create an empty message - if not output_items: + if not output_items and failed_response is None: response_output_text = ResponseOutputText(type="output_text", text="", annotations=[]) response_output_message = ResponseOutputMessage( type="message", @@ -434,6 +431,8 @@ async def aggregate_to_response(self, events: Sequence[Any], request: AgentFrame created_at=datetime.now().timestamp(), model=request.model or "devui", output=output_items, + status=failed_response.status if failed_response is not None else "completed", + error=failed_response.error if failed_response is not None else None, usage=usage, parallel_tool_calls=False, tool_choice="none", diff --git a/python/packages/devui/agent_framework_devui/_server.py b/python/packages/devui/agent_framework_devui/_server.py index 5a6e3e917ad..6a83cfecac8 100644 --- a/python/packages/devui/agent_framework_devui/_server.py +++ b/python/packages/devui/agent_framework_devui/_server.py @@ -1227,7 +1227,7 @@ async def _stream_execution( ) -> AsyncGenerator[str]: """Stream execution directly through executor.""" try: - # Collect events for final response.completed event + # Collect events for the final terminal event events: list[Any] = [] # Get conversation_id for trace storage @@ -1237,9 +1237,11 @@ async def _stream_execution( # Stream all events async for event in executor.execute_streaming(request): events.append(event) + event_type_value = getattr(event, "type", None) + event_type = event_type_value if isinstance(event_type_value, str) else None # Store trace events for context inspection (persisted with conversation) - if conversation_id and hasattr(event, "type") and event.type == "response.trace.completed": + if conversation_id and event_type == "response.trace.completed": try: trace_data = event.data if hasattr(event, "data") else None if trace_data and isinstance(conversation_id, str): @@ -1247,6 +1249,11 @@ async def _stream_execution( except Exception as e: logger.debug(f"Failed to store trace event: {e}") + # Failure responses need aggregation so their terminal payload includes + # any output emitted before the failure. + if event_type == "response.failed": + continue + # IMPORTANT: Check model_dump_json FIRST because to_json() can have newlines (pretty-printing) # which breaks SSE format. model_dump_json() returns single-line JSON. if hasattr(event, "model_dump_json"): @@ -1264,27 +1271,36 @@ async def _stream_execution( payload = json.dumps(str(event)) yield f"data: {payload}\n\n" - # Aggregate to final response and emit response.completed event (OpenAI standard) - from .models import ResponseCompletedEvent - + # Aggregate to the final response before emitting its matching terminal event. final_response = await executor.message_mapper.aggregate_to_response(events, request) - # The sequence number for response.completed should be the next number after all events - # The last event in the list should have the highest sequence number so far - # We need to increment from that + # Use the next sequence number after the last event actually sent to the client. last_seq = 0 for event in reversed(events): + if getattr(event, "type", None) == "response.failed": + continue sequence_number = getattr(event, "sequence_number", None) if isinstance(sequence_number, int): last_seq = sequence_number break - completed_event = ResponseCompletedEvent( - type="response.completed", - response=final_response, - sequence_number=last_seq + 1, - ) - yield f"data: {completed_event.model_dump_json()}\n\n" + if final_response.status == "failed": + from openai.types.responses import ResponseFailedEvent + + terminal_event = ResponseFailedEvent( + type="response.failed", + response=final_response, + sequence_number=last_seq + 1, + ) + else: + from .models import ResponseCompletedEvent + + terminal_event = ResponseCompletedEvent( + type="response.completed", + response=final_response, + sequence_number=last_seq + 1, + ) + yield f"data: {terminal_event.model_dump_json()}\n\n" # Send final done event yield "data: [DONE]\n\n" diff --git a/python/packages/devui/tests/devui/test_conversations.py b/python/packages/devui/tests/devui/test_conversations.py index 5802a548710..803fc8249e2 100644 --- a/python/packages/devui/tests/devui/test_conversations.py +++ b/python/packages/devui/tests/devui/test_conversations.py @@ -6,6 +6,8 @@ import pytest from openai.types.conversations import InputTextContent +from openai.types.conversations.message import Message as OpenAIMessage +from openai.types.responses import ResponseInputFile, ResponseInputImage from agent_framework_devui._conversations import InMemoryConversationStore @@ -187,18 +189,33 @@ async def test_add_and_list_items_preserves_all_supported_message_parts(): created_items = await store.add_items(conversation.id, items=request_items) listed_items, has_more = await store.list_items(conversation.id) - assert [item.role for item in created_items] == ["system", "user"] - assert [len(item.content) for item in created_items] == [2, 5] - assert created_items[1].content[0].image_url == "https://example.com/photo.jpg?download=1" - assert created_items[1].content[0].detail == "high" - assert created_items[1].content[1].file_id == "file_image" - assert created_items[1].content[1].detail == "low" - assert created_items[1].content[2].file_url == "data:application/pdf;base64,JVBERi0=" - assert created_items[1].content[2].filename == "report.pdf" - assert created_items[1].content[3].file_id == "file_audio" - assert created_items[1].content[3].filename == "recording.mp3" - assert created_items[1].content[4].type == "input_file" - assert created_items[1].content[4].file_id == "file_scan" + created_system, created_user = created_items + assert isinstance(created_system, OpenAIMessage) + assert isinstance(created_user, OpenAIMessage) + assert [created_system.role, created_user.role] == ["system", "user"] + assert created_system.content is not None + assert created_user.content is not None + assert [len(created_system.content), len(created_user.content)] == [2, 5] + + created_url_image = created_user.content[0] + created_hosted_image = created_user.content[1] + created_data_file = created_user.content[2] + created_audio_file = created_user.content[3] + created_scan_file = created_user.content[4] + assert isinstance(created_url_image, ResponseInputImage) + assert isinstance(created_hosted_image, ResponseInputImage) + assert isinstance(created_data_file, ResponseInputFile) + assert isinstance(created_audio_file, ResponseInputFile) + assert isinstance(created_scan_file, ResponseInputFile) + assert created_url_image.image_url == "https://example.com/photo.jpg?download=1" + assert created_url_image.detail == "high" + assert created_hosted_image.file_id == "file_image" + assert created_hosted_image.detail == "low" + assert created_data_file.file_url == "data:application/pdf;base64,JVBERi0=" + assert created_data_file.filename == "report.pdf" + assert created_audio_file.file_id == "file_audio" + assert created_audio_file.filename == "recording.mp3" + assert created_scan_file.file_id == "file_scan" stored_messages = store._conversations[conversation.id]["messages"] assert stored_messages[1].contents[0].media_type == "image/jpeg" @@ -207,14 +224,29 @@ async def test_add_and_list_items_preserves_all_supported_message_parts(): assert stored_messages[1].contents[4].media_type == "image/jpeg" assert has_more is False - assert [item.role for item in listed_items] == ["system", "user"] - assert [len(item.content) for item in listed_items] == [2, 5] - assert listed_items[1].content[0].detail == "high" - assert listed_items[1].content[1].file_id == "file_image" - assert listed_items[1].content[2].file_url == "data:application/pdf;base64,JVBERi0=" - assert listed_items[1].content[3].file_id == "file_audio" - assert listed_items[1].content[4].type == "input_file" - assert listed_items[1].content[4].file_id == "file_scan" + listed_system, listed_user = listed_items + assert isinstance(listed_system, OpenAIMessage) + assert isinstance(listed_user, OpenAIMessage) + assert [listed_system.role, listed_user.role] == ["system", "user"] + assert listed_system.content is not None + assert listed_user.content is not None + assert [len(listed_system.content), len(listed_user.content)] == [2, 5] + + listed_url_image = listed_user.content[0] + listed_hosted_image = listed_user.content[1] + listed_data_file = listed_user.content[2] + listed_audio_file = listed_user.content[3] + listed_scan_file = listed_user.content[4] + assert isinstance(listed_url_image, ResponseInputImage) + assert isinstance(listed_hosted_image, ResponseInputImage) + assert isinstance(listed_data_file, ResponseInputFile) + assert isinstance(listed_audio_file, ResponseInputFile) + assert isinstance(listed_scan_file, ResponseInputFile) + assert listed_url_image.detail == "high" + assert listed_hosted_image.file_id == "file_image" + assert listed_data_file.file_url == "data:application/pdf;base64,JVBERi0=" + assert listed_audio_file.file_id == "file_audio" + assert listed_scan_file.file_id == "file_scan" @pytest.mark.asyncio diff --git a/python/packages/devui/tests/devui/test_mapper.py b/python/packages/devui/tests/devui/test_mapper.py index dacea8f8729..29b5240927f 100644 --- a/python/packages/devui/tests/devui/test_mapper.py +++ b/python/packages/devui/tests/devui/test_mapper.py @@ -355,6 +355,27 @@ async def test_response_id_is_stable_across_lifecycle_and_aggregation(mapper: Me assert response.output == [] +async def test_failed_response_retains_partial_output( + mapper: MessageMapper, test_request: AgentFrameworkRequest +) -> None: + """A failed response includes text emitted before the failure.""" + partial_events = await mapper.convert_event( + create_test_agent_update([Content.from_text(text="Partial output")]), + test_request, + ) + failed_events = await mapper.convert_event( + AgentFailedEvent(error=RuntimeError("failed after output")), + test_request, + ) + + response = await mapper.aggregate_to_response([*partial_events, *failed_events], test_request) + + assert response.status == "failed" + assert response.error is not None + assert response.error.message == "failed after output" + assert response.output_text == "Partial output" + + async def test_agent_run_response_mapping(mapper: MessageMapper, test_request: AgentFrameworkRequest) -> None: """Test that mapper handles complete AgentResponse (non-streaming).""" response = create_agent_run_response("Complete response from run()") diff --git a/python/packages/devui/tests/devui/test_multimodal_workflow.py b/python/packages/devui/tests/devui/test_multimodal_workflow.py index c582bc37d94..4699c813c5f 100644 --- a/python/packages/devui/tests/devui/test_multimodal_workflow.py +++ b/python/packages/devui/tests/devui/test_multimodal_workflow.py @@ -108,17 +108,26 @@ def test_convert_openai_input_preserves_message_roles_and_boundaries(self): {"type": "input_text", "text": "Second part"}, ], }, + { + "type": "message", + "role": "assistant", + "content": [ + {"type": "output_text", "text": "Assistant output"}, + {"type": "text", "text": "Generic assistant text"}, + ], + }, ] result = executor._convert_input_to_chat_message(openai_input) assert isinstance(result, list) assert all(isinstance(message, Message) for message in result) - assert [message.role for message in result] == ["system", "developer", "user"] + assert [message.role for message in result] == ["system", "developer", "user", "assistant"] assert [[content.text for content in message.contents] for message in result] == [ ["System guidance"], ["Developer guidance"], ["First part", "Second part"], + ["Assistant output", "Generic assistant text"], ] def test_convert_openai_input_preserves_file_ids_and_canonical_media_types(self): @@ -168,6 +177,21 @@ def test_convert_openai_input_rejects_unsupported_only_content(self): with pytest.raises(ValueError, match="did not contain any supported message content"): executor._convert_input_to_chat_message(openai_input) + def test_convert_openai_input_rejects_each_unsupported_only_message(self): + """A valid message must not hide another message with no supported content.""" + executor = AgentFrameworkExecutor(MagicMock(spec=EntityDiscovery), MagicMock(spec=MessageMapper)) + openai_input = [ + {"type": "message", "role": "user", "content": "Valid message"}, + { + "type": "message", + "role": "assistant", + "content": [{"type": "unsupported_content", "value": "ignored"}], + }, + ] + + with pytest.raises(ValueError, match="did not contain any supported message content"): + executor._convert_input_to_chat_message(openai_input) + async def test_parse_workflow_input_rejects_unsupported_only_content(self): """Workflow parsing must not fall back to the raw unsupported payload.""" executor = AgentFrameworkExecutor(MagicMock(spec=EntityDiscovery), MagicMock(spec=MessageMapper)) diff --git a/python/packages/devui/tests/devui/test_server.py b/python/packages/devui/tests/devui/test_server.py index 9ebd4ee40d1..310a7580d0e 100644 --- a/python/packages/devui/tests/devui/test_server.py +++ b/python/packages/devui/tests/devui/test_server.py @@ -4,24 +4,30 @@ import asyncio import inspect +import json import logging import sys import tempfile +from collections.abc import AsyncGenerator from pathlib import Path from typing import Any import pytest +from agent_framework import AgentResponseUpdate, Content from conftest import MockAgent # pyrefly: ignore[missing-import] # pyright: ignore[reportMissingImports] from fastapi.testclient import TestClient import agent_framework_devui from agent_framework_devui import DevServer +from agent_framework_devui._discovery import EntityDiscovery +from agent_framework_devui._executor import AgentFrameworkExecutor +from agent_framework_devui._mapper import MessageMapper from agent_framework_devui._utils import ( extract_executor_message_types, parse_input_for_type, select_primary_input_type, ) -from agent_framework_devui.models._openai_custom import AgentFrameworkRequest +from agent_framework_devui.models._openai_custom import AgentFailedEvent, AgentFrameworkRequest class _StubExecutor: @@ -34,6 +40,28 @@ def __init__(self, *, input_types=None, handlers=None): self._handlers = dict(handlers) +class _FailedStreamingExecutor(AgentFrameworkExecutor): + """Executor stub that emits text before a terminal failure.""" + + def __init__(self) -> None: + super().__init__(EntityDiscovery(None), MessageMapper()) + + async def execute_streaming(self, request: AgentFrameworkRequest) -> AsyncGenerator[Any]: + partial_update = AgentResponseUpdate( + contents=[Content.from_text(text="Partial output")], + role="assistant", + message_id="partial_message", + response_id="partial_response", + ) + for event in await self.message_mapper.convert_event(partial_update, request): + yield event + for event in await self.message_mapper.convert_event( + AgentFailedEvent(error=RuntimeError("failed after output")), + request, + ): + yield event + + # Note: test_entities_dir fixture is provided by conftest.py @@ -108,6 +136,25 @@ async def test_server_execution_streaming(test_entities_dir): assert event_count > 0 +async def test_stream_execution_emits_one_failed_terminal_event_with_partial_output(): + """Failed streams retain prior output and never emit a completion event.""" + server = DevServer(auth_enabled=False) + executor = _FailedStreamingExecutor() + request = AgentFrameworkRequest(input="hello", stream=True) + + chunks = [chunk async for chunk in server._stream_execution(executor, request)] + payloads = [json.loads(chunk.removeprefix("data: ").strip()) for chunk in chunks if chunk != "data: [DONE]\n\n"] + terminal_events = [ + payload for payload in payloads if payload.get("type") in {"response.failed", "response.completed"} + ] + + assert [event["type"] for event in terminal_events] == ["response.failed"] + failed_response = terminal_events[0]["response"] + assert failed_response["status"] == "failed" + assert failed_response["error"]["message"] == "failed after output" + assert failed_response["output"][0]["content"][0]["text"] == "Partial output" + + def test_configuration(): """Test basic configuration.""" server = DevServer(entities_dir="test", port=9000, host="localhost", auth_enabled=False) From a8a8487e1bb8ef268370aee6376dfc3eee179dd5 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Mon, 31 Aug 2026 15:27:00 +0200 Subject: [PATCH 3/4] Python: preserve approvals on invalid DevUI batches Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../devui/agent_framework_devui/_executor.py | 11 ++++++- .../tests/devui/test_approval_validation.py | 29 +++++++++++++++++++ 2 files changed, 39 insertions(+), 1 deletion(-) diff --git a/python/packages/devui/agent_framework_devui/_executor.py b/python/packages/devui/agent_framework_devui/_executor.py index cec18108a0b..b231e293ed2 100644 --- a/python/packages/devui/agent_framework_devui/_executor.py +++ b/python/packages/devui/agent_framework_devui/_executor.py @@ -679,6 +679,7 @@ def _convert_openai_input_to_chat_message(self, input_items: list[Any], Message: One Message, or a list of Messages when input contains multiple message items """ messages: list[Any] = [] + approval_request_ids: set[str] = set() # Process each input item for item in input_items: @@ -814,7 +815,11 @@ def _convert_openai_input_to_chat_message(self, input_items: list[Any], Message: # Only accept responses that match a request we issued. # Always use the server-stored function_call data. - stored_fc = self._pending_approvals.pop(request_id, None) + stored_fc = ( + None + if request_id in approval_request_ids + else self._pending_approvals.get(request_id) + ) if stored_fc is None: logger.warning( "Rejected function_approval_response with unknown " @@ -848,6 +853,7 @@ def _convert_openai_input_to_chat_message(self, input_items: list[Any], Message: additional_properties=approval_additional_props, ) contents.append(approval_response) + approval_request_ids.add(request_id) logger.info( "Validated FunctionApprovalResponseContent: id=%s, " "approved=%s, function=%s, policy_violation=%s", @@ -874,6 +880,9 @@ def _convert_openai_input_to_chat_message(self, input_items: list[Any], Message: if not messages: raise InputConversionError("OpenAI input did not contain any supported message content") + for request_id in approval_request_ids: + self._pending_approvals.pop(request_id, None) + logger.info("Created %d Message object(s) from OpenAI input", len(messages)) return messages[0] if len(messages) == 1 else messages diff --git a/python/packages/devui/tests/devui/test_approval_validation.py b/python/packages/devui/tests/devui/test_approval_validation.py index 052ea6e0e7f..0bc73724b22 100644 --- a/python/packages/devui/tests/devui/test_approval_validation.py +++ b/python/packages/devui/tests/devui/test_approval_validation.py @@ -171,6 +171,35 @@ def test_approval_consumed_on_use(executor: AgentFrameworkExecutor) -> None: executor._convert_input_to_chat_message(input_data) +def test_approval_not_consumed_when_later_message_is_invalid(executor: AgentFrameworkExecutor) -> None: + """Batch validation failure leaves an earlier valid approval available to retry.""" + executor._pending_approvals["req_retry"] = { + "call_id": "call_retry", + "name": "retry_tool", + "arguments": {}, + } + invalid_batch = [ + *_make_approval_response_input(request_id="req_retry", approved=True), + { + "type": "message", + "role": "user", + "content": [{"type": "unsupported_content", "value": "ignored"}], + }, + ] + + with pytest.raises(ValueError, match="did not contain any supported message content"): + executor._convert_input_to_chat_message(invalid_batch) + + assert "req_retry" in executor._pending_approvals + + result = executor._convert_input_to_chat_message( + _make_approval_response_input(request_id="req_retry", approved=True) + ) + + assert result.contents[0].type == "function_approval_response" + assert "req_retry" not in executor._pending_approvals + + def test_rejected_approval_uses_server_data(executor: AgentFrameworkExecutor) -> None: """Even rejected (approved=False) responses use server-stored function_call data.""" executor._pending_approvals["req_deny"] = { From 2aa4cdc93b1a82c0cb083ed1f2c8ee8b36600718 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Tue, 1 Sep 2026 11:21:36 +0200 Subject: [PATCH 4/4] Python: address DevUI parser follow-up review Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 832e81b1-76ee-47ca-981f-be365d987336 --- .../agent_framework_devui/_conversations.py | 40 +++--------- .../devui/agent_framework_devui/_executor.py | 52 ++++++--------- .../devui/agent_framework_devui/_mapper.py | 63 +++++++++++++++---- .../devui/agent_framework_devui/_server.py | 40 ------------ .../devui/agent_framework_devui/_utils.py | 29 +++++++++ .../devui/tests/devui/test_conversations.py | 31 +++++++++ .../packages/devui/tests/devui/test_mapper.py | 43 +++++++++++++ .../packages/devui/tests/devui/test_server.py | 24 +++---- .../agent_framework_openai/_chat_client.py | 23 ++++++- .../tests/openai/test_openai_chat_client.py | 49 +++++++++++++++ 10 files changed, 262 insertions(+), 132 deletions(-) diff --git a/python/packages/devui/agent_framework_devui/_conversations.py b/python/packages/devui/agent_framework_devui/_conversations.py index 22a42e05139..2f3fe8feb46 100644 --- a/python/packages/devui/agent_framework_devui/_conversations.py +++ b/python/packages/devui/agent_framework_devui/_conversations.py @@ -8,13 +8,11 @@ from __future__ import annotations -import mimetypes import time import uuid from abc import ABC, abstractmethod from collections.abc import MutableSequence from typing import Any, Literal, cast -from urllib.parse import urlparse from agent_framework import AgentSession, Content, Message from agent_framework._workflows._checkpoint import InMemoryCheckpointStorage, WorkflowCheckpoint @@ -30,38 +28,14 @@ ResponseInputImage, ) +from ._utils import infer_media_type + # Type alias for OpenAI Message role literals MessageRole = Literal["unknown", "user", "assistant", "system", "critic", "discriminator", "developer", "tool"] # Checkpoint item type constants CONVERSATION_ITEM_TYPE_CHECKPOINT = "checkpoint" CONVERSATION_TYPE_CHECKPOINT_CONTAINER = "checkpoint_container" -_CANONICAL_MEDIA_TYPES = { - ".jpeg": "image/jpeg", - ".jpg": "image/jpeg", - ".m4a": "audio/mp4", - ".mp3": "audio/mpeg", -} - - -def _infer_media_type(*, filename: str | None = None, uri: str | None = None, default: str) -> str: - """Infer a canonical media type from explicit data or a filename.""" - if uri and uri.startswith("data:"): - media_type = uri[5:].split(";", 1)[0].split(",", 1)[0] - if "/" in media_type: - return media_type - - for candidate in (filename, urlparse(uri).path if uri else None): - if not candidate: - continue - candidate_lower = candidate.lower() - for suffix, canonical_media_type in _CANONICAL_MEDIA_TYPES.items(): - if candidate_lower.endswith(suffix): - return canonical_media_type - if guessed_media_type := mimetypes.guess_type(candidate)[0]: - return guessed_media_type - - return default class ConversationStore(ABC): @@ -551,7 +525,7 @@ async def list_items( def _to_agent_content(content: dict[str, Any]) -> Content | None: """Convert one supported OpenAI conversation message part.""" content_type = content.get("type") - if content_type in ("text", "input_text"): + if content_type in ("text", "input_text", "output_text"): text = content.get("text") return Content.from_text(text=text) if isinstance(text, str) else None @@ -565,7 +539,7 @@ def _to_agent_content(content: dict[str, Any]) -> Content | None: if isinstance(image_url, str) and image_url: return Content.from_uri( uri=image_url, - media_type=_infer_media_type(uri=image_url, default="image/png"), + media_type=infer_media_type(uri=image_url, default="image/png"), additional_properties=image_properties, ) file_id = content.get("file_id") @@ -588,7 +562,7 @@ def _to_agent_content(content: dict[str, Any]) -> Content | None: file_data = content.get("file_data") if isinstance(file_data, str) and file_data: - media_type = _infer_media_type( + media_type = infer_media_type( filename=filename, uri=file_data, default="application/octet-stream", @@ -604,7 +578,7 @@ def _to_agent_content(content: dict[str, Any]) -> Content | None: if isinstance(file_url, str) and file_url: return Content.from_uri( uri=file_url, - media_type=_infer_media_type( + media_type=infer_media_type( filename=filename, uri=file_url, default="application/octet-stream", @@ -616,7 +590,7 @@ def _to_agent_content(content: dict[str, Any]) -> Content | None: if isinstance(file_id, str) and file_id: return Content.from_hosted_file( file_id=file_id, - media_type=_infer_media_type(filename=filename, default="application/octet-stream"), + media_type=infer_media_type(filename=filename, default="application/octet-stream"), name=filename or None, additional_properties=file_properties, ) diff --git a/python/packages/devui/agent_framework_devui/_executor.py b/python/packages/devui/agent_framework_devui/_executor.py index b231e293ed2..35fcf8e6803 100644 --- a/python/packages/devui/agent_framework_devui/_executor.py +++ b/python/packages/devui/agent_framework_devui/_executor.py @@ -6,10 +6,8 @@ import json import logging -import mimetypes from collections.abc import AsyncGenerator from typing import Any, cast -from urllib.parse import urlparse from agent_framework import Content, SupportsAgentRun, Workflow @@ -17,18 +15,13 @@ from ._discovery import EntityDiscovery from ._mapper import MessageMapper from ._tracing import capture_traces +from ._utils import infer_media_type from .models import AgentFrameworkRequest, OpenAIResponse from .models._discovery_models import EntityInfo logger = logging.getLogger(__name__) _OPENAI_MESSAGE_ROLES = {"user", "assistant", "system", "developer"} -_CANONICAL_MEDIA_TYPES = { - ".jpeg": "image/jpeg", - ".jpg": "image/jpeg", - ".m4a": "audio/mp4", - ".mp3": "audio/mpeg", -} def _get_event_type(event: Any) -> str | None: @@ -39,26 +32,6 @@ def _get_event_type(event: Any) -> str | None: return getattr(event, "type", None) -def _infer_media_type(*, filename: str | None = None, uri: str | None = None, default: str) -> str: - """Infer a canonical media type from explicit data or a filename.""" - if uri and uri.startswith("data:"): - media_type = uri[5:].split(";", 1)[0].split(",", 1)[0] - if "/" in media_type: - return media_type - - for candidate in (filename, urlparse(uri).path if uri else None): - if not candidate: - continue - candidate_lower = candidate.lower() - for suffix, canonical_media_type in _CANONICAL_MEDIA_TYPES.items(): - if candidate_lower.endswith(suffix): - return canonical_media_type - if guessed_media_type := mimetypes.guess_type(candidate)[0]: - return guessed_media_type - - return default - - class EntityNotFoundError(Exception): """Raised when an entity is not found.""" @@ -263,6 +236,7 @@ async def execute_streaming(self, request: AgentFrameworkRequest) -> AsyncGenera OpenAI response stream events """ try: + events: list[Any] = [] entity_id = request.get_entity_id() if not entity_id: logger.error("No entity_id specified in request") @@ -283,8 +257,13 @@ async def execute_streaming(self, request: AgentFrameworkRequest) -> AsyncGenera and cast(dict[str, Any], event).get("type") == "response.function_approval.requested" ): self._track_approval_request(cast(dict[str, Any], event)) + events.append(event) + if _get_event_type(event) == "response.failed": + continue yield event + yield await self.message_mapper.finalize_stream(events, request) + except Exception as e: logger.exception(f"Error in streaming execution: {e}") # Could yield error event here @@ -301,7 +280,14 @@ async def execute_sync(self, request: AgentFrameworkRequest) -> OpenAIResponse: # Collect all streaming events events = [event async for event in self.execute_streaming(request)] - # Aggregate into final response + for event in reversed(events): + if _get_event_type(event) not in ("response.completed", "response.failed"): + continue + response = getattr(event, "response", None) + if isinstance(response, OpenAIResponse): + return response + + # Preserve the existing empty-response fallback when execution produced no events. return await self.message_mapper.aggregate_to_response(events, request) async def execute_entity(self, entity_id: str, request: AgentFrameworkRequest) -> AsyncGenerator[Any]: @@ -722,7 +708,7 @@ def _convert_openai_input_to_chat_message(self, input_items: list[Any], Message: "detail": detail if isinstance(detail, str) else "auto", } if isinstance(image_url, str) and image_url: - media_type = _infer_media_type(uri=image_url, default="image/png") + media_type = infer_media_type(uri=image_url, default="image/png") contents.append( Content.from_uri( uri=image_url, @@ -758,7 +744,7 @@ def _convert_openai_input_to_chat_message(self, input_items: list[Any], Message: file_properties["detail"] = detail if isinstance(file_data, str) and file_data: - media_type = _infer_media_type( + media_type = infer_media_type( filename=filename, uri=file_data, default="application/octet-stream", @@ -776,7 +762,7 @@ def _convert_openai_input_to_chat_message(self, input_items: list[Any], Message: ) ) elif isinstance(file_url, str) and file_url: - media_type = _infer_media_type( + media_type = infer_media_type( filename=filename, uri=file_url, default="application/octet-stream", @@ -789,7 +775,7 @@ def _convert_openai_input_to_chat_message(self, input_items: list[Any], Message: ) ) elif isinstance(file_id, str) and file_id: - media_type = _infer_media_type( + media_type = infer_media_type( filename=filename, default="application/octet-stream", ) diff --git a/python/packages/devui/agent_framework_devui/_mapper.py b/python/packages/devui/agent_framework_devui/_mapper.py index acaab2e92fb..21cad790cfc 100644 --- a/python/packages/devui/agent_framework_devui/_mapper.py +++ b/python/packages/devui/agent_framework_devui/_mapper.py @@ -32,6 +32,7 @@ InputTokensDetails, OpenAIResponse, OutputTokensDetails, + ResponseCompletedEvent, ResponseErrorEvent, ResponseFunctionCallArgumentsDeltaEvent, ResponseFunctionResultComplete, @@ -296,6 +297,7 @@ async def aggregate_to_response(self, events: Sequence[Any], request: AgentFrame # Track text content parts per message (keyed by item_id) text_parts_by_message: dict[str, list[str]] = {} + message_order: list[str] = [] # Track function calls (keyed by call_id) to accumulate arguments function_calls: dict[str, dict[str, Any]] = {} @@ -303,6 +305,9 @@ async def aggregate_to_response(self, events: Sequence[Any], request: AgentFrame # Track function results (keyed by call_id) function_results: dict[str, dict[str, Any]] = {} + # Track complete message items that do not use the text-delta lifecycle. + complete_messages: dict[str, ResponseOutputMessage] = {} + for event in events: event_type = getattr(event, "type", None) @@ -311,6 +316,8 @@ async def aggregate_to_response(self, events: Sequence[Any], request: AgentFrame item_id = getattr(event, "item_id", "default") if item_id not in text_parts_by_message: text_parts_by_message[item_id] = [] + if item_id not in message_order: + message_order.append(item_id) text_parts_by_message[item_id].append(event.delta) # Handle output_item.added events (function_call, message, etc.) @@ -352,8 +359,11 @@ async def aggregate_to_response(self, events: Sequence[Any], request: AgentFrame # Other output items (message, etc.) - track for later elif item_type == "message": - # Messages will be built from text_parts_by_message - pass + if isinstance(item, ResponseOutputMessage): + if item.id not in message_order: + message_order.append(item.id) + if item.content: + complete_messages[item.id] = item # Handle function call arguments delta - accumulate arguments elif event_type == "response.function_call_arguments.delta": @@ -387,22 +397,21 @@ async def aggregate_to_response(self, events: Sequence[Any], request: AgentFrame # but we don't include them in the Response output _ = function_results # Acknowledge but don't use - # Build final text message from accumulated deltas - # Combine all text parts (usually there's just one message) - all_text_parts: list[str] = [] - for _item_id, parts in text_parts_by_message.items(): - all_text_parts.extend(parts) - - full_content = "".join(all_text_parts) - - # Only add message if there's text content - if full_content: + # Build messages in first-seen order. Workflow output messages can + # arrive complete, while agent messages use an empty item plus deltas. + for item_id in message_order: + if complete_message := complete_messages.get(item_id): + output_items.append(complete_message) + continue + full_content = "".join(text_parts_by_message.get(item_id, [])) + if not full_content: + continue response_output_text = ResponseOutputText(type="output_text", text=full_content, annotations=[]) response_output_message = ResponseOutputMessage( type="message", role="assistant", content=[response_output_text], - id=f"msg_{uuid.uuid4().hex[:8]}", + id=item_id if item_id != "default" else f"msg_{uuid.uuid4().hex[:8]}", status="completed", ) output_items.append(response_output_message) @@ -450,6 +459,34 @@ async def aggregate_to_response(self, events: Sequence[Any], request: AgentFrame logger.debug(f"Cleaned up context for request {request_key} after aggregation") self._usage_accumulator.pop(str(request_key), None) + async def finalize_stream( + self, events: Sequence[Any], request: AgentFrameworkRequest + ) -> ResponseCompletedEvent | ResponseFailedEvent: + """Create the single terminal event for a mapped response stream.""" + final_response = await self.aggregate_to_response(events, request) + + last_sequence_number = 0 + for event in reversed(events): + if getattr(event, "type", None) == "response.failed": + continue + sequence_number = getattr(event, "sequence_number", None) + if isinstance(sequence_number, int): + last_sequence_number = sequence_number + break + + if final_response.status == "failed": + return ResponseFailedEvent( + type="response.failed", + response=final_response, + sequence_number=last_sequence_number + 1, + ) + + return ResponseCompletedEvent( + type="response.completed", + response=final_response, + sequence_number=last_sequence_number + 1, + ) + def _get_or_create_context(self, request: AgentFrameworkRequest) -> dict[str, Any]: """Get or create conversion context for this request. diff --git a/python/packages/devui/agent_framework_devui/_server.py b/python/packages/devui/agent_framework_devui/_server.py index 6a83cfecac8..bc94dee9a21 100644 --- a/python/packages/devui/agent_framework_devui/_server.py +++ b/python/packages/devui/agent_framework_devui/_server.py @@ -1227,16 +1227,12 @@ async def _stream_execution( ) -> AsyncGenerator[str]: """Stream execution directly through executor.""" try: - # Collect events for the final terminal event - events: list[Any] = [] - # Get conversation_id for trace storage conversation_getter = getattr(request, "_get_conversation_id", None) conversation_id = conversation_getter() if callable(conversation_getter) else None # Stream all events async for event in executor.execute_streaming(request): - events.append(event) event_type_value = getattr(event, "type", None) event_type = event_type_value if isinstance(event_type_value, str) else None @@ -1249,11 +1245,6 @@ async def _stream_execution( except Exception as e: logger.debug(f"Failed to store trace event: {e}") - # Failure responses need aggregation so their terminal payload includes - # any output emitted before the failure. - if event_type == "response.failed": - continue - # IMPORTANT: Check model_dump_json FIRST because to_json() can have newlines (pretty-printing) # which breaks SSE format. model_dump_json() returns single-line JSON. if hasattr(event, "model_dump_json"): @@ -1271,37 +1262,6 @@ async def _stream_execution( payload = json.dumps(str(event)) yield f"data: {payload}\n\n" - # Aggregate to the final response before emitting its matching terminal event. - final_response = await executor.message_mapper.aggregate_to_response(events, request) - - # Use the next sequence number after the last event actually sent to the client. - last_seq = 0 - for event in reversed(events): - if getattr(event, "type", None) == "response.failed": - continue - sequence_number = getattr(event, "sequence_number", None) - if isinstance(sequence_number, int): - last_seq = sequence_number - break - - if final_response.status == "failed": - from openai.types.responses import ResponseFailedEvent - - terminal_event = ResponseFailedEvent( - type="response.failed", - response=final_response, - sequence_number=last_seq + 1, - ) - else: - from .models import ResponseCompletedEvent - - terminal_event = ResponseCompletedEvent( - type="response.completed", - response=final_response, - sequence_number=last_seq + 1, - ) - yield f"data: {terminal_event.model_dump_json()}\n\n" - # Send final done event yield "data: [DONE]\n\n" diff --git a/python/packages/devui/agent_framework_devui/_utils.py b/python/packages/devui/agent_framework_devui/_utils.py index 0ce40a61c70..2165a9592c7 100644 --- a/python/packages/devui/agent_framework_devui/_utils.py +++ b/python/packages/devui/agent_framework_devui/_utils.py @@ -5,14 +5,43 @@ import inspect import json import logging +import mimetypes from dataclasses import fields, is_dataclass from types import UnionType from typing import Any, Union, cast, get_args, get_origin, get_type_hints +from urllib.parse import urlparse from agent_framework import Message logger = logging.getLogger(__name__) +_CANONICAL_MEDIA_TYPES = { + ".jpeg": "image/jpeg", + ".jpg": "image/jpeg", + ".m4a": "audio/mp4", + ".mp3": "audio/mpeg", +} + + +def infer_media_type(*, filename: str | None = None, uri: str | None = None, default: str) -> str: + """Infer a canonical media type from explicit data or a filename.""" + if uri and uri.startswith("data:"): + media_type = uri[5:].split(";", 1)[0].split(",", 1)[0] + if "/" in media_type: + return media_type + + for candidate in (filename, urlparse(uri).path if uri else None): + if not candidate: + continue + candidate_lower = candidate.lower() + for suffix, canonical_media_type in _CANONICAL_MEDIA_TYPES.items(): + if candidate_lower.endswith(suffix): + return canonical_media_type + if guessed_media_type := mimetypes.guess_type(candidate)[0]: + return guessed_media_type + + return default + def _string_key_dict(value: object) -> dict[str, Any] | None: """Cast value to a dict.""" diff --git a/python/packages/devui/tests/devui/test_conversations.py b/python/packages/devui/tests/devui/test_conversations.py index 803fc8249e2..32eda04e3aa 100644 --- a/python/packages/devui/tests/devui/test_conversations.py +++ b/python/packages/devui/tests/devui/test_conversations.py @@ -157,6 +157,37 @@ async def test_add_items(): assert text_content.text == "Hello" +@pytest.mark.asyncio +async def test_add_items_accepts_assistant_output_text(): + """Assistant Responses history is accepted by the conversation parser.""" + store = InMemoryConversationStore() + conversation = store.create_conversation(metadata={"agent_id": "test_agent"}) + + created_items = await store.add_items( + conversation.id, + items=[ + { + "role": "assistant", + "content": [{"type": "output_text", "text": "Prior answer", "annotations": []}], + } + ], + ) + listed_items, _ = await store.list_items(conversation.id) + + created_message = created_items[0] + listed_message = listed_items[0] + assert isinstance(created_message, OpenAIMessage) + assert isinstance(listed_message, OpenAIMessage) + assert created_message.role == "assistant" + assert listed_message.role == "assistant" + assert created_message.content is not None + assert listed_message.content is not None + created_text = cast(InputTextContent, created_message.content[0]) + listed_text = cast(InputTextContent, listed_message.content[0]) + assert created_text.text == "Prior answer" + assert listed_text.text == "Prior answer" + + @pytest.mark.asyncio async def test_add_and_list_items_preserves_all_supported_message_parts(): """Conversation conversion retains message boundaries and every supported part.""" diff --git a/python/packages/devui/tests/devui/test_mapper.py b/python/packages/devui/tests/devui/test_mapper.py index 29b5240927f..89e06c41c15 100644 --- a/python/packages/devui/tests/devui/test_mapper.py +++ b/python/packages/devui/tests/devui/test_mapper.py @@ -376,6 +376,49 @@ async def test_failed_response_retains_partial_output( assert response.output_text == "Partial output" +async def test_failed_response_retains_populated_message_output_item( + mapper: MessageMapper, test_request: AgentFrameworkRequest +) -> None: + """A failed workflow response retains complete message output items.""" + output_events = await mapper.convert_event( + WorkflowEvent("output", executor_id="final_executor", data="Partial workflow output"), + test_request, + ) + failed_events = await mapper.convert_event( + AgentFailedEvent(error=RuntimeError("failed after workflow output")), + test_request, + ) + + response = await mapper.aggregate_to_response([*output_events, *failed_events], test_request) + + assert response.status == "failed" + assert response.error is not None + assert response.error.message == "failed after workflow output" + assert response.output_text == "Partial workflow output" + + +async def test_aggregation_preserves_delta_and_complete_message_order( + mapper: MessageMapper, test_request: AgentFrameworkRequest +) -> None: + """Delta-built and complete messages retain their first-seen order.""" + delta_events = await mapper.convert_event( + create_test_agent_update([Content.from_text(text="First")]), + test_request, + ) + complete_events = await mapper.convert_event( + WorkflowEvent("output", executor_id="final_executor", data="Second"), + test_request, + ) + + response = await mapper.aggregate_to_response([*delta_events, *complete_events], test_request) + + assert response.output_text == "FirstSecond" + assert [item.id for item in response.output if item.type == "message"] == [ + delta_events[0].item.id, + complete_events[0].item.id, + ] + + async def test_agent_run_response_mapping(mapper: MessageMapper, test_request: AgentFrameworkRequest) -> None: """Test that mapper handles complete AgentResponse (non-streaming).""" response = create_agent_run_response("Complete response from run()") diff --git a/python/packages/devui/tests/devui/test_server.py b/python/packages/devui/tests/devui/test_server.py index 310a7580d0e..56055fd137a 100644 --- a/python/packages/devui/tests/devui/test_server.py +++ b/python/packages/devui/tests/devui/test_server.py @@ -11,6 +11,7 @@ from collections.abc import AsyncGenerator from pathlib import Path from typing import Any +from unittest.mock import MagicMock import pytest from agent_framework import AgentResponseUpdate, Content @@ -44,22 +45,21 @@ class _FailedStreamingExecutor(AgentFrameworkExecutor): """Executor stub that emits text before a terminal failure.""" def __init__(self) -> None: - super().__init__(EntityDiscovery(None), MessageMapper()) - - async def execute_streaming(self, request: AgentFrameworkRequest) -> AsyncGenerator[Any]: - partial_update = AgentResponseUpdate( + discovery = MagicMock(spec=EntityDiscovery) + discovery.get_entity_info.return_value = MagicMock() + super().__init__(discovery, MessageMapper()) + + async def execute_entity( + self, entity_id: str, request: AgentFrameworkRequest + ) -> AsyncGenerator[AgentResponseUpdate | AgentFailedEvent]: + _ = entity_id, request + yield AgentResponseUpdate( contents=[Content.from_text(text="Partial output")], role="assistant", message_id="partial_message", response_id="partial_response", ) - for event in await self.message_mapper.convert_event(partial_update, request): - yield event - for event in await self.message_mapper.convert_event( - AgentFailedEvent(error=RuntimeError("failed after output")), - request, - ): - yield event + yield AgentFailedEvent(error=RuntimeError("failed after output")) # Note: test_entities_dir fixture is provided by conftest.py @@ -140,7 +140,7 @@ async def test_stream_execution_emits_one_failed_terminal_event_with_partial_out """Failed streams retain prior output and never emit a completion event.""" server = DevServer(auth_enabled=False) executor = _FailedStreamingExecutor() - request = AgentFrameworkRequest(input="hello", stream=True) + request = AgentFrameworkRequest(input="hello", stream=True, metadata={"entity_id": "failed"}) chunks = [chunk async for chunk in server._stream_execution(executor, request)] payloads = [json.loads(chunk.removeprefix("data: ").strip()) for chunk in chunks if chunk != "data: [DONE]\n\n"] diff --git a/python/packages/openai/agent_framework_openai/_chat_client.py b/python/packages/openai/agent_framework_openai/_chat_client.py index 55ff20950ad..d0a724aa583 100644 --- a/python/packages/openai/agent_framework_openai/_chat_client.py +++ b/python/packages/openai/agent_framework_openai/_chat_client.py @@ -1897,7 +1897,19 @@ def _prepare_content_for_openai( ret["summary"].append({"type": "summary_text", "text": content.text}) return ret case "data" | "uri": - if content.has_top_level_media_type("image"): + openai_content_type = content.additional_properties.get("openai_content_type") + if openai_content_type == "input_file": + filename = content.additional_properties.get("filename") + file_obj = { + "type": "input_file", + "file_data": content.uri, + } + if filename: + file_obj["filename"] = filename + return _attach_prompt_cache_breakpoint(file_obj, content) + if openai_content_type == "input_image" or ( + openai_content_type is None and content.has_top_level_media_type("image") + ): result: dict[str, Any] = { "type": "input_image", "image_url": content.uri, @@ -2048,6 +2060,15 @@ def _prepare_content_for_openai( # the citation context for round-tripping. if role == "assistant": return {} + openai_content_type = content.additional_properties.get("openai_content_type") + if openai_content_type == "input_image" or ( + openai_content_type is None and content.media_type and content.has_top_level_media_type("image") + ): + return { + "type": "input_image", + "file_id": content.file_id, + "detail": content.additional_properties.get("detail", "auto"), + } return { "type": "input_file", "file_id": content.file_id, diff --git a/python/packages/openai/tests/openai/test_openai_chat_client.py b/python/packages/openai/tests/openai/test_openai_chat_client.py index aab0fc1fe24..0cf9e621fc5 100644 --- a/python/packages/openai/tests/openai/test_openai_chat_client.py +++ b/python/packages/openai/tests/openai/test_openai_chat_client.py @@ -3661,6 +3661,55 @@ def test_hosted_file_content_preparation() -> None: assert result["file_id"] == "file_abc123" +def test_hosted_image_content_preparation() -> None: + """Hosted image IDs retain their image semantics and detail.""" + client = OpenAIChatClient(model="test-model", api_key="test-key") + hosted_image = Content.from_hosted_file( + file_id="file_image", + additional_properties={"openai_content_type": "input_image", "detail": "high"}, + ) + + result = client._prepare_content_for_openai("user", hosted_image) + + assert result == { + "type": "input_image", + "file_id": "file_image", + "detail": "high", + } + + explicit_image_file = Content.from_hosted_file( + file_id="file_photo", + media_type="image/jpeg", + name="photo.jpg", + additional_properties={"openai_content_type": "input_file", "filename": "photo.jpg"}, + ) + + result = client._prepare_content_for_openai("user", explicit_image_file) + + assert result == { + "type": "input_file", + "file_id": "file_photo", + } + + +def test_explicit_input_file_overrides_image_media_type() -> None: + """Explicit input-file semantics take precedence over inferred image media.""" + client = OpenAIChatClient(model="test-model", api_key="test-key") + image_file = Content.from_uri( + uri="data:image/jpeg;base64,abc", + media_type="image/jpeg", + additional_properties={"openai_content_type": "input_file", "filename": "scan.jpg"}, + ) + + result = client._prepare_content_for_openai("user", image_file) + + assert result == { + "type": "input_file", + "file_data": "data:image/jpeg;base64,abc", + "filename": "scan.jpg", + } + + def test_assistant_text_preserves_citation_annotations_on_roundtrip() -> None: """Citation annotations on assistant text should survive serialization back to the Responses API.