Validate SDK durable timeout budgets

Author: durable-workflow-ops

src/durable_workflow/workflow.py

@@ -267,6 +267,13 @@ class ChildWorkflowRetryPolicy(ActivityRetryPolicy):
 ChildWorkflowRetryPolicyInput = ChildWorkflowRetryPolicy | ActivityRetryPolicy | Mapping[str, Any]
 
 
+def _validate_positive_timeout(name: str, value: int | None) -> None:
+    if value is None:
+        return
+    if value < 1:
+        raise ValueError(f"{name} must be >= 1 second")
+
+
 @dataclass
 class ScheduleActivity:
     """Command requesting an activity task.
@@ -296,6 +303,8 @@ def to_server_command(
         size_warning: serializer.PayloadSizeWarningConfig | None = serializer.DEFAULT_PAYLOAD_SIZE_WARNING,
         warning_context: PayloadWarningContext = None,
     ) -> dict[str, Any]:
+        self._validate_timeouts()
+
         command: dict[str, Any] = {
             "type": "schedule_activity",
             "activity_type": self.activity_type,
@@ -328,6 +337,38 @@ def to_server_command(
             command["heartbeat_timeout"] = self.heartbeat_timeout
         return command
 
+    def _validate_timeouts(self) -> None:
+        timeouts = {
+            "start_to_close_timeout": self.start_to_close_timeout,
+            "schedule_to_start_timeout": self.schedule_to_start_timeout,
+            "schedule_to_close_timeout": self.schedule_to_close_timeout,
+            "heartbeat_timeout": self.heartbeat_timeout,
+        }
+        for name, value in timeouts.items():
+            _validate_positive_timeout(name, value)
+
+        if (
+            self.heartbeat_timeout is not None
+            and self.start_to_close_timeout is not None
+            and self.heartbeat_timeout > self.start_to_close_timeout
+        ):
+            raise ValueError("heartbeat_timeout must be <= start_to_close_timeout")
+
+        if self.schedule_to_close_timeout is None:
+            return
+
+        if (
+            self.start_to_close_timeout is not None
+            and self.start_to_close_timeout > self.schedule_to_close_timeout
+        ):
+            raise ValueError("start_to_close_timeout must be <= schedule_to_close_timeout")
+
+        if (
+            self.schedule_to_start_timeout is not None
+            and self.schedule_to_start_timeout > self.schedule_to_close_timeout
+        ):
+            raise ValueError("schedule_to_start_timeout must be <= schedule_to_close_timeout")
+
 
 @dataclass
 class StartTimer:
@@ -555,6 +596,8 @@ def to_server_command(
         size_warning: serializer.PayloadSizeWarningConfig | None = serializer.DEFAULT_PAYLOAD_SIZE_WARNING,
         warning_context: PayloadWarningContext = None,
     ) -> dict[str, Any]:
+        self._validate_timeouts()
+
         cmd: dict[str, Any] = {
             "type": "start_child_workflow",
             "workflow_type": self.workflow_type,
@@ -587,6 +630,17 @@ def to_server_command(
             cmd["run_timeout_seconds"] = self.run_timeout_seconds
         return cmd
 
+    def _validate_timeouts(self) -> None:
+        _validate_positive_timeout("execution_timeout_seconds", self.execution_timeout_seconds)
+        _validate_positive_timeout("run_timeout_seconds", self.run_timeout_seconds)
+
+        if (
+            self.execution_timeout_seconds is not None
+            and self.run_timeout_seconds is not None
+            and self.run_timeout_seconds > self.execution_timeout_seconds
+        ):
+            raise ValueError("run_timeout_seconds must be <= execution_timeout_seconds")
+
 
 @dataclass
 class RecordVersionMarker:

tests/test_replay.py

@@ -220,6 +220,61 @@ def test_schedule_activity_server_command_includes_retry_policy_and_timeouts(sel
         assert server_cmd["schedule_to_close_timeout"] == 300
         assert server_cmd["heartbeat_timeout"] == 15
 
+    @pytest.mark.parametrize(
+        ("field", "value", "message"),
+        [
+            ("start_to_close_timeout", 0, "start_to_close_timeout must be >= 1 second"),
+            ("schedule_to_start_timeout", 0, "schedule_to_start_timeout must be >= 1 second"),
+            ("schedule_to_close_timeout", 0, "schedule_to_close_timeout must be >= 1 second"),
+            ("heartbeat_timeout", 0, "heartbeat_timeout must be >= 1 second"),
+        ],
+    )
+    def test_schedule_activity_rejects_non_positive_timeout_budgets(
+        self,
+        field: str,
+        value: int,
+        message: str,
+    ) -> None:
+        cmd = ScheduleActivity(
+            activity_type="charge-card",
+            arguments=[],
+            **{field: value},
+        )
+
+        with pytest.raises(ValueError, match=message):
+            cmd.to_server_command("default-queue")
+
+    @pytest.mark.parametrize(
+        ("kwargs", "message"),
+        [
+            (
+                {"start_to_close_timeout": 10, "heartbeat_timeout": 11},
+                "heartbeat_timeout must be <= start_to_close_timeout",
+            ),
+            (
+                {"start_to_close_timeout": 301, "schedule_to_close_timeout": 300},
+                "start_to_close_timeout must be <= schedule_to_close_timeout",
+            ),
+            (
+                {"schedule_to_start_timeout": 301, "schedule_to_close_timeout": 300},
+                "schedule_to_start_timeout must be <= schedule_to_close_timeout",
+            ),
+        ],
+    )
+    def test_schedule_activity_rejects_incoherent_timeout_envelopes(
+        self,
+        kwargs: dict[str, int],
+        message: str,
+    ) -> None:
+        cmd = ScheduleActivity(
+            activity_type="charge-card",
+            arguments=[],
+            **kwargs,
+        )
+
+        with pytest.raises(ValueError, match=message):
+            cmd.to_server_command("default-queue")
+
 
 class TestTwoActivities:
     def test_first_schedules(self) -> None:
@@ -728,6 +783,39 @@ def test_server_command_shape(self) -> None:
         assert sc["execution_timeout_seconds"] == 600
         assert sc["run_timeout_seconds"] == 120
 
+    @pytest.mark.parametrize(
+        ("field", "value", "message"),
+        [
+            ("execution_timeout_seconds", 0, "execution_timeout_seconds must be >= 1 second"),
+            ("run_timeout_seconds", 0, "run_timeout_seconds must be >= 1 second"),
+        ],
+    )
+    def test_server_command_rejects_non_positive_child_timeout_budgets(
+        self,
+        field: str,
+        value: int,
+        message: str,
+    ) -> None:
+        cmd = StartChildWorkflow(
+            workflow_type="sub",
+            arguments=[],
+            **{field: value},
+        )
+
+        with pytest.raises(ValueError, match=message):
+            cmd.to_server_command("default-q")
+
+    def test_server_command_rejects_child_run_timeout_larger_than_execution_timeout(self) -> None:
+        cmd = StartChildWorkflow(
+            workflow_type="sub",
+            arguments=[],
+            execution_timeout_seconds=120,
+            run_timeout_seconds=121,
+        )
+
+        with pytest.raises(ValueError, match="run_timeout_seconds must be <= execution_timeout_seconds"):
+            cmd.to_server_command("default-q")
+
     def test_server_command_defaults(self) -> None:
         cmd = StartChildWorkflow(workflow_type="sub", arguments=[])
         sc = cmd.to_server_command("default-q")