pause

Author: luke-beamcloud

fastloop/context.py

@@ -12,6 +12,7 @@
     LoopStoppedError,
     WorkflowGotoError,
     WorkflowNextError,
+    WorkflowPauseError,
     WorkflowRepeatError,
 )
 from .logging import setup_logger
@@ -61,6 +62,7 @@ def __init__(
             i.type(): i.events() for i in integrations
         }
         self._integration_clients: dict[str, Any] = {}
+        self.resume_payload: dict[str, Any] | None = None
 
     def _reset_cycle_tracking(self) -> None:
         self.event_this_cycle = False
@@ -296,3 +298,7 @@ def goto(
     ) -> None:
         """Jump to a specific block in a workflow, optionally after a delay."""
         raise WorkflowGotoError(block_index, delay_seconds, reason)
+
+    def pause_until_resumed(self, reason: str | None = None) -> None:
+        """Pause the workflow indefinitely until resumed via API."""
+        raise WorkflowPauseError(reason)

fastloop/exceptions.py

@@ -62,6 +62,11 @@ def __init__(
         self.reason = reason
 
 
+class WorkflowPauseError(Exception):
+    def __init__(self, reason: str | None = None):
+        self.reason = reason
+
+
 class WorkflowNotFoundError(Exception):
     pass
 

fastloop/fastloop.py

@@ -681,6 +681,35 @@ async def _cancel_handler(workflow_run_id: str):
                         status_code=HTTPStatus.NOT_FOUND, detail=str(e)
                     ) from e
 
+            async def _resume_handler(
+                workflow_run_id: str, request: dict[str, Any] = {}
+            ):
+                try:
+                    workflow = await self.state_manager.get_workflow(workflow_run_id)
+                    if workflow.status != LoopStatus.PAUSED:
+                        raise HTTPException(
+                            status_code=HTTPStatus.BAD_REQUEST,
+                            detail=f"Workflow {workflow_run_id} is not paused",
+                        )
+                    payload = request.get("payload") if request else None
+                    if payload is not None:
+                        await self.state_manager.set_workflow_resume_payload(
+                            workflow_run_id, payload
+                        )
+                    await self.state_manager.mark_workflow_for_resume(workflow_run_id)
+                    restarted = await self.restart_workflow(workflow_run_id)
+                    if restarted:
+                        return JSONResponse(content={"message": "Workflow resumed"})
+                    else:
+                        raise HTTPException(
+                            status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
+                            detail="Failed to resume workflow",
+                        )
+                except WorkflowNotFoundError as e:
+                    raise HTTPException(
+                        status_code=HTTPStatus.NOT_FOUND, detail=str(e)
+                    ) from e
+
             async def _event_handler(request: dict[str, Any]):
                 workflow_run_id = request.get("workflow_run_id")
                 event_type = request.get("type")
@@ -725,6 +754,9 @@ async def _event_handler(request: dict[str, Any]):
             self.add_api_route(
                 f"/{name}/{{workflow_run_id}}/cancel", _cancel_handler, methods=["POST"]
             )
+            self.add_api_route(
+                f"/{name}/{{workflow_run_id}}/resume", _resume_handler, methods=["POST"]
+            )
 
             return func_or_class
 

fastloop/state/state.py

@@ -247,6 +247,22 @@ async def set_workflow_block_output(
     async def get_workflow_block_output(self, workflow_run_id: str) -> Any:
         pass
 
+    @abstractmethod
+    async def set_workflow_resume_payload(
+        self, workflow_run_id: str, payload: dict[str, Any] | None
+    ) -> None:
+        pass
+
+    @abstractmethod
+    async def get_workflow_resume_payload(
+        self, workflow_run_id: str
+    ) -> dict[str, Any] | None:
+        pass
+
+    @abstractmethod
+    async def mark_workflow_for_resume(self, workflow_run_id: str) -> None:
+        pass
+
 
 def create_state_manager(
     *,

fastloop/state/state_redis.py

@@ -67,6 +67,9 @@ class RedisKeys:
     WORKFLOW_BLOCK_OUTPUT = (
         f"{KEY_PREFIX}:{{app_name}}:workflow_block_output:{{workflow_run_id}}"
     )
+    WORKFLOW_RESUME_PAYLOAD = (
+        f"{KEY_PREFIX}:{{app_name}}:workflow_resume_payload:{{workflow_run_id}}"
+    )
     TASK_INDEX = f"{KEY_PREFIX}:{{app_name}}:task_index"
     TASK_STATE = f"{KEY_PREFIX}:{{app_name}}:task:{{task_id}}"
     TASK_CLAIM = f"{KEY_PREFIX}:{{app_name}}:task_claim:{{task_id}}"
@@ -1070,6 +1073,36 @@ async def get_workflow_block_output(self, workflow_run_id: str) -> Any:
             return cloudpickle.loads(output_bytes)
         return None
 
+    async def set_workflow_resume_payload(
+        self, workflow_run_id: str, payload: dict[str, Any] | None
+    ) -> None:
+        payload_key = RedisKeys.WORKFLOW_RESUME_PAYLOAD.format(
+            app_name=self.app_name, workflow_run_id=workflow_run_id
+        )
+        if payload is None:
+            await self.rdb.delete(payload_key)
+        else:
+            await self.rdb.set(payload_key, json.dumps(payload))
+
+    async def get_workflow_resume_payload(
+        self, workflow_run_id: str
+    ) -> dict[str, Any] | None:
+        payload_key = RedisKeys.WORKFLOW_RESUME_PAYLOAD.format(
+            app_name=self.app_name, workflow_run_id=workflow_run_id
+        )
+        payload_bytes = await self.rdb.get(payload_key)
+        if payload_bytes:
+            return json.loads(payload_bytes.decode("utf-8"))
+        return None
+
+    async def mark_workflow_for_resume(self, workflow_run_id: str) -> None:
+        workflow = await self.get_workflow(workflow_run_id)
+        if workflow.status != LoopStatus.PAUSED:
+            raise ValueError(f"Workflow {workflow_run_id} is not paused")
+        workflow.status = LoopStatus.IDLE
+        workflow.scheduled_wake_time = time.time()
+        await self.update_workflow(workflow_run_id, workflow)
+
     async def create_task(self, task: TaskState) -> TaskState:
         state_key = RedisKeys.TASK_STATE.format(
             app_name=self.app_name, task_id=task.task_id

fastloop/state/state_s3.py

@@ -741,3 +741,39 @@ async def get_workflow_block_output(self, workflow_id: str) -> Any:
             return cloudpickle.loads(response["Body"].read())
         except ClientError:
             return None
+
+    async def set_workflow_resume_payload(
+        self, workflow_id: str, payload: dict[str, Any] | None
+    ) -> None:
+        key = (
+            f"{self.prefix}/{self.app_name}/workflow_resume_payload/{workflow_id}.json"
+        )
+        if payload is None:
+            try:
+                self.s3.delete_object(Bucket=self.bucket, Key=key)
+            except ClientError:
+                pass
+        else:
+            self.s3.put_object(
+                Bucket=self.bucket, Key=key, Body=json.dumps(payload).encode("utf-8")
+            )
+
+    async def get_workflow_resume_payload(
+        self, workflow_id: str
+    ) -> dict[str, Any] | None:
+        key = (
+            f"{self.prefix}/{self.app_name}/workflow_resume_payload/{workflow_id}.json"
+        )
+        try:
+            response = self.s3.get_object(Bucket=self.bucket, Key=key)
+            return json.loads(response["Body"].read().decode("utf-8"))
+        except ClientError:
+            return None
+
+    async def mark_workflow_for_resume(self, workflow_id: str) -> None:
+        workflow = await self.get_workflow(workflow_id)
+        if workflow.status != LoopStatus.PAUSED:
+            raise ValueError(f"Workflow {workflow_id} is not paused")
+        workflow.status = LoopStatus.IDLE
+        workflow.scheduled_wake_time = time.time()
+        await self.update_workflow(workflow_id, workflow)

fastloop/types.py

@@ -14,13 +14,15 @@ class LoopStatus(StrEnum):
     PENDING = "pending"
     RUNNING = "running"
     IDLE = "idle"
+    PAUSED = "paused"
     STOPPED = "stopped"
     FAILED = "failed"
 
 
 class ScheduleType(StrEnum):
     IMMEDIATE = "immediate"
     DELAY = "delay"
+    PAUSE = "pause"
     STOP = "stop"
 
 

fastloop/workflow.py

@@ -21,6 +21,7 @@
     WorkflowGotoError,
     WorkflowMaxRetriesError,
     WorkflowNextError,
+    WorkflowPauseError,
     WorkflowRepeatError,
 )
 from .logging import setup_logger
@@ -160,12 +161,19 @@ async def _apply_plan(
         workflow_run_id: str,
         idx: int,
         blocks: list[WorkflowBlock],
-    ) -> tuple[int | None, float | None]:
+        block_output: Any = None,
+    ) -> tuple[int | None, float | None, bool]:
         if plan_result is None:
-            return idx + 1, None
+            return idx + 1, None, False
 
         if plan_result.schedule_type == ScheduleType.STOP:
-            return None, None
+            return None, None, False
+
+        if plan_result.schedule_type == ScheduleType.PAUSE:
+            await self._schedule_pause(
+                workflow_run_id, block_output, plan_result.reason
+            )
+            return idx + 1, None, True
 
         next_idx = plan_result.next_block_index
         if next_idx is None:
@@ -186,7 +194,7 @@ async def _apply_plan(
         if plan_result.schedule_type == ScheduleType.DELAY:
             delay = plan_result.delay_seconds
 
-        return next_idx, delay
+        return next_idx, delay, False
 
     async def _schedule_delay(
         self,
@@ -211,6 +219,28 @@ async def _schedule_delay(
         )
         raise LoopPausedError()
 
+    async def _schedule_pause(
+        self,
+        workflow_run_id: str,
+        block_output: Any,
+        reason: str | None = None,
+    ) -> None:
+        """Pause a workflow indefinitely until resumed via API."""
+        await self.state_manager.set_workflow_block_output(
+            workflow_run_id, block_output
+        )
+        await self.state_manager.update_workflow_status(
+            workflow_run_id, LoopStatus.PAUSED
+        )
+        logger.info(
+            "Workflow paused until resumed",
+            extra={
+                "workflow_run_id": workflow_run_id,
+                "reason": reason,
+            },
+        )
+        raise LoopPausedError()
+
     async def _run(
         self,
         func: Callable[..., Any],
@@ -267,6 +297,14 @@ async def _run(
                             workflow_run_id
                         )
                     )
+                    context.resume_payload = (
+                        await self.state_manager.get_workflow_resume_payload(
+                            workflow_run_id
+                        )
+                    )
+                    await self.state_manager.set_workflow_resume_payload(
+                        workflow_run_id, None
+                    )
 
                     try:
                         block_output = await _call_with_result(
@@ -293,10 +331,13 @@ async def _run(
                                     },
                                 )
 
-                        next_idx, delay = await self._apply_plan(
-                            plan_result, workflow_run_id, idx, blocks
+                        next_idx, delay, paused = await self._apply_plan(
+                            plan_result, workflow_run_id, idx, blocks, block_output
                         )
 
+                        if paused:
+                            continue
+
                         if next_idx is None:
                             await _call(on_block_complete, context, current_block, None)
                             raise LoopStoppedError()
@@ -358,6 +399,11 @@ async def _run(
                                 workflow_run_id, e.delay_seconds, None, e.reason
                             )
 
+                    except WorkflowPauseError as e:
+                        await self._schedule_pause(
+                            workflow_run_id, context.block_output, e.reason
+                        )
+
                     except (asyncio.CancelledError, LoopPausedError, LoopStoppedError):
                         raise
 
@@ -431,10 +477,11 @@ async def _run(
             )
             await _call(on_stop, context)
         except LoopPausedError:
-            await self.state_manager.update_workflow_status(
-                workflow_run_id, LoopStatus.IDLE
-            )
-            # Don't call on_stop - workflow is just paused, not finished
+            workflow = await self.state_manager.get_workflow(workflow_run_id)
+            if workflow.status != LoopStatus.PAUSED:
+                await self.state_manager.update_workflow_status(
+                    workflow_run_id, LoopStatus.IDLE
+                )
         finally:
             self.tasks.pop(workflow_run_id, None)
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "fastloop"
-version = "0.1.113"
+version = "0.1.114"
 description = "A Python package for deploying stateful loops"
 readme = "README.md"
 requires-python = ">=3.12"