cancel workflows

Author: luke-beamcloud

examples/workflows.py

@@ -243,20 +243,23 @@ async def execute(
 #        ]
 #      }'
 #
-# 2. Subscribe to SSE events (use workflow_id from step 1):
-#    curl -N http://localhost:8111/events/<workflow_id>/sse
+# 2. Subscribe to SSE events (use workflow_run_id from step 1):
+#    curl -N http://localhost:8111/events/<workflow_run_id>/sse
 #
 # 3. Send event to workflow:
-#    curl -X POST http://localhost:8111/onboarding/<workflow_id>/event \
+#    curl -X POST http://localhost:8111/onboarding/<workflow_run_id>/event \
 #      -H "Content-Type: application/json" \
-#      -d '{"type": "user_input", "value": "John Doe", "workflow_id": "<id>"}'
+#      -d '{"type": "user_input", "value": "John Doe", "workflow_run_id": "<id>"}'
 #
 # 4. Check workflow status:
-#    curl http://localhost:8111/onboarding/<workflow_id>
+#    curl http://localhost:8111/onboarding/<workflow_run_id>
 #
-# 5. If service restarts, workflow resumes from current block automatically
+# 5. Cancel a running workflow:
+#    curl -X POST http://localhost:8111/onboarding/<workflow_run_id>/cancel
 #
-# 6. Start email monitor workflow with plan function:
+# 6. If service restarts, workflow resumes from current block automatically
+#
+# 7. Start email monitor workflow with plan function:
 #    curl -X POST http://localhost:8111/email_monitor \
 #      -H "Content-Type: application/json" \
 #      -d '{

fastloop/exceptions.py

@@ -69,15 +69,15 @@ class WorkflowNotFoundError(Exception):
 class WorkflowMaxRetriesError(Exception):
     def __init__(
         self,
-        workflow_id: str,
+        workflow_run_id: str,
         block_index: int,
         attempts: int,
         last_error: str | None = None,
     ):
-        self.workflow_id = workflow_id
+        self.workflow_run_id = workflow_run_id
         self.block_index = block_index
         self.attempts = attempts
         self.last_error = last_error
         super().__init__(
-            f"Workflow {workflow_id} block {block_index} failed after {attempts} attempts"
+            f"Workflow run {workflow_run_id} block {block_index} failed after {attempts} attempts"
         )

fastloop/fastloop.py

@@ -159,7 +159,7 @@ class WorkflowBlock(BaseModel):
 class WorkflowState:
     """Persisted state for a running workflow."""
 
-    workflow_id: str
+    workflow_run_id: str
     workflow_name: str | None = None
     created_at: float = field(default_factory=lambda: datetime.now().timestamp())
     status: LoopStatus = LoopStatus.PENDING
@@ -499,30 +499,30 @@ def __init__(self, state_manager: StateManager):
         self.state_manager = state_manager
 
     async def _persist_block_attempt(
-        self, workflow_id: str, idx: int, error: str | None = None
+        self, workflow_run_id: str, idx: int, error: str | None = None
     ) -> int:
-        workflow = await self.state_manager.get_workflow(workflow_id)
+        workflow = await self.state_manager.get_workflow(workflow_run_id)
         attempts = workflow.block_attempts.get(idx, 0) + 1
         workflow.block_attempts[idx] = attempts
         workflow.last_error = error
-        await self.state_manager.update_workflow(workflow_id, workflow)
+        await self.state_manager.update_workflow(workflow_run_id, workflow)
         return attempts
 
     async def _mark_block_completed(
-        self, workflow_id: str, idx: int, next_idx: int, payload: dict | None
+        self, workflow_run_id: str, idx: int, next_idx: int, payload: dict | None
     ) -> None:
-        workflow = await self.state_manager.get_workflow(workflow_id)
+        workflow = await self.state_manager.get_workflow(workflow_run_id)
         if idx not in workflow.completed_blocks:
             workflow.completed_blocks.append(idx)
         workflow.current_block_index = next_idx
         workflow.next_payload = payload
         workflow.block_attempts.pop(idx, None)
-        await self.state_manager.update_workflow(workflow_id, workflow)
+        await self.state_manager.update_workflow(workflow_run_id, workflow)
 
     async def _apply_plan(
         self,
         plan_result: BlockPlan | None,
-        workflow_id: str,
+        workflow_run_id: str,
         idx: int,
         blocks: list[WorkflowBlock],
     ) -> tuple[int | None, float | None]:
@@ -540,7 +540,7 @@ async def _apply_plan(
             logger.warning(
                 "Plan returned invalid block index, using next sequential",
                 extra={
-                    "workflow_id": workflow_id,
+                    "workflow_run_id": workflow_run_id,
                     "requested_index": next_idx,
                     "block_count": len(blocks),
                 },
@@ -555,19 +555,21 @@ async def _apply_plan(
 
     async def _schedule_delay(
         self,
-        workflow_id: str,
+        workflow_run_id: str,
         delay_seconds: float,
         block_output: Any,
         reason: str | None = None,
     ) -> None:
         """Schedule a workflow to resume after a delay using Redis scheduler."""
         wake_time = datetime.now().timestamp() + delay_seconds
-        await self.state_manager.set_workflow_block_output(workflow_id, block_output)
-        await self.state_manager.set_workflow_wake_time(workflow_id, wake_time)
+        await self.state_manager.set_workflow_block_output(
+            workflow_run_id, block_output
+        )
+        await self.state_manager.set_workflow_wake_time(workflow_run_id, wake_time)
         logger.info(
             "Workflow scheduled to resume",
             extra={
-                "workflow_id": workflow_id,
+                "workflow_run_id": workflow_run_id,
                 "delay_seconds": delay_seconds,
                 "reason": reason,
             },
@@ -578,17 +580,17 @@ async def _run(
         self,
         func: Callable[..., Any],
         context: Any,
-        workflow_id: str,
+        workflow_run_id: str,
         on_stop: Callable[..., Any] | None,
         on_block_complete: Callable[..., Any] | None,
         on_error: Callable[..., Any] | None,
         plan_func: Callable[..., Any] | None,
         retry_policy: RetryPolicy,
     ) -> None:
         try:
-            async with self.state_manager.with_workflow_claim(workflow_id):
+            async with self.state_manager.with_workflow_claim(workflow_run_id):
                 while True:
-                    workflow = await self.state_manager.get_workflow(workflow_id)
+                    workflow = await self.state_manager.get_workflow(workflow_run_id)
                     if workflow.status in (LoopStatus.STOPPED, LoopStatus.FAILED):
                         break
 
@@ -600,7 +602,7 @@ async def _run(
 
                     if idx in workflow.completed_blocks:
                         await self._mark_block_completed(
-                            workflow_id, idx, idx + 1, workflow.next_payload
+                            workflow_run_id, idx, idx + 1, workflow.next_payload
                         )
                         continue
 
@@ -611,7 +613,9 @@ async def _run(
                     context.current_block = current_block
                     context.previous_payload = workflow.next_payload
                     context.block_output = (
-                        await self.state_manager.get_workflow_block_output(workflow_id)
+                        await self.state_manager.get_workflow_block_output(
+                            workflow_run_id
+                        )
                     )
 
                     try:
@@ -631,7 +635,7 @@ async def _run(
                                 logger.info(
                                     "Plan function returned result",
                                     extra={
-                                        "workflow_id": workflow_id,
+                                        "workflow_run_id": workflow_run_id,
                                         "block_index": idx,
                                         "plan": plan_result.to_dict()
                                         if isinstance(plan_result, BlockPlan)
@@ -640,7 +644,7 @@ async def _run(
                                 )
 
                         next_idx, delay = await self._apply_plan(
-                            plan_result, workflow_id, idx, blocks
+                            plan_result, workflow_run_id, idx, blocks
                         )
 
                         if next_idx is None:
@@ -651,28 +655,28 @@ async def _run(
                         # If staying on same block (retry/loop), don't mark completed
                         if next_idx != idx:
                             await self._mark_block_completed(
-                                workflow_id, idx, next_idx, None
+                                workflow_run_id, idx, next_idx, None
                             )
                             await _call(on_block_complete, context, current_block, None)
                         else:
                             # Staying on same block - just update the current index
                             workflow = await self.state_manager.get_workflow(
-                                workflow_id
+                                workflow_run_id
                             )
                             workflow.current_block_index = next_idx
                             await self.state_manager.update_workflow(
-                                workflow_id, workflow
+                                workflow_run_id, workflow
                             )
 
                         if delay and delay > 0:
                             reason = plan_result.reason if plan_result else None
                             await self._schedule_delay(
-                                workflow_id, delay, block_output, reason
+                                workflow_run_id, delay, block_output, reason
                             )
 
                     except WorkflowNextError as e:
                         await self._mark_block_completed(
-                            workflow_id, idx, idx + 1, e.payload
+                            workflow_run_id, idx, idx + 1, e.payload
                         )
                         await _call(
                             on_block_complete, context, current_block, e.payload
@@ -687,21 +691,21 @@ async def _run(
                             logger.warning(
                                 "goto() called with invalid index, stopping",
                                 extra={
-                                    "workflow_id": workflow_id,
+                                    "workflow_run_id": workflow_run_id,
                                     "requested_index": next_idx,
                                     "block_count": len(blocks),
                                 },
                             )
                             raise LoopStoppedError() from None
 
                         await self._mark_block_completed(
-                            workflow_id, idx, next_idx, None
+                            workflow_run_id, idx, next_idx, None
                         )
                         await _call(on_block_complete, context, current_block, None)
 
                         if e.delay_seconds and e.delay_seconds > 0:
                             await self._schedule_delay(
-                                workflow_id, e.delay_seconds, None, e.reason
+                                workflow_run_id, e.delay_seconds, None, e.reason
                             )
 
                     except (asyncio.CancelledError, LoopPausedError, LoopStoppedError):
@@ -712,15 +716,15 @@ async def _run(
                         logger.error(
                             "Workflow block error",
                             extra={
-                                "workflow_id": workflow_id,
+                                "workflow_run_id": workflow_run_id,
                                 "block_index": idx,
                                 "error": error_str,
                                 "traceback": traceback.format_exc(),
                             },
                         )
 
                         attempts = await self._persist_block_attempt(
-                            workflow_id, idx, error_str
+                            workflow_run_id, idx, error_str
                         )
 
                         should_retry = False
@@ -738,7 +742,7 @@ async def _run(
                             logger.info(
                                 "Retrying workflow block",
                                 extra={
-                                    "workflow_id": workflow_id,
+                                    "workflow_run_id": workflow_run_id,
                                     "block_index": idx,
                                     "attempt": attempts,
                                     "delay": delay,
@@ -748,39 +752,41 @@ async def _run(
                             continue
 
                         max_retries_error = WorkflowMaxRetriesError(
-                            workflow_id, idx, attempts, error_str
+                            workflow_run_id, idx, attempts, error_str
                         )
                         logger.error(
                             "Workflow block failed after max retries",
                             extra={
-                                "workflow_id": workflow_id,
+                                "workflow_run_id": workflow_run_id,
                                 "block_index": idx,
                                 "attempts": attempts,
                             },
                         )
                         await _call(on_error, context, current_block, max_retries_error)
                         await self.state_manager.update_workflow_status(
-                            workflow_id, LoopStatus.FAILED
+                            workflow_run_id, LoopStatus.FAILED
                         )
                         await _call(on_stop, context)
                         return
 
         except asyncio.CancelledError:
             pass
         except LoopClaimError:
-            logger.warning("Workflow claim failed", extra={"workflow_id": workflow_id})
+            logger.warning(
+                "Workflow claim failed", extra={"workflow_run_id": workflow_run_id}
+            )
         except LoopStoppedError:
             await self.state_manager.update_workflow_status(
-                workflow_id, LoopStatus.STOPPED
+                workflow_run_id, LoopStatus.STOPPED
             )
             await _call(on_stop, context)
         except LoopPausedError:
             await self.state_manager.update_workflow_status(
-                workflow_id, LoopStatus.IDLE
+                workflow_run_id, LoopStatus.IDLE
             )
             # Don't call on_stop - workflow is just paused, not finished
         finally:
-            self.tasks.pop(workflow_id, None)
+            self.tasks.pop(workflow_run_id, None)
 
     async def start(
         self,
@@ -794,16 +800,16 @@ async def start(
         plan: Callable[..., Any] | None = None,
         retry_policy: RetryPolicy | None = None,
     ) -> bool:
-        if workflow.workflow_id in self.tasks:
+        if workflow.workflow_run_id in self.tasks:
             return False
 
         await _call(on_start, context)
 
-        self.tasks[workflow.workflow_id] = asyncio.create_task(
+        self.tasks[workflow.workflow_run_id] = asyncio.create_task(
             self._run(
                 func,
                 context,
-                workflow.workflow_id,
+                workflow.workflow_run_id,
                 on_stop,
                 on_block_complete,
                 on_error,
@@ -813,8 +819,8 @@ async def start(
         )
         return True
 
-    async def stop(self, workflow_id: str) -> bool:
-        task = self.tasks.pop(workflow_id, None)
+    async def stop(self, workflow_run_id: str) -> bool:
+        task = self.tasks.pop(workflow_run_id, None)
         if not task:
             return False
         task.cancel()