[Python] Durable thread compaction and history providers (ADR 0032)

.gitignore

@@ -445,3 +445,7 @@ FodyWeavers.xsd
 *.msix
 *.msm
 *.msp
+
+# Local environment files with credentials (templates use .env.example and stay tracked).
+# Must come after the '!python/packages/**' negation above so it wins for test .env files.
+**/.env

python/packages/azurefunctions/agent_framework_azurefunctions/_app.py

@@ -44,6 +44,11 @@
     execute_workflow_activity,
     plan_workflow_registration,
 )
+from agent_framework_durabletask._retention import (
+    DEFAULT_MAX_STATE_BYTES,
+    DEFAULT_RETENTION,
+    RetentionMode,
+)
 from agent_framework_durabletask._workflows.naming import (
     SUBWORKFLOW_REQUEST_SEPARATOR,
     split_subworkflow_request_id,
@@ -244,6 +249,9 @@ def __init__(
         poll_interval_seconds: float = DEFAULT_POLL_INTERVAL_SECONDS,
         enable_mcp_tool_trigger: bool = False,
         default_callback: AgentResponseCallbackProtocol | None = None,
+        retention: RetentionMode = DEFAULT_RETENTION,
+        workflow_retention: RetentionMode | None = None,
+        max_state_bytes: int = DEFAULT_MAX_STATE_BYTES,
     ):
         """Initialize the AgentFunctionApp.
 
@@ -263,6 +271,16 @@ def __init__(
         :param poll_interval_seconds: Delay in seconds between polling attempts.
             Defaults to ``DEFAULT_POLL_INTERVAL_SECONDS``.
         :param default_callback: Optional callback invoked for agents without specific callbacks.
+        :param retention: Default conversation retention for agents hosted by this app, including
+            agents inside hosted workflows. ``auto`` deletes only under storage pressure,
+            ``keep_all`` never deletes and lets the entity fail at the backend limit, and
+            ``follow_compaction`` first deletes what compaction excluded, then uses the same
+            pressure eviction as ``auto`` if that is not enough. ``add_agent`` can override it
+            per agent.
+        :param max_state_bytes: Budget for serialized entity state.
+        :param workflow_retention: Retention for agent nodes inside hosted workflows. When None,
+            ``retention`` applies. Worth setting separately, since a workflow node's entity lives
+            for one orchestration while a standalone agent's can live indefinitely.
 
         :note: If no agents are provided, they can be added later using :meth:`add_agent`.
         """
@@ -283,6 +301,9 @@ def __init__(
         self.enable_http_endpoints = enable_http_endpoints
         self.enable_mcp_tool_trigger = enable_mcp_tool_trigger
         self.default_callback = default_callback
+        self._retention: RetentionMode = retention
+        self._workflow_retention: RetentionMode | None = workflow_retention
+        self._max_state_bytes = max_state_bytes
 
         try:
             retries = int(max_poll_retries)
@@ -419,6 +440,7 @@ def _register_workflow_primitives(self, workflow: Workflow) -> None:
                 agent_executor.agent,
                 callback=self.default_callback,
                 entity_id=workflow_scoped_executor_id(workflow.name, agent_executor.id),
+                retention=self._workflow_retention,
             )
         for executor in plan.activity_executors:
             # Set up a Functions activity trigger for each non-agent executor, scoped
@@ -826,6 +848,7 @@ def add_agent(
         enable_mcp_tool_trigger: bool | None = None,
         *,
         entity_id: str | None = None,
+        retention: RetentionMode | None = None,
     ) -> None:
         """Add an agent to the function app after initialization.
 
@@ -842,6 +865,7 @@ def add_agent(
                 durable entity (and the ``agents`` / ``get_agent`` key) matches the
                 identity the orchestrator dispatches to. Mirrors
                 ``DurableAIAgentWorker.add_agent(entity_id=...)``.
+            retention: Per-agent retention override. When None, the app-level setting is used.
 
         Raises:
             ValueError: If the agent doesn't have a 'name' attribute.
@@ -890,9 +914,15 @@ def add_agent(
         )
 
         effective_callback = callback or self.default_callback
+        effective_retention: RetentionMode = self._retention if retention is None else retention
 
         self._setup_agent_functions(
-            agent, registration_name, effective_callback, effective_enable_http_endpoint, effective_enable_mcp_endpoint
+            agent,
+            registration_name,
+            effective_callback,
+            effective_enable_http_endpoint,
+            effective_enable_mcp_endpoint,
+            retention=effective_retention,
         )
 
         logger.debug(f"[AgentFunctionApp] Agent '{registration_name}' added successfully")
@@ -937,6 +967,8 @@ def _setup_agent_functions(
         callback: AgentResponseCallbackProtocol | None,
         enable_http_endpoint: bool,
         enable_mcp_tool_trigger: bool,
+        *,
+        retention: RetentionMode = DEFAULT_RETENTION,
     ) -> None:
         """Set up the HTTP trigger, entity, and MCP tool trigger for a specific agent.
 
@@ -946,6 +978,7 @@ def _setup_agent_functions(
             callback: Optional callback to receive response updates
             enable_http_endpoint: Whether to create HTTP endpoint
             enable_mcp_tool_trigger: Whether to create MCP tool trigger
+            retention: How much of the conversation durable state may discard.
         """
         logger.debug(f"[AgentFunctionApp] Setting up functions for agent '{agent_name}'...")
 
@@ -956,7 +989,7 @@ def _setup_agent_functions(
                 "[AgentFunctionApp] HTTP run route disabled for agent '%s'",
                 agent_name,
             )
-        self._setup_agent_entity(agent, agent_name, callback)
+        self._setup_agent_entity(agent, agent_name, callback, retention=retention)
 
         if enable_mcp_tool_trigger:
             agent_description = agent.description
@@ -1098,13 +1131,16 @@ def _setup_agent_entity(
         agent: SupportsAgentRun,
         agent_name: str,
         callback: AgentResponseCallbackProtocol | None,
+        *,
+        retention: RetentionMode = DEFAULT_RETENTION,
     ) -> None:
         """Register the durable entity responsible for agent state.
 
         Args:
             agent: The agent instance
             agent_name: The agent name (used for both entity identification and function naming)
             callback: Optional callback for response updates
+            retention: How much of the conversation durable state may discard.
         """
         # Use the prefixed entity name for both registration and function naming
         entity_name_with_prefix = AgentSessionId.to_entity_name(agent_name)
@@ -1117,7 +1153,7 @@ def entity_function(context: df.DurableEntityContext) -> None:
             - run_agent: (Deprecated) Execute the agent with a message
             - reset: Clear conversation history
             """
-            entity_handler = create_agent_entity(agent, callback)
+            entity_handler = create_agent_entity(agent, callback, retention=retention)
             entity_handler(context)
 
         # Set function name for Azure Functions (used in function.json generation)

python/packages/azurefunctions/agent_framework_azurefunctions/_entities.py

@@ -21,6 +21,7 @@
     AgentResponseCallbackProtocol,
     run_agent_coroutine,
 )
+from agent_framework_durabletask._retention import DEFAULT_MAX_STATE_BYTES, DEFAULT_RETENTION, RetentionMode
 
 logger = logging.getLogger("agent_framework.azurefunctions")
 
@@ -47,17 +48,29 @@ def _set_state_dict(self, state: dict[str, Any]) -> None:
     def _get_session_id_from_entity(self) -> str:
         return str(self._context.entity_key)
 
+    def _get_entity_name_from_entity(self) -> str:
+        return str(self._context.entity_name)
+
 
 def create_agent_entity(
     agent: SupportsAgentRun,
     callback: AgentResponseCallbackProtocol | None = None,
+    *,
+    retention: RetentionMode = DEFAULT_RETENTION,
+    max_state_bytes: int = DEFAULT_MAX_STATE_BYTES,
 ) -> Callable[[df.DurableEntityContext], None]:
     """Factory function to create an agent entity class.
 
     Args:
         agent: The Microsoft Agent Framework agent instance (must implement SupportsAgentRun)
         callback: Optional callback invoked during streaming and final responses
 
+    Keyword Args:
+        retention: How much of the conversation durable state may discard. ``auto`` deletes only
+            under storage pressure, ``keep_all`` never deletes, and ``follow_compaction`` first
+            deletes what compaction excluded, then uses pressure eviction if needed.
+        max_state_bytes: Budget for serialized entity state.
+
     Returns:
         Entity function configured with the agent
     """
@@ -69,7 +82,13 @@ async def _entity_coroutine(context: df.DurableEntityContext) -> None:
             logger.debug("[entity_function] Operation: %s", context.operation_name)
 
             state_provider = AzureFunctionEntityStateProvider(context)
-            entity = AgentEntity(agent, callback, state_provider=state_provider)
+            entity = AgentEntity(
+                agent,
+                callback,
+                state_provider=state_provider,
+                retention=retention,
+                max_state_bytes=max_state_bytes,
+            )
 
             operation = context.operation_name
 

python/packages/azurefunctions/agent_framework_azurefunctions/_orchestration.py

@@ -149,28 +149,8 @@ def __init__(self, context: AgentOrchestrationContextType):
     def generate_unique_id(self) -> str:
         return str(self.context.new_uuid())
 
-    def get_run_request(
-        self,
-        message: str,
-        *,
-        options: dict[str, Any] | None = None,
-    ) -> RunRequest:
-        """Get the current run request from the orchestration context.
-
-        Args:
-            message: The message to send to the agent
-            options: Optional options dictionary. Supported keys include
-                ``response_format``, ``enable_tool_calls``, and ``wait_for_response``.
-                Additional keys are forwarded to the agent execution.
-
-        Returns:
-            RunRequest: The current run request
-        """
-        # Create a copy to avoid modifying the caller's dict
-
-        request = super().get_run_request(message, options=options)
-        request.orchestration_id = self.context.instance_id
-        return request
+    def _orchestration_id(self) -> str | None:
+        return self.context.instance_id
 
     def run_durable_agent(
         self,

python/packages/azurefunctions/agent_framework_azurefunctions/_workflow_af_context.py

@@ -12,7 +12,7 @@
 from datetime import datetime
 from typing import Any
 
-from agent_framework_durabletask import AgentSessionId, DurableAgentSession, DurableAIAgent
+from agent_framework_durabletask import WorkflowOrchestrationContext, build_agent_task
 from azure.durable_functions import DurableOrchestrationContext
 
 from ._orchestration import AzureFunctionsAgentExecutor
@@ -56,12 +56,20 @@ def current_utc_datetime(self) -> datetime:
 
     # -- Agent / Activity dispatch --------------------------------------------
 
-    def prepare_agent_task(self, executor_id: str, message: str, orchestration_instance_id: str) -> Any:
-        session_id = AgentSessionId(name=executor_id, key=orchestration_instance_id)
-        session = DurableAgentSession(durable_session_id=session_id)
-        az_executor = AzureFunctionsAgentExecutor(self._context)
-        agent = DurableAIAgent(az_executor, executor_id)
-        return agent.run(message, session=session)
+    def prepare_agent_task(
+        self,
+        executor_id: str,
+        message: str,
+        orchestration_instance_id: str,
+        context_messages: list[dict[str, Any]] | None = None,
+    ) -> Any:
+        return build_agent_task(
+            AzureFunctionsAgentExecutor(self._context),
+            executor_id,
+            message,
+            orchestration_instance_id,
+            context_messages,
+        )
 
     def prepare_activity_task(self, activity_name: str, input_json: str) -> Any:
         orchestration_context: Any = self._context
@@ -103,3 +111,8 @@ def cancel_task(self, task: Any) -> None:
 
     def get_task_result(self, task: Any) -> Any:
         return getattr(task, "result", None)
+
+
+# Ensure the adapter satisfies the protocol. Validated statically by the type checker,
+# so a signature change on the protocol is caught here rather than at a distant call site.
+_protocol_check: type[WorkflowOrchestrationContext] = AzureFunctionsWorkflowContext

python/packages/azurefunctions/tests/integration_tests/test_01_single_agent.py

@@ -93,27 +93,31 @@ def test_legacy_thread_id_in_query_still_accepted(self) -> None:
         assert response.headers.get("x-ms-thread-id") is None
 
     def test_conversation_continuity(self) -> None:
-        """Test conversation context is maintained across requests."""
+        """History must accumulate *and* reach the model on later turns."""
         session_id = "test-continuity"
 
-        # First message
+        # First message establishes a fact that exists nowhere else.
         response1 = self.helper.post_json(
             f"{self.base_url}/run",
-            {"message": "Tell me a short joke about weather in Seattle.", "session_id": session_id},
+            {"message": "My favorite animal is the axolotl. Tell me a short joke about it.", "session_id": session_id},
         )
         assert response1.status_code in [200, 202]
 
         if response1.status_code == 200:
             data1 = response1.json()
             assert data1["message_count"] == 2  # Initial + reply
 
-            # Second message in same session
+            # Second message in same session; only answerable from persisted history.
             response2 = self.helper.post_json(
-                f"{self.base_url}/run", {"message": "What about San Francisco?", "session_id": session_id}
+                f"{self.base_url}/run",
+                {"message": "What is my favorite animal? Reply with just the animal name.", "session_id": session_id},
             )
             assert response2.status_code == 200
             data2 = response2.json()
             assert data2["message_count"] == 4
+            assert "axolotl" in str(data2["response"]).lower(), (
+                f"Agent lost conversation context across turns. Got: {data2['response']!r}"
+            )
         else:
             # In async mode, we can't easily test message count
             # Just verify we can make multiple calls

python/packages/azurefunctions/tests/integration_tests/test_14_conversation_compaction.py

@@ -0,0 +1,83 @@
+# Copyright (c) Microsoft. All rights reserved.
+"""
+Integration Tests for the Conversation Compaction Sample
+
+Verifies that an agent configured the ordinary core way - an in-memory history provider plus a
+compaction provider - runs durably under the Azure Functions host with no durable-specific
+configuration, mirroring the standalone durabletask coverage.
+
+The function app is automatically started by the test fixture.
+
+Prerequisites:
+- Azure OpenAI credentials configured (see packages/azurefunctions/tests/integration_tests/.env.example)
+- Azurite or Azure Storage account configured
+
+Usage:
+    uv run pytest packages/azurefunctions/tests/integration_tests/test_14_conversation_compaction.py -v
+"""
+
+import uuid
+
+import pytest
+
+# Matches function_app.py: only the most recent groups stay in the model's context.
+KEEP_LAST_GROUPS = 4
+
+# Module-level markers - applied to all tests in this file
+pytestmark = [
+    pytest.mark.flaky,
+    pytest.mark.integration,
+    pytest.mark.sample("14_conversation_compaction"),
+    pytest.mark.usefixtures("function_app_for_test"),
+]
+
+
+class TestSampleConversationCompaction:
+    """Tests for 14_conversation_compaction sample."""
+
+    @pytest.fixture(autouse=True)
+    def _setup(self, base_url: str, sample_helper) -> None:
+        """Provide agent-specific base URL and helper for the tests."""
+        self.base_url = f"{base_url}/api/agents/Historian"
+        self.helper = sample_helper
+
+    def _run(self, message: str, session_id: str) -> dict:
+        """Send one turn to the agent and return the parsed response.
+
+        Args:
+            message: The user message for this turn.
+            session_id: The session id tying the turns into one conversation.
+
+        Returns:
+            The parsed JSON response body.
+        """
+        response = self.helper.post_json(f"{self.base_url}/run", {"message": message, "session_id": session_id})
+        assert response.status_code in [200, 202]
+        return response.json()
+
+    def test_health_check(self, base_url: str, sample_helper) -> None:
+        """Test health check endpoint."""
+        response = sample_helper.get(f"{base_url}/api/health")
+        assert response.status_code == 200
+        assert response.json()["status"] == "healthy"
+
+    def test_recent_context_survives_compaction(self) -> None:
+        """A fact inside the retained window is still answerable after the window fills."""
+        session_id = f"compaction-recent-{uuid.uuid4().hex[:8]}"
+
+        for index in range(KEEP_LAST_GROUPS):
+            self._run(f"Name animal number {index + 1}.", session_id)
+
+        self._run("My project codename is BLUEHERON.", session_id)
+        answer = self._run("What is my project codename? Reply with just the codename.", session_id)
+
+        assert "blueheron" in str(answer["response"]).lower()
+
+    def test_conversation_continues_across_turns(self) -> None:
+        """Durable history reaches the model, so the agent recalls an earlier turn."""
+        session_id = f"compaction-continuity-{uuid.uuid4().hex[:8]}"
+
+        self._run("My favorite animal is the axolotl.", session_id)
+        answer = self._run("What is my favorite animal? Reply with just the animal name.", session_id)
+
+        assert "axolotl" in str(answer["response"]).lower()