Repository: meta-llama/llama-stack
File: src/llama_stack/providers/inline/agents/meta_reference/responses/openai_responses.py (Python)

Code before:
                # Store and sync before yielding terminal events
                # This ensures the storage/syncing happens even if the consumer breaks after receiving the event
                if (
                    stream_chunk.type in {"response.completed", "response.incomplete"}
                    and final_response
                    and failed_response is None
                ):
                    messages_to_store = list(
                        filter(lambda x: not isinstance(x, OpenAISystemMessageParam), orchestrator.final_messages)
                    )
                    if store:
                        # TODO: we really should work off of output_items instead of "final_messages"
                        await self._store_response(
                            response=final_response,
                            input=all_input,
                            messages=messages_to_store,
                        )

                    if conversation:
                        await self._sync_response_to_conversation(conversation, input, output_items)
                        await self.responses_store.store_conversation_messages(conversation, messages_to_store)

                yield stream_chunk
        finally:
            # Clean up MCP sessions at the end of the request (fix for #4452)
            await mcp_session_manager.close_all()

    async def delete_openai_response(self, response_id: str) -> OpenAIDeleteResponseObject:
        return await self.responses_store.delete_response_object(response_id)

    async def _sync_response_to_conversation(
        self, conversation_id: str, input: str | list[OpenAIResponseInput] | None, output_items: list[ConversationItem]
    ) -> None:
        """Sync content and response messages to the conversation."""
        # Type as ConversationItem union to avoid list invariance issues
        conversation_items: list[ConversationItem] = []

        if isinstance(input, str):
            conversation_items.append(
                OpenAIResponseMessage(role="user", content=[OpenAIResponseInputMessageContentText(text=input)])
            )
        elif isinstance(input, list):
            conversation_items.extend(input)

        conversation_items.extend(output_items)

        adapter = TypeAdapter(list[ConversationItem])
        validated_items = adapter.validate_python(conversation_items)
        await self.conversations_api.add_items(conversation_id, AddItemsRequest(items=validated_items))


Review comment (suggestion):
Given the amount of things happening in `close_all()`, shouldn't we wrap it with `asyncio.shield()` to prevent halfway cancellations which would leak resources (sockets, fd...)? shield() prevents cancellation from aborting cleanup, ensuring all sessions are closed deterministically before the task fully cancels.

Code after:
                        item = stream_chunk.item
                        output_items.append(item)
                    case _:
                        pass  # Other event types

                # Store and sync before yielding terminal events
                # This ensures the storage/syncing happens even if the consumer breaks after receiving the event
                if (
                    stream_chunk.type in {"response.completed", "response.incomplete"}
                    and final_response
                    and failed_response is None
                ):
                    messages_to_store = list(
                        filter(lambda x: not isinstance(x, OpenAISystemMessageParam), orchestrator.final_messages)
                    )
                    if store:
                        # TODO: we really should work off of output_items instead of "final_messages"
                        await self._store_response(
                            response=final_response,
                            input=all_input,
                            messages=messages_to_store,
                        )

                    if conversation:
                        await self._sync_response_to_conversation(conversation, input, output_items)
                        await self.responses_store.store_conversation_messages(conversation, messages_to_store)

                yield stream_chunk
        finally:
            # Clean up MCP sessions at the end of the request (fix for #4452)
            # Use shield() to prevent cancellation from interrupting cleanup and leaking resources
            # Wrap in try/except as cleanup errors should not mask the original response
            try:
                await asyncio.shield(mcp_session_manager.close_all())
            except BaseException as e:
                # Debug level - cleanup errors are expected in streaming scenarios where
                # anyio cancel scopes may be in a different task context
                logger.debug(f"Error during MCP session cleanup: {e}")

    async def delete_openai_response(self, response_id: str) -> OpenAIDeleteResponseObject:
        return await self.responses_store.delete_response_object(response_id)

    async def _sync_response_to_conversation(
        self, conversation_id: str, input: str | list[OpenAIResponseInput] | None, output_items: list[ConversationItem]
    ) -> None:
        """Sync content and response messages to the conversation."""
        # Type as ConversationItem union to avoid list invariance issues
        conversation_items: list[ConversationItem] = []

        if isinstance(input, str):
            conversation_items.append(