wip

Author: luke-beamcloud

examples/workflows.py

@@ -33,7 +33,17 @@ def on_block_done(ctx: LoopContext, block: WorkflowBlock, payload: dict | None):
 
 
 def on_error(ctx: LoopContext, block: WorkflowBlock, error: Exception):
-    print(f"  ✗ Block error: {block.type} - {error}")
+    """
+    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
+    print(f"  ✗ Block failed: {block.type} - {error}")
 
 
 # --- Workflow ---

fastloop/loop.py

@@ -467,7 +467,11 @@ async def _run(
                                 "traceback": traceback.format_exc(),
                             },
                         )
-                        await _call(on_error, context, current_block, e)
+                        if on_error:
+                            try:
+                                await _call(on_error, context, current_block, e)
+                            except WorkflowRepeatError:
+                                continue  # Retry the block
                         raise LoopStoppedError() from e
 
         except asyncio.CancelledError:

pyproject.toml

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

tests/test_workflow_integration.py

@@ -321,6 +321,42 @@ async def func(ctx, blocks, block):
         assert errors[0][0] == "bad"
         assert "boom" in errors[0][1]
 
+    async def test_on_error_can_retry(self, mock_state):
+        """on_error can raise ctx.repeat() to retry the block."""
+        attempts = [0]
+
+        mock_state.get_workflow = AsyncMock(
+            return_value=WorkflowState(
+                workflow_id="test",
+                blocks=[{"type": "flaky", "text": ""}],
+                status=LoopStatus.RUNNING,
+            )
+        )
+
+        def on_error(ctx, block, error):
+            if attempts[0] < 3:
+                ctx.repeat()  # Retry
+
+        async def func(ctx, blocks, block):
+            attempts[0] += 1
+            if attempts[0] < 3:
+                raise ValueError("transient error")
+            # Succeeds on 3rd attempt
+
+        ctx = MagicMock()
+        ctx.repeat = LoopContext.repeat.__get__(ctx, LoopContext)
+
+        wm = WorkflowManager(mock_state)
+        await wm.start(
+            func,
+            ctx,
+            WorkflowState(workflow_id="test", blocks=[], status=LoopStatus.RUNNING),
+            on_error=on_error,
+        )
+        await asyncio.sleep(0.1)
+
+        assert attempts[0] == 3  # Tried 3 times, succeeded on 3rd
+
 
 # --- Context Attributes ---