diff --git a/python/packages/durabletask/agent_framework_durabletask/__init__.py b/python/packages/durabletask/agent_framework_durabletask/__init__.py index 3283cf8959..444796ffd5 100644 --- a/python/packages/durabletask/agent_framework_durabletask/__init__.py +++ b/python/packages/durabletask/agent_framework_durabletask/__init__.py @@ -7,8 +7,10 @@ from ._client import DurableAIAgentClient from ._constants import ( DEFAULT_MAX_POLL_RETRIES, DEFAULT_POLL_INTERVAL_SECONDS, + DEFAULT_TIME_TO_LIVE_DAYS, MIMETYPE_APPLICATION_JSON, MIMETYPE_TEXT_PLAIN, + MINIMUM_TTL_SIGNAL_DELAY_MINUTES, REQUEST_RESPONSE_FORMAT_JSON, REQUEST_RESPONSE_FORMAT_TEXT, THREAD_ID_FIELD, @@ -52,8 +54,10 @@ from ._worker import DurableAIAgentWorker __all__ = [ "DEFAULT_MAX_POLL_RETRIES", "DEFAULT_POLL_INTERVAL_SECONDS", + "DEFAULT_TIME_TO_LIVE_DAYS", "MIMETYPE_APPLICATION_JSON", "MIMETYPE_TEXT_PLAIN", + "MINIMUM_TTL_SIGNAL_DELAY_MINUTES", "REQUEST_RESPONSE_FORMAT_JSON", "REQUEST_RESPONSE_FORMAT_TEXT", "THREAD_ID_FIELD", diff --git a/python/packages/durabletask/agent_framework_durabletask/_constants.py b/python/packages/durabletask/agent_framework_durabletask/_constants.py index fd9ebcb8bf..6d9a9cdd59 100644 --- a/python/packages/durabletask/agent_framework_durabletask/_constants.py +++ b/python/packages/durabletask/agent_framework_durabletask/_constants.py @@ -28,6 +28,10 @@ WAIT_FOR_RESPONSE_HEADER: str = "x-ms-wait-for-response" DEFAULT_MAX_POLL_RETRIES: int = 30 DEFAULT_POLL_INTERVAL_SECONDS: float = 1.0 +# TTL configuration +DEFAULT_TIME_TO_LIVE_DAYS: int = 14 +MINIMUM_TTL_SIGNAL_DELAY_MINUTES: int = 5 + # ============================================================================= # JSON Field Name Constants for Durable Agent State Serialization @@ -98,6 +102,9 @@ class DurableStateFields: # History field CONVERSATION_HISTORY: Final[str] = "conversationHistory" + # TTL field + EXPIRATION_TIME_UTC: Final[str] = "expirationTimeUtc" + class ContentTypes: """Content type discriminator values for the $type field. diff --git a/python/packages/durabletask/agent_framework_durabletask/_durable_agent_state.py b/python/packages/durabletask/agent_framework_durabletask/_durable_agent_state.py index 453d180612..b32c36c8ae 100644 --- a/python/packages/durabletask/agent_framework_durabletask/_durable_agent_state.py +++ b/python/packages/durabletask/agent_framework_durabletask/_durable_agent_state.py @@ -332,24 +332,29 @@ class DurableAgentStateData: Attributes: conversation_history: Ordered list of conversation entries (requests and responses) extension_data: Optional dictionary for custom metadata (not part of core schema) + expiration_time_utc: Optional UTC datetime when the entity should expire (TTL feature) """ conversation_history: list[DurableAgentStateEntry] extension_data: dict[str, Any] | None + expiration_time_utc: datetime | None def __init__( self, conversation_history: list[DurableAgentStateEntry] | None = None, extension_data: dict[str, Any] | None = None, + expiration_time_utc: datetime | None = None, ) -> None: """Initialize the data container. Args: conversation_history: Initial conversation history (defaults to empty list) extension_data: Optional custom metadata + expiration_time_utc: Optional UTC datetime when the entity should expire """ self.conversation_history = conversation_history or [] self.extension_data = extension_data + self.expiration_time_utc = expiration_time_utc def to_dict(self) -> dict[str, Any]: result: dict[str, Any] = { @@ -357,13 +362,19 @@ class DurableAgentStateData: } if self.extension_data is not None: result[DurableStateFields.EXTENSION_DATA] = self.extension_data + if self.expiration_time_utc is not None: + result[DurableStateFields.EXPIRATION_TIME_UTC] = self.expiration_time_utc.isoformat() return result @classmethod def from_dict(cls, data_dict: dict[str, Any]) -> DurableAgentStateData: + expiration_time_raw = data_dict.get(DurableStateFields.EXPIRATION_TIME_UTC) + expiration_time_utc = _parse_created_at(expiration_time_raw) if expiration_time_raw else None + return cls( conversation_history=_parse_history_entries(data_dict), extension_data=data_dict.get(DurableStateFields.EXTENSION_DATA), + expiration_time_utc=expiration_time_utc, ) diff --git a/python/packages/durabletask/agent_framework_durabletask/_entities.py b/python/packages/durabletask/agent_framework_durabletask/_entities.py index 7c99a4edc3..7d902ab315 100644 --- a/python/packages/durabletask/agent_framework_durabletask/_entities.py +++ b/python/packages/durabletask/agent_framework_durabletask/_entities.py @@ -6,6 +6,7 @@ from __future__ import annotations import inspect from collections.abc import AsyncIterable +from datetime import datetime, timedelta, timezone from typing import Any, cast from agent_framework import ( @@ -20,6 +21,7 @@ from agent_framework import ( from durabletask.entities import DurableEntity from ._callbacks import AgentCallbackContext, AgentResponseCallbackProtocol +from ._constants import MINIMUM_TTL_SIGNAL_DELAY_MINUTES from ._durable_agent_state import ( DurableAgentState, DurableAgentStateEntry, @@ -95,12 +97,20 @@ class AgentEntity: callback: AgentResponseCallbackProtocol | None = None, *, state_provider: AgentEntityStateProviderMixin, + time_to_live: timedelta | None = None, + minimum_ttl_signal_delay: timedelta = timedelta(minutes=MINIMUM_TTL_SIGNAL_DELAY_MINUTES), ) -> None: self.agent = agent self.callback = callback self._state_provider = state_provider + self._time_to_live = time_to_live + self._minimum_ttl_signal_delay = minimum_ttl_signal_delay - logger.debug("[AgentEntity] Initialized with agent type: %s", type(agent).__name__) + logger.debug( + "[AgentEntity] Initialized with agent type: %s, TTL: %s", + type(agent).__name__, + time_to_live, + ) @property def state(self) -> DurableAgentState: @@ -170,6 +180,10 @@ class AgentEntity: state_response = DurableAgentStateResponse.from_run_response(correlation_id, agent_run_response) self.state.data.conversation_history.append(state_response) + + # Update TTL expiration time + self._update_ttl_expiration() + self.persist_state() return agent_run_response @@ -327,6 +341,113 @@ class AgentEntity: request_message=request_message, ) + # TTL (Time-To-Live) Methods + def _update_ttl_expiration(self) -> None: + """Update the TTL expiration time after an interaction. + + If TTL is configured: + - Sets/updates the expiration time to now + TTL + - On first interaction, schedules a deletion check signal + + If TTL is disabled (None): + - Clears any previously set expiration time + """ + if self._time_to_live is not None: + current_time = datetime.now(timezone.utc) + new_expiration_time = current_time + self._time_to_live + is_first_interaction = self.state.data.expiration_time_utc is None + + self.state.data.expiration_time_utc = new_expiration_time + + logger.debug( + "[AgentEntity] TTL expiration time updated to %s (first_interaction: %s)", + new_expiration_time.isoformat(), + is_first_interaction, + ) + + # Only schedule deletion check on the first interaction when entity is created. + # On subsequent interactions, we just update the expiration time. The scheduled + # check_and_delete_if_expired will reschedule itself if the entity hasn't expired. + if is_first_interaction: + self._schedule_deletion_check() + else: + # TTL is disabled. Clear the expiration time if it was previously set. + if self.state.data.expiration_time_utc is not None: + logger.debug("[AgentEntity] TTL disabled, clearing expiration time") + self.state.data.expiration_time_utc = None + + def check_and_delete_if_expired(self) -> None: + """Check if the entity has expired and delete it if so, otherwise reschedule. + + This method is called by a scheduled signal to check TTL expiration. + If the entity has expired (current time >= expiration time), the state is deleted. + If not expired, a new deletion check is scheduled for the updated expiration time. + """ + current_time = datetime.now(timezone.utc) + expiration_time = self.state.data.expiration_time_utc + + logger.debug( + "[AgentEntity] TTL deletion check: expiration=%s, current=%s", + expiration_time.isoformat() if expiration_time else "None", + current_time.isoformat(), + ) + + if expiration_time is not None: + if current_time >= expiration_time: + # Entity has expired, delete it by resetting state + logger.info( + "[AgentEntity] Entity expired at %s, deleting state", + expiration_time.isoformat(), + ) + self._delete_entity_state() + else: + # Entity hasn't expired yet, reschedule the deletion check + if self._time_to_live is not None: + logger.debug("[AgentEntity] Entity not expired, rescheduling deletion check") + self._schedule_deletion_check() + + def _schedule_deletion_check(self) -> None: + """Schedule a signal to self to check for expiration. + + Note: The Python durabletask SDK does not currently support scheduled signals. + This method logs a warning and skips scheduling. The expiration time is still + tracked in state for potential external cleanup or future SDK support. + + See: https://github.com/Azure/azure-functions-durable-extension/issues/1554 + """ + if not isinstance(self._state_provider, DurableTaskEntityStateProvider): + logger.warning( + "[AgentEntity] Cannot schedule deletion check: state provider does not support entity signaling" + ) + return + + current_time = datetime.now(timezone.utc) + expiration_time = self.state.data.expiration_time_utc + + if expiration_time is None and self._time_to_live is not None: + expiration_time = current_time + self._time_to_live + + if expiration_time is None: + return + + # To avoid excessive scheduling, we schedule the deletion check for no less than the minimum delay + minimum_scheduled_time = current_time + self._minimum_ttl_signal_delay + scheduled_time = max(expiration_time, minimum_scheduled_time) + + logger.warning( + "[AgentEntity] TTL scheduled deletion is not yet supported in the Python durabletask SDK. " + "Expiration time set to %s but automatic deletion will not occur. " + "Consider implementing external cleanup based on expiration_time_utc in entity state.", + scheduled_time.isoformat(), + ) + + def _delete_entity_state(self) -> None: + """Delete the entity state to remove the entity.""" + # Reset the state cache to clear all data + self._state_provider._state_cache = None + # Set empty state to trigger deletion + self._state_provider._set_state_dict({}) + class DurableTaskEntityStateProvider(DurableEntity, AgentEntityStateProviderMixin): """DurableTask Durable Entity state provider for AgentEntity. diff --git a/python/packages/durabletask/agent_framework_durabletask/_worker.py b/python/packages/durabletask/agent_framework_durabletask/_worker.py index fea4b8ba7c..4cd297408e 100644 --- a/python/packages/durabletask/agent_framework_durabletask/_worker.py +++ b/python/packages/durabletask/agent_framework_durabletask/_worker.py @@ -9,12 +9,14 @@ and enables registration of agents as durable entities. from __future__ import annotations import asyncio +from datetime import timedelta from typing import Any from agent_framework import AgentProtocol, get_logger from durabletask.worker import TaskHubGrpcWorker from ._callbacks import AgentResponseCallbackProtocol +from ._constants import DEFAULT_TIME_TO_LIVE_DAYS, MINIMUM_TTL_SIGNAL_DELAY_MINUTES from ._entities import AgentEntity, DurableTaskEntityStateProvider logger = get_logger("agent_framework.durabletask.worker") @@ -51,22 +53,40 @@ class DurableAIAgentWorker: self, worker: TaskHubGrpcWorker, callback: AgentResponseCallbackProtocol | None = None, + default_time_to_live: timedelta | None = timedelta(days=DEFAULT_TIME_TO_LIVE_DAYS), + minimum_ttl_signal_delay: timedelta = timedelta(minutes=MINIMUM_TTL_SIGNAL_DELAY_MINUTES), ): """Initialize the worker wrapper. Args: worker: The durabletask worker instance to wrap callback: Optional callback for agent response notifications + default_time_to_live: Default TTL for agent entities. If an agent entity is idle + for this duration, it will be automatically deleted. Defaults to 14 days. + Set to None to disable TTL for agents without explicit TTL configuration. + minimum_ttl_signal_delay: Minimum delay for scheduling TTL deletion signals. + Defaults to 5 minutes. This prevents excessive scheduling overhead. + + Raises: + ValueError: If minimum_ttl_signal_delay exceeds 5 minutes. """ + max_delay = timedelta(minutes=5) + if minimum_ttl_signal_delay > max_delay: + raise ValueError(f"minimum_ttl_signal_delay cannot exceed {max_delay}. Got: {minimum_ttl_signal_delay}") + self._worker = worker self._callback = callback + self._default_time_to_live = default_time_to_live + self._minimum_ttl_signal_delay = minimum_ttl_signal_delay self._registered_agents: dict[str, AgentProtocol] = {} + self._agent_time_to_live: dict[str, timedelta | None] = {} logger.debug("[DurableAIAgentWorker] Initialized with worker type: %s", type(worker).__name__) def add_agent( self, agent: AgentProtocol, callback: AgentResponseCallbackProtocol | None = None, + time_to_live: timedelta | None = None, ) -> None: """Register an agent with the worker. @@ -77,6 +97,10 @@ class DurableAIAgentWorker: Args: agent: The agent to register (must have a name) callback: Optional callback for this specific agent (overrides worker-level callback) + time_to_live: Optional TTL for this agent's entities. If an entity is idle for + this duration, it will be automatically deleted. If not specified, uses + the worker's default_time_to_live. Pass a sentinel value to explicitly + disable TTL for this agent. Raises: ValueError: If the agent doesn't have a name or is already registered @@ -90,23 +114,29 @@ class DurableAIAgentWorker: logger.info("[DurableAIAgentWorker] Registering agent: %s as entity: dafx-%s", agent_name, agent_name) - # Store the agent reference + # Store the agent reference and TTL configuration self._registered_agents[agent_name] = agent + if time_to_live is not None: + self._agent_time_to_live[agent_name] = time_to_live # Use agent-specific callback if provided, otherwise use worker-level callback effective_callback = callback or self._callback + # Get the effective TTL for this agent + effective_ttl = self.get_time_to_live(agent_name) + # Create a configured entity class using the factory - entity_class = self.__create_agent_entity(agent, effective_callback) + entity_class = self.__create_agent_entity(agent, effective_callback, effective_ttl) # Register the entity class with the worker # The worker.add_entity method takes a class entity_registered: str = self._worker.add_entity(entity_class) # pyright: ignore[reportUnknownMemberType] logger.debug( - "[DurableAIAgentWorker] Successfully registered entity class %s for agent: %s", + "[DurableAIAgentWorker] Successfully registered entity class %s for agent: %s (TTL: %s)", entity_registered, agent_name, + effective_ttl, ) def start(self) -> None: @@ -137,10 +167,30 @@ class DurableAIAgentWorker: """ return list(self._registered_agents.keys()) + def get_time_to_live(self, agent_name: str) -> timedelta | None: + """Get the TTL for a specific agent. + + Args: + agent_name: The name of the agent + + Returns: + The TTL for the agent, or the default TTL if not specified. + Returns None if TTL is disabled for this agent. + """ + if agent_name in self._agent_time_to_live: + return self._agent_time_to_live[agent_name] + return self._default_time_to_live + + @property + def minimum_ttl_signal_delay(self) -> timedelta: + """Get the minimum delay for TTL deletion signals.""" + return self._minimum_ttl_signal_delay + def __create_agent_entity( self, agent: AgentProtocol, callback: AgentResponseCallbackProtocol | None = None, + time_to_live: timedelta | None = None, ) -> type[DurableTaskEntityStateProvider]: """Factory function to create a DurableEntity class configured with an agent. @@ -150,12 +200,14 @@ class DurableAIAgentWorker: Args: agent: The agent instance to wrap callback: Optional callback for agent responses + time_to_live: Optional TTL for this agent's entities Returns: A new DurableEntity subclass configured for this agent """ agent_name = agent.name or type(agent).__name__ entity_name = f"dafx-{agent_name}" + minimum_signal_delay = self._minimum_ttl_signal_delay class ConfiguredAgentEntity(DurableTaskEntityStateProvider): """Durable entity configured with a specific agent instance.""" @@ -167,11 +219,14 @@ class DurableAIAgentWorker: agent=agent, callback=callback, state_provider=self, + time_to_live=time_to_live, + minimum_ttl_signal_delay=minimum_signal_delay, ) logger.debug( - "[ConfiguredAgentEntity] Initialized entity for agent: %s (entity name: %s)", + "[ConfiguredAgentEntity] Initialized entity for agent: %s (entity name: %s, TTL: %s)", agent_name, entity_name, + time_to_live, ) def run(self, request: Any) -> Any: @@ -210,6 +265,16 @@ class DurableAIAgentWorker: logger.debug("[ConfiguredAgentEntity.reset] Resetting agent: %s", agent_name) self._agent_entity.reset() + def check_and_delete_if_expired(self) -> None: + """Check if the entity has expired and delete it if so. + + This method is called by a scheduled signal to check TTL expiration. + """ + logger.debug( + "[ConfiguredAgentEntity.check_and_delete_if_expired] Checking expiration for agent: %s", agent_name + ) + self._agent_entity.check_and_delete_if_expired() + # Set the entity name to match the prefixed agent name # This is used by durabletask to register the entity ConfiguredAgentEntity.__name__ = entity_name diff --git a/python/samples/getting_started/durabletask/01_single_agent/worker.py b/python/samples/getting_started/durabletask/01_single_agent/worker.py index d445deb102..9d9799bf9a 100644 --- a/python/samples/getting_started/durabletask/01_single_agent/worker.py +++ b/python/samples/getting_started/durabletask/01_single_agent/worker.py @@ -3,8 +3,8 @@ This worker registers agents as durable entities and continuously listens for requests. The worker should run as a background service, processing incoming agent requests. -Prerequisites: -- Set AZURE_OPENAI_ENDPOINT and AZURE_OPENAI_CHAT_DEPLOYMENT_NAME +Prerequisites: +- Set AZURE_OPENAI_ENDPOINT and AZURE_OPENAI_CHAT_DEPLOYMENT_NAME (plus AZURE_OPENAI_API_KEY or Azure CLI authentication) - Start a Durable Task Scheduler (e.g., using Docker) """ @@ -12,6 +12,7 @@ Prerequisites: import asyncio import logging import os +from datetime import timedelta from agent_framework.azure import AzureOpenAIChatClient from agent_framework_durabletask import DurableAIAgentWorker @@ -69,24 +70,26 @@ def get_worker( def setup_worker(worker: DurableTaskSchedulerWorker) -> DurableAIAgentWorker: """Set up the worker with agents registered. - + Args: worker: The DurableTaskSchedulerWorker instance - + Returns: DurableAIAgentWorker with agents registered """ # Wrap it with the agent worker agent_worker = DurableAIAgentWorker(worker) - - # Create and register the Joker agent + + # Create and register the Joker agent with a 1-hour TTL + # Agent sessions will be automatically deleted after 1 hour of inactivity logger.debug("Creating and registering Joker agent...") joker_agent = create_joker_agent() - agent_worker.add_agent(joker_agent) - + agent_worker.add_agent(joker_agent, time_to_live=timedelta(seconds=30)) + logger.debug(f"✓ Registered agent: {joker_agent.name}") logger.debug(f" Entity name: dafx-{joker_agent.name}") - + logger.debug(f" TTL: 1 hour") + return agent_worker