Python: [Python] structured_outputs_transform reuses ChatHistory across calls (prompt pollution) #14483

import asyncio
from typing import Any
from pydantic import BaseModel, PrivateAttr
from semantic_kernel.agents.orchestration.tools import structured_outputs_transform
from semantic_kernel.connectors.ai.chat_completion_client_base import ChatCompletionClientBase
from semantic_kernel.connectors.ai.prompt_execution_settings import PromptExecutionSettings
from semantic_kernel.contents import ChatMessageContent
from semantic_kernel.contents.utils.author_role import AuthorRole

class Out(BaseModel):
    answer: str

class FakeSettings(PromptExecutionSettings):
    response_format: Any = None

class FakeService(ChatCompletionClientBase):
    _calls: list = PrivateAttr(default_factory=list)
    def __init__(self) -> None:
        super().__init__(service_id="fake", ai_model_id="fake")
    def get_prompt_execution_settings_class(self):
        return FakeSettings
    async def get_chat_message_contents(self, chat_history, settings, **kwargs):
        self._calls.append(len(chat_history.messages))
        return [ChatMessageContent(role=AuthorRole.ASSISTANT, content='{"answer":"x"}')]
    async def get_chat_message_content(self, chat_history, settings, **kwargs):
        self._calls.append(len(chat_history.messages))
        return ChatMessageContent(role=AuthorRole.ASSISTANT, content='{"answer":"x"}')

async def main():
    svc = FakeService()
    transform = structured_outputs_transform(Out, svc)
    await transform(ChatMessageContent(role=AuthorRole.USER, content="first"))
    await transform(ChatMessageContent(role=AuthorRole.USER, content="second"))
    await transform(ChatMessageContent(role=AuthorRole.USER, content="third"))
    print(svc._calls)  # ACTUAL: [2, 3, 4]  EXPECTED: [2, 2, 2]

asyncio.run(main())