From bf2a3c341859d6c4bd1a3d254e4dfba246dede11 Mon Sep 17 00:00:00 2001 From: Casper Nielsen Date: Thu, 18 Dec 2025 10:30:37 +0100 Subject: [PATCH 01/11] chore: remove unused import Signed-off-by: Casper Nielsen --- dapr_agents/agents/durable.py | 1 - 1 file changed, 1 deletion(-) diff --git a/dapr_agents/agents/durable.py b/dapr_agents/agents/durable.py index 617ba1bb..7adc70a1 100644 --- a/dapr_agents/agents/durable.py +++ b/dapr_agents/agents/durable.py @@ -26,7 +26,6 @@ from dapr_agents.types import ( AgentError, LLMChatResponse, - ToolExecutionRecord, ToolMessage, UserMessage, ) From 17f478d553185ea63f64bfd05262fa9e7f643ea5 Mon Sep 17 00:00:00 2001 From: Casper Nielsen Date: Thu, 18 Dec 2025 10:32:12 +0100 Subject: [PATCH 02/11] feat: create retrypolicy and pass to wf activities as default retry policy Signed-off-by: Casper Nielsen --- dapr_agents/agents/durable.py | 32 ++++++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/dapr_agents/agents/durable.py b/dapr_agents/agents/durable.py index 7adc70a1..29694d90 100644 --- a/dapr_agents/agents/durable.py +++ b/dapr_agents/agents/durable.py @@ -1,8 +1,10 @@ from __future__ import annotations +from datetime import timedelta import json import logging from typing import Any, Dict, Iterable, List, Optional +from os import getenv import dapr.ext.workflow as wf @@ -75,6 +77,10 @@ def __init__( agent_metadata: Optional[Dict[str, Any]] = None, workflow_grpc: Optional[WorkflowGrpcOptions] = None, runtime: Optional[wf.WorkflowRuntime] = None, + max_attempts: int = 1, + initial_backoff: int = 15, + max_backoff: int = 30, + backoff_multiplier: float = 1.5, ) -> None: """ Initialize behavior, infrastructure, and workflow runtime. @@ -103,6 +109,10 @@ def __init__( agent_metadata: Extra metadata to publish to the registry. workflow_grpc: Optional gRPC overrides for the workflow runtime channel. runtime: Optional pre-existing workflow runtime to attach to. + max_attempts: Maximum number of retry attempts for workflow operations. Default is 1 (no retries). Set DAPR_API_MAX_RETRIES env variable to override default. + initial_backoff: Initial backoff duration in seconds. Default is 15 seconds. + max_backoff: Maximum backoff duration in seconds. Default is 30 seconds. + backoff_multiplier: Backoff multiplier for exponential backoff. Default is 1.5. """ super().__init__( pubsub=pubsub, @@ -131,6 +141,21 @@ def __init__( self._registered = False self._started = False + try: + retries = int(getenv("DAPR_API_MAX_RETRIES", max_attempts)) + except ValueError: + retries = max_attempts + + if retries < 1: + raise (ValueError("max_attempts or DAPR_API_MAX_RETRIES must be at least 1.")) + + self._retry_policy: wf.RetryPolicy = wf.RetryPolicy( + max_number_of_attempts=retries, + first_retry_interval=timedelta(seconds=initial_backoff), + max_retry_interval=timedelta(seconds=max_backoff), + backoff_coefficient=backoff_multiplier, + ) + # ------------------------------------------------------------------ # Runtime accessors # ------------------------------------------------------------------ @@ -202,6 +227,7 @@ def agent_workflow(self, ctx: wf.DaprWorkflowContext, message: dict): "start_time": ctx.current_utc_datetime.isoformat(), "trace_context": otel_span_context, }, + retry_policy=self._retry_policy, ) final_message: Dict[str, Any] = {} @@ -225,6 +251,7 @@ def agent_workflow(self, ctx: wf.DaprWorkflowContext, message: dict): "instance_id": ctx.instance_id, "time": ctx.current_utc_datetime.isoformat(), }, + retry_policy=self._retry_policy, ) tool_calls = assistant_response.get("tool_calls") or [] @@ -245,6 +272,7 @@ def agent_workflow(self, ctx: wf.DaprWorkflowContext, message: dict): "time": ctx.current_utc_datetime.isoformat(), "order": idx, }, + retry_policy=self._retry_policy, ) for idx, tc in enumerate(tool_calls) ] @@ -256,6 +284,7 @@ def agent_workflow(self, ctx: wf.DaprWorkflowContext, message: dict): "tool_results": tool_results, "instance_id": ctx.instance_id, }, + retry_policy=self._retry_policy, ) task = None # prepare for next turn @@ -297,6 +326,7 @@ def agent_workflow(self, ctx: wf.DaprWorkflowContext, message: dict): yield ctx.call_activity( self.broadcast_message_to_agents, input={"message": final_message}, + retry_policy=self._retry_policy, ) # Optionally send a direct response back to the trigger origin. @@ -308,6 +338,7 @@ def agent_workflow(self, ctx: wf.DaprWorkflowContext, message: dict): "target_agent": source, "target_instance_id": trigger_instance_id, }, + retry_policy=self._retry_policy, ) # Finalize the workflow entry in durable state. @@ -319,6 +350,7 @@ def agent_workflow(self, ctx: wf.DaprWorkflowContext, message: dict): "end_time": ctx.current_utc_datetime.isoformat(), "triggering_workflow_instance_id": trigger_instance_id, }, + retry_policy=self._retry_policy, ) if not ctx.is_replaying: From f3fc5fcf25c60912d298aff88dab08a2b52f454e Mon Sep 17 00:00:00 2001 From: Casper Nielsen Date: Thu, 18 Dec 2025 10:33:19 +0100 Subject: [PATCH 03/11] feat: add mock retry policy on mocked wf runtime Signed-off-by: Casper Nielsen --- tests/agents/durableagent/test_durable_agent.py | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/tests/agents/durableagent/test_durable_agent.py b/tests/agents/durableagent/test_durable_agent.py index 21832d23..347134e0 100644 --- a/tests/agents/durableagent/test_durable_agent.py +++ b/tests/agents/durableagent/test_durable_agent.py @@ -43,6 +43,21 @@ def patch_dapr_check(monkeypatch): mock_runtime = Mock(spec=wf.WorkflowRuntime) monkeypatch.setattr(wf, "WorkflowRuntime", lambda: mock_runtime) + class MockRetryPolicy: + def __init__( + self, + max_number_of_attempts=1, + first_retry_interval=timedelta(seconds=1), + max_retry_interval=timedelta(seconds=60), + backoff_coefficient=2.0, + ): + self.max_number_of_attempts = max_number_of_attempts + self.first_retry_interval = first_retry_interval + self.max_retry_interval = max_retry_interval + self.backoff_coefficient = backoff_coefficient + + monkeypatch.setattr(wf, "RetryPolicy", MockRetryPolicy) + # Return the mock runtime for tests that need it yield mock_runtime From 5cbd400d845d5b9d1c74ce8a951c4a09da73a04f Mon Sep 17 00:00:00 2001 From: Casper Nielsen Date: Thu, 18 Dec 2025 10:34:05 +0100 Subject: [PATCH 04/11] test: add test cases for verifying input checks, env var overwrite and each activity gets applied retry policy Signed-off-by: Casper Nielsen --- .../agents/durableagent/test_durable_agent.py | 185 ++++++++++++++++++ 1 file changed, 185 insertions(+) diff --git a/tests/agents/durableagent/test_durable_agent.py b/tests/agents/durableagent/test_durable_agent.py index 347134e0..ef7e3707 100644 --- a/tests/agents/durableagent/test_durable_agent.py +++ b/tests/agents/durableagent/test_durable_agent.py @@ -2,6 +2,7 @@ # Right now we have to do a bunch of patching at the class-level instead of patching at the instance-level. # In future, we should do dependency injection instead of patching at the class-level to make it easier to test. # This applies to all areas in this file where we have with patch.object()... +from datetime import timedelta import os from unittest.mock import AsyncMock, Mock, patch, MagicMock @@ -1005,3 +1006,187 @@ def test_durable_agent_state_initialization(self, basic_durable_agent): assert isinstance(validated_state, AgentWorkflowState) assert "instances" in basic_durable_agent.state assert basic_durable_agent.state["instances"] == {} + + def test_durable_agent_retry_policy_initialization(self, mock_llm): + """Test that DurableAgent correctly initializes with retry policy parameters.""" + agent = DurableAgent( + name="RetryTestAgent", + role="Retry Test Assistant", + llm=mock_llm, + pubsub=AgentPubSubConfig( + pubsub_name="testpubsub", + agent_topic="RetryTestAgent", + ), + max_attempts=5, + initial_backoff=10, + max_backoff=60, + backoff_multiplier=2.0, + ) + + assert agent._retry_policy is not None + assert agent._retry_policy.max_number_of_attempts == 5 + assert agent._retry_policy.first_retry_interval.total_seconds() == 10 + assert agent._retry_policy.max_retry_interval.total_seconds() == 60 + assert agent._retry_policy.backoff_coefficient == 2.0 + + def test_durable_agent_retry_policy_defaults(self, mock_llm): + """Test that DurableAgent uses correct default retry values.""" + agent = DurableAgent( + name="RetryDefaultAgent", + role="Retry Default Assistant", + llm=mock_llm, + pubsub=AgentPubSubConfig( + pubsub_name="testpubsub", + agent_topic="RetryDefaultAgent", + ), + ) + + assert agent._retry_policy is not None + assert agent._retry_policy.max_number_of_attempts == 1 + assert agent._retry_policy.first_retry_interval.total_seconds() == 15 + assert agent._retry_policy.max_retry_interval.total_seconds() == 30 + assert agent._retry_policy.backoff_coefficient == 1.5 + + def test_durable_agent_retry_policy_env_override(self, mock_llm, monkeypatch): + """Test that DAPR_API_MAX_RETRIES environment variable overrides max_attempts.""" + monkeypatch.setenv("DAPR_API_MAX_RETRIES", "10") + + agent = DurableAgent( + name="RetryEnvAgent", + role="Retry Env Assistant", + llm=mock_llm, + pubsub=AgentPubSubConfig( + pubsub_name="testpubsub", + agent_topic="RetryEnvAgent", + ), + max_attempts=3, + ) + + # Should use env var value over max_attempts + assert agent._retry_policy.max_number_of_attempts == 10 + + def test_durable_agent_retry_policy_invalid_env(self, mock_llm, monkeypatch): + """Test that invalid DAPR_API_MAX_RETRIES falls back to max_attempts.""" + monkeypatch.setenv("DAPR_API_MAX_RETRIES", "invalid") + + agent = DurableAgent( + name="RetryInvalidEnvAgent", + role="Retry Invalid Env Assistant", + llm=mock_llm, + pubsub=AgentPubSubConfig( + pubsub_name="testpubsub", + agent_topic="RetryInvalidEnvAgent", + ), + max_attempts=3, + ) + + # Should fall back to max_attempts since env var is invalid + assert agent._retry_policy.max_number_of_attempts == 3 + + def test_durable_agent_retry_policy_min_attempts_validation(self, mock_llm): + """Test that max_attempts cannot be less than 1.""" + with pytest.raises(ValueError, match="max_attempts or DAPR_API_MAX_RETRIES must be at least 1."): + DurableAgent( + name="RetryZeroAgent", + role="Retry Zero Assistant", + llm=mock_llm, + pubsub=AgentPubSubConfig( + pubsub_name="testpubsub", + agent_topic="RetryZeroAgent", + ), + max_attempts=0, + ) + + def test_agent_workflow_applies_retry_policy( + self, basic_durable_agent, mock_workflow_context + ): + """Test that agent_workflow applies retry policy to activity calls.""" + message = { + "task": "Test task with retries", + "workflow_instance_id": "parent-instance-123", + } + + call_activity_calls = [] + + def track_call_activity(activity, **kwargs): + call_activity_calls.append({ + "activity": activity, + "input": kwargs.get("input"), + "retry_policy": kwargs.get("retry_policy"), + }) + + if hasattr(activity, "__name__"): + activity_name = activity.__name__ + elif hasattr(activity, "__func__"): + activity_name = activity.__func__.__name__ + else: + activity_name = str(activity) + + if activity_name == "call_llm": + return { + "content": "Test response", + "tool_calls": [ + { + "id": "call_test_123", + "type": "function", + "function": { + "name": "test_tool", + "arguments": '{"arg": "value"}' + } + } + ], + "role": "assistant" + } + elif activity_name == "run_tool": + return {"tool_call_id": "call_test_123", "content": "tool result", "role": "tool", "name": "test_tool"} + elif activity_name in ["record_initial_entry", "finalize_workflow", "save_tool_results"]: + return None + + mock_workflow_context.instance_id = "test-instance-123" + mock_workflow_context.call_activity = Mock(side_effect=track_call_activity) + + # Set up minimal state + entry = AgentWorkflowEntry( + input_value="Test task with retries", + source=None, + triggering_workflow_instance_id="parent-instance-123", + workflow_instance_id="test-instance-123", + workflow_name="AgenticWorkflow", + status="RUNNING", + messages=[], + tool_history=[], + ) + basic_durable_agent._state_model.instances["test-instance-123"] = entry + + # Run the workflow generator + workflow_gen = basic_durable_agent.agent_workflow( + mock_workflow_context, message + ) + + # Step through the generator, sending results back + result = None + try: + while True: + result = workflow_gen.send(result) + except StopIteration as e: + result = e.value + + # Verify that retry_policy was passed to critical activities + assert len(call_activity_calls) >= 5, f"Expected at least 3 activity calls, got {len(call_activity_calls)}" + + # All activities should have retry_policy parameter + for call in call_activity_calls: + assert "retry_policy" in call, f"Missing retry_policy in call: {call}" + assert call["retry_policy"] == basic_durable_agent._retry_policy, \ + f"Expected retry_policy {basic_durable_agent._retry_policy}, got {call['retry_policy']}" + + # Verify the key activities were called + activity_names = [ + getattr(call["activity"], "__name__", str(call["activity"])) + for call in call_activity_calls + ] + assert "record_initial_entry" in activity_names, f"Missing record_initial_entry in {activity_names}" + assert "call_llm" in activity_names, f"Missing call_llm in {activity_names}" + assert "run_tool" in activity_names, f"Missing run_tool in {activity_names}" + assert "save_tool_results" in activity_names, f"Missing save_tool_results in {activity_names}" + assert "finalize_workflow" in activity_names, f"Missing finalize_workflow in {activity_names}" From 26fa686442045542d39abd3e2fda0871cb172ebd Mon Sep 17 00:00:00 2001 From: Casper Nielsen Date: Thu, 18 Dec 2025 11:14:20 +0100 Subject: [PATCH 05/11] chore: formatting Signed-off-by: Casper Nielsen --- dapr_agents/agents/durable.py | 4 +- .../agents/durableagent/test_durable_agent.py | 68 ++++++++++++------- 2 files changed, 48 insertions(+), 24 deletions(-) diff --git a/dapr_agents/agents/durable.py b/dapr_agents/agents/durable.py index 29694d90..3bbb51a7 100644 --- a/dapr_agents/agents/durable.py +++ b/dapr_agents/agents/durable.py @@ -147,7 +147,9 @@ def __init__( retries = max_attempts if retries < 1: - raise (ValueError("max_attempts or DAPR_API_MAX_RETRIES must be at least 1.")) + raise ( + ValueError("max_attempts or DAPR_API_MAX_RETRIES must be at least 1.") + ) self._retry_policy: wf.RetryPolicy = wf.RetryPolicy( max_number_of_attempts=retries, diff --git a/tests/agents/durableagent/test_durable_agent.py b/tests/agents/durableagent/test_durable_agent.py index ef7e3707..8faa9a66 100644 --- a/tests/agents/durableagent/test_durable_agent.py +++ b/tests/agents/durableagent/test_durable_agent.py @@ -1085,7 +1085,9 @@ def test_durable_agent_retry_policy_invalid_env(self, mock_llm, monkeypatch): def test_durable_agent_retry_policy_min_attempts_validation(self, mock_llm): """Test that max_attempts cannot be less than 1.""" - with pytest.raises(ValueError, match="max_attempts or DAPR_API_MAX_RETRIES must be at least 1."): + with pytest.raises( + ValueError, match="max_attempts or DAPR_API_MAX_RETRIES must be at least 1." + ): DurableAgent( name="RetryZeroAgent", role="Retry Zero Assistant", @@ -1107,21 +1109,23 @@ def test_agent_workflow_applies_retry_policy( } call_activity_calls = [] - + def track_call_activity(activity, **kwargs): - call_activity_calls.append({ - "activity": activity, - "input": kwargs.get("input"), - "retry_policy": kwargs.get("retry_policy"), - }) - + call_activity_calls.append( + { + "activity": activity, + "input": kwargs.get("input"), + "retry_policy": kwargs.get("retry_policy"), + } + ) + if hasattr(activity, "__name__"): activity_name = activity.__name__ elif hasattr(activity, "__func__"): activity_name = activity.__func__.__name__ else: activity_name = str(activity) - + if activity_name == "call_llm": return { "content": "Test response", @@ -1131,15 +1135,24 @@ def track_call_activity(activity, **kwargs): "type": "function", "function": { "name": "test_tool", - "arguments": '{"arg": "value"}' - } + "arguments": '{"arg": "value"}', + }, } ], - "role": "assistant" + "role": "assistant", } elif activity_name == "run_tool": - return {"tool_call_id": "call_test_123", "content": "tool result", "role": "tool", "name": "test_tool"} - elif activity_name in ["record_initial_entry", "finalize_workflow", "save_tool_results"]: + return { + "tool_call_id": "call_test_123", + "content": "tool result", + "role": "tool", + "name": "test_tool", + } + elif activity_name in [ + "record_initial_entry", + "finalize_workflow", + "save_tool_results", + ]: return None mock_workflow_context.instance_id = "test-instance-123" @@ -1162,7 +1175,7 @@ def track_call_activity(activity, **kwargs): workflow_gen = basic_durable_agent.agent_workflow( mock_workflow_context, message ) - + # Step through the generator, sending results back result = None try: @@ -1172,21 +1185,30 @@ def track_call_activity(activity, **kwargs): result = e.value # Verify that retry_policy was passed to critical activities - assert len(call_activity_calls) >= 5, f"Expected at least 3 activity calls, got {len(call_activity_calls)}" + assert ( + len(call_activity_calls) >= 5 + ), f"Expected at least 3 activity calls, got {len(call_activity_calls)}" # All activities should have retry_policy parameter for call in call_activity_calls: assert "retry_policy" in call, f"Missing retry_policy in call: {call}" - assert call["retry_policy"] == basic_durable_agent._retry_policy, \ - f"Expected retry_policy {basic_durable_agent._retry_policy}, got {call['retry_policy']}" - + assert ( + call["retry_policy"] == basic_durable_agent._retry_policy + ), f"Expected retry_policy {basic_durable_agent._retry_policy}, got {call['retry_policy']}" + # Verify the key activities were called activity_names = [ - getattr(call["activity"], "__name__", str(call["activity"])) + getattr(call["activity"], "__name__", str(call["activity"])) for call in call_activity_calls ] - assert "record_initial_entry" in activity_names, f"Missing record_initial_entry in {activity_names}" + assert ( + "record_initial_entry" in activity_names + ), f"Missing record_initial_entry in {activity_names}" assert "call_llm" in activity_names, f"Missing call_llm in {activity_names}" assert "run_tool" in activity_names, f"Missing run_tool in {activity_names}" - assert "save_tool_results" in activity_names, f"Missing save_tool_results in {activity_names}" - assert "finalize_workflow" in activity_names, f"Missing finalize_workflow in {activity_names}" + assert ( + "save_tool_results" in activity_names + ), f"Missing save_tool_results in {activity_names}" + assert ( + "finalize_workflow" in activity_names + ), f"Missing finalize_workflow in {activity_names}" From ee6c8c746c7019e1f0880d2ebfa5e008a7ca6abf Mon Sep 17 00:00:00 2001 From: Casper Nielsen Date: Thu, 18 Dec 2025 15:02:55 +0100 Subject: [PATCH 06/11] fix: let the retry config be a dataclass for grouping Signed-off-by: Casper Nielsen --- dapr_agents/agents/configs.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/dapr_agents/agents/configs.py b/dapr_agents/agents/configs.py index 348be554..b9e782ee 100644 --- a/dapr_agents/agents/configs.py +++ b/dapr_agents/agents/configs.py @@ -265,3 +265,21 @@ class AgentExecutionConfig: # TODO: add stop_at_tokens max_iterations: int = 10 tool_choice: Optional[str] = "auto" + + +@dataclass +class DurableRetryConfig: + """ + Configuration for durable retry policies in workflows. + + Attributes: + max_attempts: Maximum number of retry attempts. + initial_backoff_seconds: Initial backoff interval in seconds. + max_backoff_seconds: Maximum backoff interval in seconds. + backoff_multiplier: Multiplier for exponential backoff. + """ + + max_attempts: Optional[int] = 1 + initial_backoff_seconds: Optional[int] = 5 + max_backoff_seconds: Optional[int] = 30 + backoff_multiplier: Optional[float] = 1.5 From 647fa760c73a43d036fabf06a4ec8605457a2966 Mon Sep 17 00:00:00 2001 From: Casper Nielsen Date: Thu, 18 Dec 2025 15:03:29 +0100 Subject: [PATCH 07/11] fix: consume dataclass in durable agent and tests Signed-off-by: Casper Nielsen --- dapr_agents/agents/durable.py | 23 ++++++++----------- .../agents/durableagent/test_durable_agent.py | 19 ++++++++------- 2 files changed, 21 insertions(+), 21 deletions(-) diff --git a/dapr_agents/agents/durable.py b/dapr_agents/agents/durable.py index 3bbb51a7..750e8445 100644 --- a/dapr_agents/agents/durable.py +++ b/dapr_agents/agents/durable.py @@ -16,6 +16,7 @@ AgentRegistryConfig, AgentStateConfig, WorkflowGrpcOptions, + DurableRetryConfig, ) from dapr_agents.agents.prompting import AgentProfileConfig from dapr_agents.agents.schemas import ( @@ -77,10 +78,7 @@ def __init__( agent_metadata: Optional[Dict[str, Any]] = None, workflow_grpc: Optional[WorkflowGrpcOptions] = None, runtime: Optional[wf.WorkflowRuntime] = None, - max_attempts: int = 1, - initial_backoff: int = 15, - max_backoff: int = 30, - backoff_multiplier: float = 1.5, + retry_policy: Optional[DurableRetryConfig] = DurableRetryConfig(), ) -> None: """ Initialize behavior, infrastructure, and workflow runtime. @@ -109,10 +107,7 @@ def __init__( agent_metadata: Extra metadata to publish to the registry. workflow_grpc: Optional gRPC overrides for the workflow runtime channel. runtime: Optional pre-existing workflow runtime to attach to. - max_attempts: Maximum number of retry attempts for workflow operations. Default is 1 (no retries). Set DAPR_API_MAX_RETRIES env variable to override default. - initial_backoff: Initial backoff duration in seconds. Default is 15 seconds. - max_backoff: Maximum backoff duration in seconds. Default is 30 seconds. - backoff_multiplier: Backoff multiplier for exponential backoff. Default is 1.5. + retry_policy: Durable retry policy configuration. """ super().__init__( pubsub=pubsub, @@ -142,9 +137,9 @@ def __init__( self._started = False try: - retries = int(getenv("DAPR_API_MAX_RETRIES", max_attempts)) + retries = int(getenv("DAPR_API_MAX_RETRIES", "")) except ValueError: - retries = max_attempts + retries = retry_policy.max_attempts if retries < 1: raise ( @@ -153,9 +148,11 @@ def __init__( self._retry_policy: wf.RetryPolicy = wf.RetryPolicy( max_number_of_attempts=retries, - first_retry_interval=timedelta(seconds=initial_backoff), - max_retry_interval=timedelta(seconds=max_backoff), - backoff_coefficient=backoff_multiplier, + first_retry_interval=timedelta( + seconds=retry_policy.initial_backoff_seconds + ), + max_retry_interval=timedelta(seconds=retry_policy.max_backoff_seconds), + backoff_coefficient=retry_policy.backoff_multiplier, ) # ------------------------------------------------------------------ diff --git a/tests/agents/durableagent/test_durable_agent.py b/tests/agents/durableagent/test_durable_agent.py index 8faa9a66..d05812df 100644 --- a/tests/agents/durableagent/test_durable_agent.py +++ b/tests/agents/durableagent/test_durable_agent.py @@ -16,6 +16,7 @@ AgentRegistryConfig, AgentMemoryConfig, AgentExecutionConfig, + DurableRetryConfig, ) from dapr_agents.agents.schemas import ( AgentWorkflowMessage, @@ -1017,10 +1018,12 @@ def test_durable_agent_retry_policy_initialization(self, mock_llm): pubsub_name="testpubsub", agent_topic="RetryTestAgent", ), - max_attempts=5, - initial_backoff=10, - max_backoff=60, - backoff_multiplier=2.0, + retry_policy=DurableRetryConfig( + max_attempts=5, + initial_backoff_seconds=10, + max_backoff_seconds=60, + backoff_multiplier=2.0, + ), ) assert agent._retry_policy is not None @@ -1043,7 +1046,7 @@ def test_durable_agent_retry_policy_defaults(self, mock_llm): assert agent._retry_policy is not None assert agent._retry_policy.max_number_of_attempts == 1 - assert agent._retry_policy.first_retry_interval.total_seconds() == 15 + assert agent._retry_policy.first_retry_interval.total_seconds() == 5 assert agent._retry_policy.max_retry_interval.total_seconds() == 30 assert agent._retry_policy.backoff_coefficient == 1.5 @@ -1059,7 +1062,7 @@ def test_durable_agent_retry_policy_env_override(self, mock_llm, monkeypatch): pubsub_name="testpubsub", agent_topic="RetryEnvAgent", ), - max_attempts=3, + retry_policy=DurableRetryConfig(max_attempts=3), ) # Should use env var value over max_attempts @@ -1077,7 +1080,7 @@ def test_durable_agent_retry_policy_invalid_env(self, mock_llm, monkeypatch): pubsub_name="testpubsub", agent_topic="RetryInvalidEnvAgent", ), - max_attempts=3, + retry_policy=DurableRetryConfig(max_attempts=3), ) # Should fall back to max_attempts since env var is invalid @@ -1096,7 +1099,7 @@ def test_durable_agent_retry_policy_min_attempts_validation(self, mock_llm): pubsub_name="testpubsub", agent_topic="RetryZeroAgent", ), - max_attempts=0, + retry_policy=DurableRetryConfig(max_attempts=0), ) def test_agent_workflow_applies_retry_policy( From c6c029ca86482adf956a8f93b4c4266d3f65a93d Mon Sep 17 00:00:00 2001 From: Casper Nielsen Date: Thu, 18 Dec 2025 15:47:32 +0100 Subject: [PATCH 08/11] fix: rename to WorkflowRetryPolicy Signed-off-by: Casper Nielsen --- dapr_agents/agents/configs.py | 2 +- dapr_agents/agents/durable.py | 4 ++-- tests/agents/durableagent/test_durable_agent.py | 10 +++++----- 3 files changed, 8 insertions(+), 8 deletions(-) diff --git a/dapr_agents/agents/configs.py b/dapr_agents/agents/configs.py index b9e782ee..f60332c8 100644 --- a/dapr_agents/agents/configs.py +++ b/dapr_agents/agents/configs.py @@ -268,7 +268,7 @@ class AgentExecutionConfig: @dataclass -class DurableRetryConfig: +class WorkflowRetryPolicy: """ Configuration for durable retry policies in workflows. diff --git a/dapr_agents/agents/durable.py b/dapr_agents/agents/durable.py index 750e8445..93e55227 100644 --- a/dapr_agents/agents/durable.py +++ b/dapr_agents/agents/durable.py @@ -16,7 +16,7 @@ AgentRegistryConfig, AgentStateConfig, WorkflowGrpcOptions, - DurableRetryConfig, + WorkflowRetryPolicy, ) from dapr_agents.agents.prompting import AgentProfileConfig from dapr_agents.agents.schemas import ( @@ -78,7 +78,7 @@ def __init__( agent_metadata: Optional[Dict[str, Any]] = None, workflow_grpc: Optional[WorkflowGrpcOptions] = None, runtime: Optional[wf.WorkflowRuntime] = None, - retry_policy: Optional[DurableRetryConfig] = DurableRetryConfig(), + retry_policy: WorkflowRetryPolicy = WorkflowRetryPolicy(), ) -> None: """ Initialize behavior, infrastructure, and workflow runtime. diff --git a/tests/agents/durableagent/test_durable_agent.py b/tests/agents/durableagent/test_durable_agent.py index d05812df..ee808b56 100644 --- a/tests/agents/durableagent/test_durable_agent.py +++ b/tests/agents/durableagent/test_durable_agent.py @@ -16,7 +16,7 @@ AgentRegistryConfig, AgentMemoryConfig, AgentExecutionConfig, - DurableRetryConfig, + WorkflowRetryPolicy, ) from dapr_agents.agents.schemas import ( AgentWorkflowMessage, @@ -1018,7 +1018,7 @@ def test_durable_agent_retry_policy_initialization(self, mock_llm): pubsub_name="testpubsub", agent_topic="RetryTestAgent", ), - retry_policy=DurableRetryConfig( + retry_policy=WorkflowRetryPolicy( max_attempts=5, initial_backoff_seconds=10, max_backoff_seconds=60, @@ -1062,7 +1062,7 @@ def test_durable_agent_retry_policy_env_override(self, mock_llm, monkeypatch): pubsub_name="testpubsub", agent_topic="RetryEnvAgent", ), - retry_policy=DurableRetryConfig(max_attempts=3), + retry_policy=WorkflowRetryPolicy(max_attempts=3), ) # Should use env var value over max_attempts @@ -1080,7 +1080,7 @@ def test_durable_agent_retry_policy_invalid_env(self, mock_llm, monkeypatch): pubsub_name="testpubsub", agent_topic="RetryInvalidEnvAgent", ), - retry_policy=DurableRetryConfig(max_attempts=3), + retry_policy=WorkflowRetryPolicy(max_attempts=3), ) # Should fall back to max_attempts since env var is invalid @@ -1099,7 +1099,7 @@ def test_durable_agent_retry_policy_min_attempts_validation(self, mock_llm): pubsub_name="testpubsub", agent_topic="RetryZeroAgent", ), - retry_policy=DurableRetryConfig(max_attempts=0), + retry_policy=WorkflowRetryPolicy(max_attempts=0), ) def test_agent_workflow_applies_retry_policy( From dd516e64107d6682dedbe7b23814f145fefc81a3 Mon Sep 17 00:00:00 2001 From: Casper Nielsen Date: Thu, 18 Dec 2025 15:48:32 +0100 Subject: [PATCH 09/11] fix: ensure adding retry_timeout Signed-off-by: Casper Nielsen --- dapr_agents/agents/configs.py | 1 + dapr_agents/agents/durable.py | 3 +++ tests/agents/durableagent/test_durable_agent.py | 6 ++++++ 3 files changed, 10 insertions(+) diff --git a/dapr_agents/agents/configs.py b/dapr_agents/agents/configs.py index f60332c8..8886501f 100644 --- a/dapr_agents/agents/configs.py +++ b/dapr_agents/agents/configs.py @@ -283,3 +283,4 @@ class WorkflowRetryPolicy: initial_backoff_seconds: Optional[int] = 5 max_backoff_seconds: Optional[int] = 30 backoff_multiplier: Optional[float] = 1.5 + retry_timeout: Optional[Union[int, None]] = None diff --git a/dapr_agents/agents/durable.py b/dapr_agents/agents/durable.py index 93e55227..3dfc11d2 100644 --- a/dapr_agents/agents/durable.py +++ b/dapr_agents/agents/durable.py @@ -153,6 +153,9 @@ def __init__( ), max_retry_interval=timedelta(seconds=retry_policy.max_backoff_seconds), backoff_coefficient=retry_policy.backoff_multiplier, + retry_timeout=timedelta(seconds=retry_policy.retry_timeout) + if retry_policy.retry_timeout + else None, ) # ------------------------------------------------------------------ diff --git a/tests/agents/durableagent/test_durable_agent.py b/tests/agents/durableagent/test_durable_agent.py index ee808b56..e735d3b5 100644 --- a/tests/agents/durableagent/test_durable_agent.py +++ b/tests/agents/durableagent/test_durable_agent.py @@ -4,6 +4,7 @@ # This applies to all areas in this file where we have with patch.object()... from datetime import timedelta import os +from typing import Optional from unittest.mock import AsyncMock, Mock, patch, MagicMock import pytest @@ -52,11 +53,13 @@ def __init__( first_retry_interval=timedelta(seconds=1), max_retry_interval=timedelta(seconds=60), backoff_coefficient=2.0, + retry_timeout: Optional[timedelta] = None, ): self.max_number_of_attempts = max_number_of_attempts self.first_retry_interval = first_retry_interval self.max_retry_interval = max_retry_interval self.backoff_coefficient = backoff_coefficient + self.retry_timeout = retry_timeout monkeypatch.setattr(wf, "RetryPolicy", MockRetryPolicy) @@ -1023,6 +1026,7 @@ def test_durable_agent_retry_policy_initialization(self, mock_llm): initial_backoff_seconds=10, max_backoff_seconds=60, backoff_multiplier=2.0, + retry_timeout=300, ), ) @@ -1031,6 +1035,7 @@ def test_durable_agent_retry_policy_initialization(self, mock_llm): assert agent._retry_policy.first_retry_interval.total_seconds() == 10 assert agent._retry_policy.max_retry_interval.total_seconds() == 60 assert agent._retry_policy.backoff_coefficient == 2.0 + assert agent._retry_policy.retry_timeout.total_seconds() == 300 def test_durable_agent_retry_policy_defaults(self, mock_llm): """Test that DurableAgent uses correct default retry values.""" @@ -1049,6 +1054,7 @@ def test_durable_agent_retry_policy_defaults(self, mock_llm): assert agent._retry_policy.first_retry_interval.total_seconds() == 5 assert agent._retry_policy.max_retry_interval.total_seconds() == 30 assert agent._retry_policy.backoff_coefficient == 1.5 + assert agent._retry_policy.retry_timeout is None def test_durable_agent_retry_policy_env_override(self, mock_llm, monkeypatch): """Test that DAPR_API_MAX_RETRIES environment variable overrides max_attempts.""" From 8648765babfe5b3200100b265d0bd899550f186a Mon Sep 17 00:00:00 2001 From: Casper Nielsen Date: Thu, 18 Dec 2025 15:48:48 +0100 Subject: [PATCH 10/11] chore: formatting Signed-off-by: Casper Nielsen --- dapr_agents/agents/configs.py | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/dapr_agents/agents/configs.py b/dapr_agents/agents/configs.py index 8886501f..d7a8cbf9 100644 --- a/dapr_agents/agents/configs.py +++ b/dapr_agents/agents/configs.py @@ -2,7 +2,17 @@ import re from dataclasses import dataclass, field -from typing import Any, Callable, Dict, List, MutableMapping, Optional, Sequence, Type +from typing import ( + Any, + Callable, + Dict, + List, + MutableMapping, + Optional, + Sequence, + Type, + Union, +) from pydantic import BaseModel From c12edb4b31f08a9b6773606678bb49c263a8f4f7 Mon Sep 17 00:00:00 2001 From: Casper Nielsen Date: Thu, 18 Dec 2025 15:54:22 +0100 Subject: [PATCH 11/11] fix: add attribute doc for docstring Signed-off-by: Casper Nielsen --- dapr_agents/agents/configs.py | 1 + 1 file changed, 1 insertion(+) diff --git a/dapr_agents/agents/configs.py b/dapr_agents/agents/configs.py index d7a8cbf9..6becbbaa 100644 --- a/dapr_agents/agents/configs.py +++ b/dapr_agents/agents/configs.py @@ -287,6 +287,7 @@ class WorkflowRetryPolicy: initial_backoff_seconds: Initial backoff interval in seconds. max_backoff_seconds: Maximum backoff interval in seconds. backoff_multiplier: Multiplier for exponential backoff. + retry_timeout: Optional total timeout for all retries in seconds. """ max_attempts: Optional[int] = 1