diff --git a/python/packages/devui/agent_framework_devui/_conversations.py b/python/packages/devui/agent_framework_devui/_conversations.py
index 9ad8534e18..2f3fe8feb4 100644
--- a/python/packages/devui/agent_framework_devui/_conversations.py
+++ b/python/packages/devui/agent_framework_devui/_conversations.py
@@ -14,7 +14,7 @@
 from collections.abc import MutableSequence
 from typing import Any, Literal, cast
 
-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
@@ -28,6 +28,8 @@
     ResponseInputImage,
 )
 
+from ._utils import infer_media_type
+
 # Type alias for OpenAI Message role literals
 MessageRole = Literal["unknown", "user", "assistant", "system", "critic", "discriminator", "developer", "tool"]
 
@@ -304,17 +306,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)
 
-            chat_msg = Message(role=role, contents=[text])
+            if not contents:
+                raise ValueError("Conversation message did not contain any supported content")
+
+            chat_msg = Message(role=role, contents=contents)
             chat_messages.append(chat_msg)
 
         # Add messages to internal storage
@@ -328,9 +336,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 +398,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 +521,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", "output_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 f6ea02b5cf..35fcf8e680 100644
--- a/python/packages/devui/agent_framework_devui/_executor.py
+++ b/python/packages/devui/agent_framework_devui/_executor.py
@@ -15,11 +15,14 @@
 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"}
+
 
 def _get_event_type(event: Any) -> str | None:
     """Safely get the type of an event, handling both objects and dicts."""
@@ -35,6 +38,10 @@ class EntityNotFoundError(Exception):
     pass
 
 
+class InputConversionError(ValueError):
+    """Raised when OpenAI input contains no supported Agent Framework content."""
+
+
 class AgentFrameworkExecutor:
     """Executor for Agent Framework entities - agents and workflows."""
 
@@ -229,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")
@@ -249,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
@@ -267,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]:
@@ -592,6 +612,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 +634,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 +646,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 +660,12 @@ 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] = []
+        approval_request_ids: set[str] = set()
 
         # Process each input item
         for item in input_items:
@@ -649,9 +673,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):
@@ -664,78 +694,97 @@ 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))
 
                                 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,
                                             )
                                         )
 
@@ -752,7 +801,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 "
@@ -786,6 +839,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",
@@ -801,24 +855,23 @@ 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 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.)
 
-        # 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)
+        for request_id in approval_request_ids:
+            self._pending_approvals.pop(request_id, None)
 
-        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 +910,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 +949,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 929f47e349..21cad790cf 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,
@@ -280,12 +281,23 @@ 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,
+            )
             # Collect output items in order
             output_items: list[Any] = []
 
             # 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]] = {}
@@ -293,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)
 
@@ -301,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.)
@@ -342,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":
@@ -377,28 +397,27 @@ 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)
 
             # 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",
@@ -413,26 +432,16 @@ 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",
                 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",
@@ -441,13 +450,42 @@ 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)
+
+    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.
@@ -468,8 +506,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 +861,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 +890,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 +937,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 +949,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 +1065,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 +1093,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 +1888,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/agent_framework_devui/_server.py b/python/packages/devui/agent_framework_devui/_server.py
index 5a6e3e917a..bc94dee9a2 100644
--- a/python/packages/devui/agent_framework_devui/_server.py
+++ b/python/packages/devui/agent_framework_devui/_server.py
@@ -1227,19 +1227,17 @@ async def _stream_execution(
     ) -> AsyncGenerator[str]:
         """Stream execution directly through executor."""
         try:
-            # Collect events for final response.completed 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
 
                 # 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):
@@ -1264,28 +1262,6 @@ 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
-
-            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
-            last_seq = 0
-            for event in reversed(events):
-                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"
-
             # 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 0ce40a61c7..2165a9592c 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_approval_validation.py b/python/packages/devui/tests/devui/test_approval_validation.py
index d8de793310..0bc73724b2 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,37 @@ 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_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:
diff --git a/python/packages/devui/tests/devui/test_conversations.py b/python/packages/devui/tests/devui/test_conversations.py
index bd999a81ce..32eda04e3a 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
 
@@ -155,6 +157,129 @@ 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."""
+    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)
+
+    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"
+    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
+    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
 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 b37647b05d..89e06c41c1 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,89 @@ 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_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_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_multimodal_workflow.py b/python/packages/devui/tests/devui/test_multimodal_workflow.py
index 7af7f3f308..4699c813c5 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,124 @@ 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"},
+                ],
+            },
+            {
+                "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", "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):
+        """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)
+
+    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))
+        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."""
diff --git a/python/packages/devui/tests/devui/test_server.py b/python/packages/devui/tests/devui/test_server.py
index 9ebd4ee40d..56055fd137 100644
--- a/python/packages/devui/tests/devui/test_server.py
+++ b/python/packages/devui/tests/devui/test_server.py
@@ -4,24 +4,31 @@
 
 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
+from unittest.mock import MagicMock
 
 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 +41,27 @@ 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:
+        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",
+        )
+        yield AgentFailedEvent(error=RuntimeError("failed after output"))
+
+
 # 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, 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"]
+    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)
diff --git a/python/packages/openai/agent_framework_openai/_chat_client.py b/python/packages/openai/agent_framework_openai/_chat_client.py
index 55ff20950a..d0a724aa58 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 aab0fc1fe2..0cf9e621fc 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.