improve workflows

Author: luke-beamcloud

examples/workflows.py

@@ -1,20 +1,12 @@
 """
-Workflow Example - Durable multi-step execution.
-
-Control flow:
-  ctx.next(payload)  → advance to next block
-  ctx.repeat()       → re-run current block
-  normal return      → workflow completes
+Workflow examples - function-based and class-based styles.
 """
 
-from fastloop import FastLoop, LoopContext, LoopEvent, WorkflowBlock
+from fastloop import FastLoop, LoopContext, LoopEvent, Workflow, WorkflowBlock
 
 app = FastLoop(name="workflow-demo")
 
 
-# --- Events ---
-
-
 @app.event("start_workflow")
 class StartWorkflow(LoopEvent):
     pass
@@ -25,30 +17,25 @@ class UserInput(LoopEvent):
     value: str
 
 
-# --- Callbacks (optional) ---
+@app.event("progress")
+class Progress(LoopEvent):
+    message: str
+    step: int
 
 
 def on_block_done(ctx: LoopContext, block: WorkflowBlock, payload: dict | None):
     print(f"  ✓ Block complete: {block.type}")
 
 
 def on_error(ctx: LoopContext, block: WorkflowBlock, error: Exception):
-    """
-    Called when a block raises an exception.
-    - Normal return → workflow stops
-    - Raise ctx.repeat() → retry the block
-    """
     retries = getattr(ctx, "_retries", 0)
     if retries < 3:
         ctx._retries = retries + 1
         print(f"  ⟳ Retrying block {block.type} (attempt {retries + 1})")
-        ctx.repeat()  # Raises WorkflowRepeatError → retry
+        ctx.repeat()
     print(f"  ✗ Block failed: {block.type} - {error}")
 
 
-# --- Workflow ---
-
-
 @app.workflow(
     name="onboarding",
     start_event=StartWorkflow,
@@ -60,14 +47,7 @@ async def onboarding_workflow(
     blocks: list[WorkflowBlock],
     current_block: WorkflowBlock,
 ):
-    """
-    Multi-step onboarding. State survives restarts.
-
-    ctx.block_index      - current position (0-indexed)
-    ctx.block_count      - total blocks
-    ctx.previous_payload - data from previous ctx.next(payload)
-    """
-    print(f"[{ctx.block_index + 1}/{ctx.block_count}] {current_block.type}")
+    await ctx.emit(Progress(message=current_block.text, step=ctx.block_index + 1))
 
     match current_block.type:
         case "collect_name":
@@ -93,7 +73,58 @@ async def onboarding_workflow(
         case "confirm":
             name = await ctx.get("user_name")
             email = await ctx.get("user_email")
-            print(f"Done: {name} <{email}>")
+            await ctx.emit(
+                Progress(message=f"Complete: {name} <{email}>", step=ctx.block_count)
+            )
+
+
+@app.event("start_survey")
+class StartSurvey(LoopEvent):
+    pass
+
+
+@app.workflow(name="survey", start_event=StartSurvey)
+class SurveyWorkflow(Workflow):
+    async def on_start(self, ctx: LoopContext) -> None:
+        print(f"[{ctx.loop_id}] Survey started")
+
+    async def on_stop(self, ctx: LoopContext) -> None:
+        print(f"[{ctx.loop_id}] Survey completed")
+
+    async def on_block_complete(
+        self, ctx: LoopContext, block: WorkflowBlock, payload: dict | None
+    ) -> None:
+        print(f"  ✓ Survey step complete: {block.type}")
+
+    async def on_error(
+        self, ctx: LoopContext, block: WorkflowBlock, error: Exception
+    ) -> None:
+        print(f"  ✗ Survey error: {error}")
+
+    async def execute(
+        self,
+        ctx: LoopContext,
+        blocks: list[WorkflowBlock],
+        current_block: WorkflowBlock,
+    ) -> None:
+        print(f"[{ctx.block_index + 1}/{ctx.block_count}] {current_block.type}")
+
+        match current_block.type:
+            case "question":
+                response = await ctx.wait_for(
+                    UserInput, timeout=60.0, raise_on_timeout=False
+                )
+                if response:
+                    await ctx.set(f"answer_{ctx.block_index}", response.value)
+                    ctx.next()
+                else:
+                    ctx.abort()
+
+            case "summary":
+                answers = [
+                    await ctx.get(f"answer_{i}") for i in range(ctx.block_count - 1)
+                ]
+                print(f"Survey results: {answers}")
 
 
 if __name__ == "__main__":
@@ -113,12 +144,15 @@ async def onboarding_workflow(
 #        ]
 #      }'
 #
-# 2. Send event to workflow (use workflow_id from response):
+# 2. Subscribe to SSE events (use workflow_id from step 1):
+#    curl -N http://localhost:8111/events/<workflow_id>/sse
+#
+# 3. Send event to workflow:
 #    curl -X POST http://localhost:8111/onboarding/<workflow_id>/event \
 #      -H "Content-Type: application/json" \
 #      -d '{"type": "user_input", "value": "John Doe", "workflow_id": "<id>"}'
 #
-# 3. Check workflow status:
+# 4. Check workflow status:
 #    curl http://localhost:8111/onboarding/<workflow_id>
 #
-# 4. If service restarts, workflow resumes from current block automatically
+# 5. If service restarts, workflow resumes from current block automatically

fastloop/__init__.py

@@ -1,13 +1,14 @@
 from . import integrations
 from .context import LoopContext
 from .fastloop import FastLoop
-from .loop import Loop, LoopEvent, WorkflowBlock
+from .loop import Loop, LoopEvent, Workflow, WorkflowBlock
 
 __all__ = [
     "FastLoop",
     "Loop",
     "LoopContext",
     "LoopEvent",
+    "Workflow",
     "WorkflowBlock",
     "integrations",
 ]

fastloop/context.py

@@ -60,15 +60,16 @@ def __init__(
         }
 
     def stop(self):
-        """Request the loop to stop on the next iteration."""
         self._stop_requested = True
         raise LoopStoppedError()
 
     def pause(self):
-        """Request the loop to pause on the next iteration."""
         self._pause_requested = True
         raise LoopPausedError()
 
+    def abort(self):
+        raise LoopStoppedError()
+
     def switch_to(self: T, func: Callable[[T], Awaitable[None]]):
         logger.info(
             f"Switching context to function: {get_func_import_path(func)}",

fastloop/fastloop.py

@@ -25,14 +25,25 @@
 )
 from .integrations import Integration
 from .logging import configure_logging, setup_logger
-from .loop import Loop, LoopEvent, LoopManager, WorkflowBlock, WorkflowManager
+from .loop import Loop, LoopEvent, LoopManager, Workflow, WorkflowBlock, WorkflowManager
 from .state.state import StateManager, create_state_manager
 from .types import BaseConfig, LoopStatus
 from .utils import get_func_import_path, import_func_from_path, infer_application_path
 
 logger = setup_logger()
 
 
+def _resolve_event_key(event: str | Enum | type[LoopEvent] | None) -> str | None:
+    """Convert event type/enum/string to string key."""
+    if not event:
+        return None
+    if isinstance(event, type) and issubclass(event, LoopEvent):
+        return event.type
+    if hasattr(event, "value"):
+        return event.value  # type: ignore
+    return event  # type: ignore
+
+
 class FastLoop(FastAPI):
     def __init__(
         self,
@@ -206,14 +217,7 @@ def _decorator(
                 )
                 integration.register(self, name)
 
-            start_event_key = None
-            if start_event:
-                if isinstance(start_event, type) and issubclass(start_event, LoopEvent):
-                    start_event_key = start_event.type
-                elif hasattr(start_event, "value"):
-                    start_event_key = start_event.value  # type: ignore
-                else:
-                    start_event_key = start_event
+            start_event_key = _resolve_event_key(start_event)
 
             if name not in self._loop_metadata:
                 self._loop_metadata[name] = {
@@ -456,62 +460,45 @@ def workflow(
         on_stop: Callable[..., Any] | None = None,
         on_block_complete: Callable[..., Any] | None = None,
         on_error: Callable[..., Any] | None = None,
-    ) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
-        """
-        Decorator to define a durable workflow.
-
-        A workflow processes a list of blocks sequentially. State is persisted
-        after each block transition, so workflows survive service restarts.
-
-        Args:
-            name: URL path for the workflow (e.g., "onboarding" → POST /onboarding)
-            start_event: Required event type to start the workflow
-            on_start: Called when workflow starts
-            on_stop: Called when workflow stops (complete or error)
-            on_block_complete: Called after each block completes
-            on_error: Called when a block raises an exception
-
-        The decorated function receives:
-            ctx: LoopContext with block_index, block_count, previous_payload
-            blocks: List of all WorkflowBlock objects
-            current_block: The block being executed
-
-        Control flow in the function:
-            ctx.next(payload)  → advance to next block
-            ctx.repeat()       → re-execute current block
-            normal return      → workflow completes
-
-        HTTP routes created:
-            POST /{name}                    → start workflow
-            GET  /{name}/{id}               → get workflow status
-            POST /{name}/{id}/event         → send event to workflow
-            POST /{name}/{id}/stop          → stop workflow
-        """
-
-        def _decorator(func: Callable[..., Any]) -> Callable[..., Any]:
-            # Resolve start_event to a string key
-            start_event_key = None
-            if start_event:
-                if isinstance(start_event, type) and issubclass(start_event, LoopEvent):
-                    start_event_key = start_event.type
-                elif hasattr(start_event, "value"):
-                    start_event_key = start_event.value
-                else:
-                    start_event_key = start_event
+    ) -> Callable[
+        [Callable[..., Any] | type[Workflow]], Callable[..., Any] | type[Workflow]
+    ]:
+        def _decorator(
+            func_or_class: Callable[..., Any] | type[Workflow],
+        ) -> Callable[..., Any] | type[Workflow]:
+            is_class_based = isinstance(func_or_class, type) and issubclass(
+                func_or_class, Workflow
+            )
+
+            if is_class_based:
+                workflow_instance: Workflow = func_or_class()
+                func = workflow_instance.execute
+                workflow_on_start = workflow_instance.on_start
+                workflow_on_stop = workflow_instance.on_stop
+                workflow_on_block_complete = workflow_instance.on_block_complete
+                workflow_on_error = workflow_instance.on_error
+            else:
+                workflow_instance = None  # type: ignore
+                func = func_or_class  # type: ignore
+                workflow_on_start = on_start
+                workflow_on_stop = on_stop
+                workflow_on_block_complete = on_block_complete
+                workflow_on_error = on_error
+
+            start_event_key = _resolve_event_key(start_event)
 
             if name in self._workflow_metadata:
                 raise LoopAlreadyDefinedError(f"Workflow {name} already registered")
 
             self._workflow_metadata[name] = {
                 "func": func,
-                "on_start": on_start,
-                "on_stop": on_stop,
-                "on_block_complete": on_block_complete,
-                "on_error": on_error,
+                "on_start": workflow_on_start,
+                "on_stop": workflow_on_stop,
+                "on_block_complete": workflow_on_block_complete,
+                "on_error": workflow_on_error,
+                "workflow_instance": workflow_instance,
             }
 
-            # --- HTTP Handlers ---
-
             class WorkflowStartRequest(BaseModel):
                 type: str
                 blocks: list[WorkflowBlock]
@@ -557,10 +544,10 @@ async def _start_handler(request: WorkflowStartRequest):
                     func,
                     context,
                     workflow,
-                    on_start=on_start,
-                    on_stop=on_stop,
-                    on_block_complete=on_block_complete,
-                    on_error=on_error,
+                    on_start=workflow_on_start,
+                    on_stop=workflow_on_stop,
+                    on_block_complete=workflow_on_block_complete,
+                    on_error=workflow_on_error,
                 )
                 return (
                     await self.state_manager.get_workflow(workflow.workflow_id)
@@ -621,7 +608,6 @@ async def _event_handler(request: dict[str, Any]):
                 await self.state_manager.push_event(workflow_id, event)
                 return workflow.to_dict()
 
-            # --- Register routes ---
             self.add_api_route(f"/{name}", _start_handler, methods=["POST"])
             self.add_api_route(
                 f"/{name}/{{workflow_id}}", _get_handler, methods=["GET"]
@@ -633,7 +619,7 @@ async def _event_handler(request: dict[str, Any]):
                 f"/{name}/{{workflow_id}}/stop", _stop_handler, methods=["POST"]
             )
 
-            return func
+            return func_or_class
 
         return _decorator