fixes: review feedback

Author: Spatika-Ganesh

workflowai/core/client/client.py

@@ -1,6 +1,7 @@
 import asyncio
 import importlib.metadata
 import os
+from email.utils import parsedate_to_datetime
 from typing import (
     Any,
     AsyncIterator,
@@ -138,24 +139,62 @@ async def run(
         if not stream:
             res = None
             delay = retry_delay / 1000
-            for _ in range(max_retry_count):
+            retry_count = 0
+            while retry_count <= max_retry_count:
                 try:
                     res = await self.api.post(route, request, returns=TaskRunResponse)
                     return res.to_domain(task)
                 except HTTPStatusError as e:
                     if e.response.status_code == 404:
                         raise NotFoundError("Task not found")
-                    if e.response.status_code == 429:
+                    retry_after = e.response.headers.get("Retry-After")
+                    if retry_after:
+                        try: 
+                            #for 429 errors this is non-negative decimal
+                            delay = float(retry_after) 
+                        except ValueError:
+                            try:
+                                retry_after_date = parsedate_to_datetime(retry_after)
+                                current_time = asyncio.get_event_loop().time()
+                                delay = (retry_after_date.timestamp()- current_time)
+                            except (TypeError, ValueError, OverflowError):
+                                delay = min(delay * 2, max_retry_delay / 1000)
+                        await asyncio.sleep(delay)
+                    elif e.response.status_code == 429:
                         if delay < max_retry_delay / 1000:
                             delay = min(delay * 2, max_retry_delay / 1000)
                         await asyncio.sleep(delay)
-
+                retry_count += 1
+        
         async def _stream():
-            async for chunk in self.api.stream(
-                method="POST", path=route, data=request, returns=RunTaskStreamChunk
-            ):
-                yield task.output_class.model_construct(None, **chunk.task_output)
-
+            delay = retry_delay / 1000
+            retry_count = 0
+            while retry_count <= max_retry_count:
+                try:
+                    async for chunk in self.api.stream(
+                        method="POST", path=route, data=request, returns=RunTaskStreamChunk
+                    ):
+                        yield task.output_class.model_construct(None, **chunk.task_output)
+                except HTTPStatusError as e:
+                    if e.response.status_code == 404:
+                        raise NotFoundError("Task not found")
+                    retry_after = e.response.headers.get("Retry-After")
+
+                    if retry_after:
+                        try:
+                            delay = float(retry_after)
+                        except ValueError:
+                            try:
+                                retry_after_date = parsedate_to_datetime(retry_after)
+                                current_time = asyncio.get_event_loop().time()
+                                delay = (retry_after_date.timestamp() - current_time)
+                            except (TypeError, ValueError, OverflowError):
+                                delay = min(delay * 2, max_retry_delay / 1000)
+                    elif e.response.status_code == 429:
+                        if delay < max_retry_delay / 1000:
+                            delay = min(delay * 2, max_retry_delay / 1000)
+                    await asyncio.sleep(delay)
+                retry_count += 1
         return _stream()
 
     async def import_run(