handle looping better

Author: luke-beamcloud

fastloop/fastloop.py

@@ -109,14 +109,14 @@ async def lifespan(_: FastAPI):
                 allow_headers=cors_config.get("allow_headers", ["*"]),
             )
 
-        @self.get("/events/{loop_id}/history")
-        async def events_history_endpoint(loop_id: str):  # type: ignore
-            events = await self.state_manager.get_event_history(loop_id)
-            return [event.to_dict() for event in events]  # type: ignore
-
-        @self.get("/events/{loop_id}/sse")
-        async def events_sse_endpoint(loop_id: str):  # type: ignore
-            return await self.loop_manager.events_sse(loop_id)
+        @self.get("/events/{entity_id}/history")
+        async def events_history_endpoint(entity_id: str):  # type: ignore
+            events = await self.state_manager.get_event_history(entity_id)
+            return events
+
+        @self.get("/events/{entity_id}/sse")
+        async def events_sse_endpoint(entity_id: str):  # type: ignore
+            return await self.loop_manager.events_sse(entity_id)
 
     @property
     def config(self) -> BaseConfig:

fastloop/loop.py

@@ -369,28 +369,31 @@ async def active_loop_ids(self) -> set[str]:
 
         return {loop_id for loop_id, _ in self.loop_tasks.items()}
 
-    async def events_sse(self, loop_id: str):
+    async def events_sse(self, entity_id: str):
         """
         SSE endpoint for streaming events to clients.
         Works for both loops and workflows.
         """
-        # Check if it's a loop or workflow
+        # Check if it's a loop or workflow, get creation time for event filtering
+        created_at = None
         try:
-            await self.state_manager.get_loop(loop_id)
+            loop = await self.state_manager.get_loop(entity_id)
+            created_at = loop.created_at
         except LoopNotFoundError:
             try:
-                await self.state_manager.get_workflow(loop_id)
+                workflow = await self.state_manager.get_workflow(entity_id)
+                created_at = workflow.created_at
             except WorkflowNotFoundError as e:
                 raise HTTPException(
                     status_code=404, detail="Loop/workflow not found"
                 ) from e
 
-        connection_time = int(datetime.now().timestamp())
+        connection_time = int(created_at) if created_at else 0
         last_sent_nonce = 0
         connection_id = str(uuid.uuid4())
 
-        await self.state_manager.register_client_connection(loop_id, connection_id)
-        pubsub = await self.state_manager.subscribe_to_events(loop_id)
+        await self.state_manager.register_client_connection(entity_id, connection_id)
+        pubsub = await self.state_manager.subscribe_to_events(entity_id)
 
         async def _event_generator():
             nonlocal last_sent_nonce
@@ -400,7 +403,7 @@ async def _event_generator():
                     all_events: list[
                         dict[str, Any]
                     ] = await self.state_manager.get_events_since(
-                        loop_id, connection_time
+                        entity_id, connection_time
                     )
                     server_events = [
                         e
@@ -429,24 +432,24 @@ async def _event_generator():
 
                         # Refresh connection TTL periodically
                         await self.state_manager.refresh_client_connection(
-                            loop_id, connection_id
+                            entity_id, connection_id
                         )
 
             except asyncio.CancelledError:
                 pass
             except BaseException as e:
                 logger.error(
-                    "Error in SSE stream for loop",
+                    "Error in SSE stream",
                     extra={
-                        "loop_id": loop_id,
+                        "entity_id": entity_id,
                         "error": str(e),
                         "traceback": traceback.format_exc(),
                     },
                 )
                 yield f'data: {{"type": "error", "message": "{e!s}"}}\n\n'
             finally:
                 await self.state_manager.unregister_client_connection(
-                    loop_id, connection_id
+                    entity_id, connection_id
                 )
                 if pubsub is not None:
                     await pubsub.unsubscribe()  # type: ignore
@@ -514,9 +517,17 @@ async def _mark_block_completed(
         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)
+
+        # If going backwards, clear completed status for blocks we're rewinding to
+        if next_idx <= idx:
+            workflow.completed_blocks = [
+                b for b in workflow.completed_blocks if b < next_idx
+            ]
+
         await self.state_manager.update_workflow(workflow_run_id, workflow)
 
     async def _apply_plan(

fastloop/state/state_redis.py

@@ -620,7 +620,11 @@ async def get_events_since(
         Get events that occurred since the given timestamp.
         """
         all_events = await self.get_event_history(loop_id)
-        return [event for event in all_events if event["timestamp"] >= since_timestamp]
+        return [
+            event
+            for event in all_events
+            if float(event["timestamp"]) >= since_timestamp
+        ]
 
     async def subscribe_to_events(self, loop_id: str) -> Any:
         """Subscribe to event notifications for a specific loop"""

fastloop/state/state_s3.py

@@ -447,7 +447,9 @@ async def get_events_since(
         self, loop_id: str, since_timestamp: float
     ) -> list[dict[str, Any]]:
         history = await self.get_event_history(loop_id)
-        return [event for event in history if event["timestamp"] >= since_timestamp]
+        return [
+            event for event in history if float(event["timestamp"]) >= since_timestamp
+        ]
 
     async def pop_server_event(self, loop_id: str) -> dict[str, Any] | None:
         queue_key = S3Keys.loop_event_queue_server(self.prefix, self.app_name, loop_id)

pyproject.toml

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

tests/test_workflows.py

@@ -570,8 +570,12 @@ async def test_filter_by_status(self, state_manager):
             workflow_name="stopped", blocks=[{"type": "s", "text": ""}]
         )
 
-        await state_manager.update_workflow_status(w1.workflow_run_id, LoopStatus.RUNNING)
-        await state_manager.update_workflow_status(w2.workflow_run_id, LoopStatus.STOPPED)
+        await state_manager.update_workflow_status(
+            w1.workflow_run_id, LoopStatus.RUNNING
+        )
+        await state_manager.update_workflow_status(
+            w2.workflow_run_id, LoopStatus.STOPPED
+        )
 
         running = await state_manager.get_all_workflows(status=LoopStatus.RUNNING)
         assert len(running) == 1