Python: fix: decode entities once in ChatHistory.from_rendered_prompt

python/semantic_kernel/contents/chat_history.py

@@ -1,11 +1,13 @@
 # Copyright (c) Microsoft. All rights reserved.
 
 import logging
+import re
 from collections.abc import Generator, Iterable
 from functools import singledispatchmethod
 from html import unescape
 from typing import Any, TypeVar
 from xml.etree.ElementTree import Element, tostring  # nosec
+from xml.sax.saxutils import escape  # nosec
 
 from defusedxml.ElementTree import XML, ParseError
 from pydantic import Field, field_validator, model_validator
@@ -22,6 +24,8 @@
 
 _T = TypeVar("_T", bound="ChatHistory")
 
+_CDATA_SECTION = re.compile(r"<!\[CDATA\[(.*?)\]\]>", re.DOTALL)
+
 
 class ChatHistory(KernelBaseModel):
     """This class holds the history of chat messages from a chat conversation.
@@ -340,13 +344,17 @@ def from_rendered_prompt(cls: type[_T], rendered_prompt: str) -> _T:
         prompt_tag = "root"
         messages: list["ChatMessageContent"] = []
         prompt = rendered_prompt.strip()
+        # The XML parser decodes entities in normal text but keeps a CDATA section literal, and the
+        # template engine escapes values wherever they land. Decode CDATA once here, so every text
+        # is decoded exactly once by the parser and not again afterwards.
+        xml_text = _CDATA_SECTION.sub(lambda match: escape(unescape(match.group(1))), prompt)
         try:
-            xml_prompt = XML(text=f"<{prompt_tag}>{prompt}</{prompt_tag}>")
+            xml_prompt = XML(text=f"<{prompt_tag}>{xml_text}</{prompt_tag}>")
         except ParseError as exc:
             logger.info(f"Could not parse prompt {prompt} as xml, treating as text, error was: {exc}")
             return cls(messages=[ChatMessageContent(role=AuthorRole.USER, content=unescape(prompt))])
         if xml_prompt.text and xml_prompt.text.strip():
-            messages.append(ChatMessageContent(role=AuthorRole.SYSTEM, content=unescape(xml_prompt.text.strip())))
+            messages.append(ChatMessageContent(role=AuthorRole.SYSTEM, content=xml_prompt.text.strip()))
         for item in xml_prompt:
             if item.tag == CHAT_MESSAGE_CONTENT_TAG:
                 messages.append(ChatMessageContent.from_element(item))
@@ -368,10 +376,10 @@ def from_rendered_prompt(cls: type[_T], rendered_prompt: str) -> _T:
                 # For unknown tags the tail is part of the surrounding text,
                 # so keep it in the same message instead of starting a new one.
                 if item.tail:
-                    messages[-1].content = (messages[-1].content or "") + unescape(item.tail)
+                    messages[-1].content = (messages[-1].content or "") + item.tail
                 continue
             if item.tail and item.tail.strip():
-                messages.append(ChatMessageContent(role=AuthorRole.USER, content=unescape(item.tail.strip())))
+                messages.append(ChatMessageContent(role=AuthorRole.USER, content=item.tail.strip()))
         if len(messages) == 1 and messages[0].role == AuthorRole.SYSTEM:
             messages[0].role = AuthorRole.USER
         return cls(messages=messages)

python/semantic_kernel/contents/chat_message_content.py

@@ -268,7 +268,7 @@ def from_element(cls, element: Element) -> "ChatMessageContent":
         kwargs: dict[str, Any] = {key: value for key, value in element.items()}
         items: list[KernelContent] = []
         if element.text:
-            items.append(TextContent(text=unescape(element.text)))
+            items.append(TextContent(text=element.text))
         for child in element:
             if child.tag not in TAG_CONTENT_MAP:
                 logger.warning('Unknown tag "%s" in ChatMessageContent, treating as text', child.tag)

python/semantic_kernel/contents/reasoning_content.py

@@ -2,7 +2,6 @@
 
 """Reasoning content model."""
 
-from html import unescape
 from typing import ClassVar, Literal, TypeVar
 from xml.etree.ElementTree import Element  # nosec
 
@@ -52,7 +51,7 @@ def from_element(cls: type[_T], element: Element) -> _T:
         """Create an instance from an XML Element."""
         if element.tag != cls.tag:
             raise ContentInitializationError(f"Element tag is not {cls.tag}")  # pragma: no cover
-        return cls(text=unescape(element.text) if element.text else "")
+        return cls(text=element.text or "")
 
     def to_dict(self) -> dict[str, str]:
         """Convert the instance to a dictionary suitable for message serialization."""

python/semantic_kernel/contents/text_content.py

@@ -1,6 +1,5 @@
 # Copyright (c) Microsoft. All rights reserved.
 
-from html import unescape
 from typing import ClassVar, Literal, TypeVar
 from xml.etree.ElementTree import Element  # nosec
 
@@ -52,7 +51,7 @@ def from_element(cls: type[_T], element: Element) -> _T:
         if element.tag != cls.tag:
             raise ContentInitializationError(f"Element tag is not {cls.tag}")  # pragma: no cover
 
-        return cls(text=unescape(element.text) if element.text else "", encoding=element.get("encoding", None))
+        return cls(text=element.text or "", encoding=element.get("encoding", None))
 
     def to_dict(self) -> dict[str, str]:
         """Convert the instance to a dictionary."""

python/tests/unit/contents/test_chat_history.py

@@ -670,3 +670,66 @@ def __str__(self) -> str:
     ])
     chat_history.add_tool_message([FunctionResultContent(id="test1", result=custom_result)])
     assert "CustomResultTestValue" in chat_history.serialize()
+
+
+ENTITY_TEXTS = [
+    "What does & mean in HTML?",
+    "Write <b> as text",
+    "Use : as a colon",
+    "Tom & Jerry <3",
+]
+
+
+@pytest.mark.parametrize("text", ENTITY_TEXTS)
+async def test_template_message_input_is_decoded_once(text: str):
+    template = '<message role="user">{{$input}}</message>'
+    rendered = await KernelPromptTemplate(
+        prompt_template_config=PromptTemplateConfig(name="test", description="test", template=template)
+    ).render(kernel=Kernel(), arguments=KernelArguments(input=text))
+
+    chat_history = ChatHistory.from_rendered_prompt(rendered)
+    assert chat_history.messages[0].role == AuthorRole.USER
+    assert chat_history.messages[0].content == text
+
+
+@pytest.mark.parametrize("text", ENTITY_TEXTS)
+async def test_template_plain_input_is_decoded_once(text: str):
+    template = "{{$input}}"
+    rendered = await KernelPromptTemplate(
+        prompt_template_config=PromptTemplateConfig(name="test", description="test", template=template)
+    ).render(kernel=Kernel(), arguments=KernelArguments(input=text))
+
+    chat_history = ChatHistory.from_rendered_prompt(rendered)
+    assert chat_history.messages[0].content == text
+
+
+@pytest.mark.parametrize("text", ENTITY_TEXTS)
+async def test_template_text_after_message_is_decoded_once(text: str):
+    template = '<message role="system">rules</message>{{$input}}'
+    rendered = await KernelPromptTemplate(
+        prompt_template_config=PromptTemplateConfig(name="test", description="test", template=template)
+    ).render(kernel=Kernel(), arguments=KernelArguments(input=text))
+
+    chat_history = ChatHistory.from_rendered_prompt(rendered)
+    assert chat_history.messages[1].content == text
+
+
+@pytest.mark.parametrize("text", ENTITY_TEXTS)
+async def test_template_cdata_input_is_decoded_once(text: str):
+    template = '<message role="user"><![CDATA[{{$input}}]]></message>'
+    rendered = await KernelPromptTemplate(
+        prompt_template_config=PromptTemplateConfig(name="test", description="test", template=template)
+    ).render(kernel=Kernel(), arguments=KernelArguments(input=text))
+
+    chat_history = ChatHistory.from_rendered_prompt(rendered)
+    assert chat_history.messages[0].content == text
+
+
+@pytest.mark.parametrize("text", ENTITY_TEXTS)
+def test_chat_history_prompt_round_trip_keeps_entities(text: str):
+    chat_history = ChatHistory()
+    chat_history.add_user_message(text)
+    chat_history.add_assistant_message(text)
+
+    restored = ChatHistory.from_rendered_prompt(chat_history.to_prompt())
+    assert [message.content for message in restored.messages] == [text, text]