update links

cookbook/pocketflow-fastapi-hitl/README.md

@@ -0,0 +1,82 @@
+# PocketFlow Web Human-in-the-Loop (HITL) Feedback Service
+
+This project demonstrates a minimal web application for human-in-the-loop workflows using PocketFlow, FastAPI, and Server-Sent Events (SSE). Users can submit text, have it processed (simulated), review the output, and approve or reject it, potentially triggering reprocessing until approved.
+
+<p align="center">
+  <img 
+    src="./assets/banner.png" width="800"
+  />
+</p>
+
+## Features
+
+-   **Web UI:** Simple interface for submitting tasks and providing feedback.
+-   **PocketFlow Workflow:** Manages the process -> review -> result/reprocess logic.
+-   **FastAPI Backend:** Serves the UI and handles API requests asynchronously.
+-   **Server-Sent Events (SSE):** Provides real-time status updates to the client without polling.
+
+## How to Run
+
+1.  Install Dependencies:
+    ```bash
+    pip install -r requirements.txt
+    ```
+
+2.  Run the FastAPI Server:
+    Use Uvicorn (or another ASGI server):
+    ```bash
+    uvicorn server:app --reload --port 8000
+    ```
+    *(The `--reload` flag is useful for development.)*
+
+3.  Access the Web UI:
+    Open your web browser and navigate to `http://127.0.0.1:8000`.
+
+4.  Use the Application:
+    *   Enter text into the textarea and click "Submit".
+    *   Observe the status updates pushed via SSE.
+    *   When prompted ("waiting_for_review"), use the "Approve" or "Reject" buttons.
+    *   If rejected, the process loops back. If approved, the final result is displayed.
+
+## How It Works
+
+The application uses PocketFlow to define and execute the feedback loop workflow. FastAPI handles web requests and manages the real-time SSE communication.
+
+**PocketFlow Workflow:**
+
+The core logic is orchestrated by an `AsyncFlow` defined in `flow.py`:
+
+```mermaid
+flowchart TD
+    subgraph FeedbackFlow[MinimalFeedbackFlow]
+        Process[ProcessNode] -- default --> Review[ReviewNode]
+        Review -- approved --> Result[ResultNode]
+        Review -- rejected --> Process
+    end
+```
+
+1.  **`ProcessNode`**: Receives input text, calls the minimal `process_task` utility, and stores the output.
+2.  **`ReviewNode` (Async)**:
+    *   Pushes a "waiting_for_review" status with the processed output to the SSE queue.
+    *   Waits asynchronously for an external signal (triggered by the `/feedback` API endpoint).
+    *   Based on the received feedback ("approved" or "rejected"), determines the next step in the flow. Stores the result if approved.
+3.  **`ResultNode`**: Logs the final approved result.
+
+**FastAPI & SSE Integration:**
+
+*   The `/submit` endpoint creates a unique task, initializes the PocketFlow `shared` state (including an `asyncio.Event` for review and an `asyncio.Queue` for SSE), and schedules the flow execution using `BackgroundTasks`.
+*   Nodes within the flow (specifically `ReviewNode`'s prep logic) put status updates onto the task-specific `sse_queue`.
+*   The `/stream/{task_id}` endpoint uses `StreamingResponse` to read from the task's `sse_queue` and push formatted status updates to the connected client via Server-Sent Events.
+*   The `/feedback/{task_id}` endpoint receives the human's decision, updates the `shared` state, and sets the `asyncio.Event` to unblock the waiting `ReviewNode`.
+
+This setup allows for a decoupled workflow logic (PocketFlow) and web interaction layer (FastAPI), with efficient real-time updates pushed to the user.
+
+## Files
+
+-   [`server.py`](./server.py): The main FastAPI application handling HTTP requests, SSE, state management, and background task scheduling.
+-   [`nodes.py`](./nodes.py): Defines the PocketFlow `Node` classes (`ProcessNode`, `ReviewNode`, `ResultNode`) for the workflow steps.
+-   [`flow.py`](./flow.py): Defines the PocketFlow `AsyncFlow` that connects the nodes into the feedback loop.
+-   [`utils/process_task.py`](./utils/process_task.py): Contains the minimal simulation function for task processing.
+-   [`templates/index.html`](./templates/index.html): The HTML structure for the frontend user interface.
+-   [`static/style.css`](./static/style.css): Basic CSS for styling the frontend.
+-   [`requirements.txt`](./requirements.txt): Project dependencies (FastAPI, Uvicorn, Jinja2, PocketFlow).

cookbook/pocketflow-fastapi-hitl/docs/design.md

@@ -0,0 +1,55 @@
+#  Human-in-the-Loop Web Service
+
+## 1. Requirements
+
+*   **Goal:** Create a web service for task submission, processing, human review (Approve/Reject loop via UI), and finalization.
+*   **Interface:** Simple web UI (HTML/JS) for input, status display, and feedback buttons.
+*   **Backend:** FastAPI using PocketFlow for workflow management.
+*   **Real-time Updates:** Use Server-Sent Events (SSE) to push status changes (pending, running, waiting_for_review, completed, failed) and intermediate results to the client without page reloads.
+*   **State:** Use in-memory storage for task state (Warning: Not suitable for production).
+
+## 2. Flow Design
+
+*   **Core Pattern:** Workflow with a conditional loop based on human feedback. SSE for asynchronous status communication.
+*   **Nodes:**
+    1.  `ProcessNode` (Regular): Takes input, executes the (simulated) task processing.
+    2.  `ReviewNode` (Async): Waits for human feedback signaled via an `asyncio.Event`. Pushes "waiting\_for\_review" status to the SSE queue.
+    3.  `ResultNode` (Regular): Marks the task as complete and logs the final result.
+*   **Shared Store (`shared` dict per task):**
+    *   `task_input`: Initial data from user.
+    *   `processed_output`: Result from `ProcessNode`.
+    *   `feedback`: 'approved' or 'rejected' set by the `/feedback` endpoint.
+    *   `review_event`: `asyncio.Event` used by `ReviewNode` to wait and `/feedback` to signal.
+    *   `final_result`: The approved output.
+    *   `current_attempt`: Tracks reprocessing count.
+    *   `task_id`: Unique identifier for the task.
+*   **SSE Communication:** An `asyncio.Queue` (stored alongside the `shared` store in the server's global `tasks` dict, *not directly in PocketFlow's shared store*) is used per task. Nodes (or wrapper code) put status updates onto this queue. The `/stream` endpoint reads from the queue and sends SSE messages.
+*   **Mermaid Diagram:**
+
+```mermaid
+flowchart TD
+    Process[Process Task] -- "default" --> Review{Wait for Feedback}
+    Review -- "approved" --> Result[Final Result]
+    Review -- "rejected" --> Process
+```
+
+## 3. Utilities
+
+For this specific example, the core "utility" is the processing logic itself. Let's simulate it with a simple function. The FastAPI server acts as the external interface.
+
+* `process_task(input_data)`: A placeholder function. In a real scenario, this might call an LLM (`utils/call_llm.py`).
+
+## 4. Node Design (Detailed)
+
+*   **`ProcessNode` (Node):**
+    *   `prep`: Reads `task_input`, `current_attempt` from `shared`.
+    *   `exec`: Calls `utils.process_task.process_task`.
+    *   `post`: Writes `processed_output` to `shared`, increments `current_attempt`. Returns "default".
+*   **`ReviewNode` (AsyncNode):**
+    *   `prep_async`: (As modified/wrapped by server.py) Reads `review_event`, `processed_output` from `shared`. **Puts "waiting\_for\_review" status onto the task's SSE queue.**
+    *   `exec_async`: `await shared["review_event"].wait()`.
+    *   `post_async`: Reads `feedback` from `shared`. Clears the event. Returns "approved" or "rejected". If approved, stores `processed_output` into `final_result`.
+*   **`ResultNode` (Node):**
+    *   `prep`: Reads `final_result` from `shared`.
+    *   `exec`: Prints/logs the final result.
+    *   `post`: Returns `None` (ends flow).

cookbook/pocketflow-fastapi-hitl/flow.py

@@ -0,0 +1,18 @@
+from pocketflow import AsyncFlow
+from nodes import ProcessNode, ReviewNode, ResultNode
+
+def create_feedback_flow():
+    """Creates the minimal feedback workflow."""
+    process_node = ProcessNode()
+    review_node = ReviewNode()
+    result_node = ResultNode()
+
+    # Define transitions
+    process_node >> review_node
+    review_node - "approved" >> result_node
+    review_node - "rejected" >> process_node # Loop back
+
+    # Create the AsyncFlow
+    flow = AsyncFlow(start=process_node)
+    print("Minimal feedback flow created.")
+    return flow

cookbook/pocketflow-fastapi-hitl/main.py

@@ -0,0 +1,16 @@
+from flow import qa_flow
+
+# Example main function
+# Please replace this with your own main function
+def main():
+    shared = {
+        "question": "In one sentence, what's the end of universe?",
+        "answer": None
+    }
+
+    qa_flow.run(shared)
+    print("Question:", shared["question"])
+    print("Answer:", shared["answer"])
+
+if __name__ == "__main__":
+    main()

cookbook/pocketflow-fastapi-hitl/nodes.py

@@ -0,0 +1,78 @@
+from pocketflow import Node, AsyncNode
+from utils.process_task import process_task
+
+class ProcessNode(Node):
+    def prep(self, shared):
+        task_input = shared.get("task_input", "No input")
+        print("ProcessNode Prep")
+        return task_input
+
+    def exec(self, prep_res):
+        return process_task(prep_res)
+
+    def post(self, shared, prep_res, exec_res):
+        shared["processed_output"] = exec_res
+        print("ProcessNode Post: Output stored.")
+        return "default" # Go to ReviewNode
+
+class ReviewNode(AsyncNode):
+    async def prep_async(self, shared):
+        review_event = shared.get("review_event")
+        queue = shared.get("sse_queue") # Expect queue in shared
+        processed_output = shared.get("processed_output", "N/A")
+
+        if not review_event or not queue:
+            print("ERROR: ReviewNode Prep - Missing review_event or sse_queue in shared store!")
+            return None # Signal failure
+
+        # Push status update to SSE queue
+        status_update = {
+            "status": "waiting_for_review",
+            "output_to_review": processed_output
+        }
+        await queue.put(status_update)
+        print("ReviewNode Prep: Put 'waiting_for_review' on SSE queue.")
+
+        return review_event # Return event for exec_async
+
+    async def exec_async(self, prep_res):
+        review_event = prep_res
+        if not review_event:
+            print("ReviewNode Exec: Skipping wait (no event from prep).")
+            return
+        print("ReviewNode Exec: Waiting on review_event...")
+        await review_event.wait()
+        print("ReviewNode Exec: review_event set.")
+
+    async def post_async(self, shared, prep_res, exec_res):
+        feedback = shared.get("feedback")
+        print(f"ReviewNode Post: Processing feedback '{feedback}'")
+
+        # Clear the event for potential loops
+        review_event = shared.get("review_event")
+        if review_event:
+            review_event.clear()
+        shared["feedback"] = None # Reset feedback
+
+        if feedback == "approved":
+            shared["final_result"] = shared.get("processed_output")
+            print("ReviewNode Post: Action=approved")
+            return "approved"
+        else:
+            print("ReviewNode Post: Action=rejected")
+            return "rejected"
+
+class ResultNode(Node):
+     def prep(self, shared):
+         print("ResultNode Prep")
+         return shared.get("final_result", "No final result.")
+
+     def exec(self, prep_res):
+         print(f"--- FINAL RESULT ---")
+         print(prep_res)
+         print(f"--------------------")
+         return prep_res
+
+     def post(self, shared, prep_res, exec_res):
+         print("ResultNode Post: Flow finished.")
+         return None # End flow

cookbook/pocketflow-fastapi-hitl/requirements.txt

@@ -0,0 +1,4 @@
+pocketflow>=0.0.1
+fastapi
+uvicorn[standard] # ASGI server for FastAPI
+jinja2 # For HTML templating

cookbook/pocketflow-fastapi-hitl/server.py

@@ -0,0 +1,253 @@
+import asyncio
+import uuid
+import json
+import os
+from fastapi import FastAPI, Request, HTTPException, status, BackgroundTasks # Import BackgroundTasks
+from fastapi.responses import HTMLResponse, StreamingResponse
+from fastapi.staticfiles import StaticFiles
+from fastapi.templating import Jinja2Templates
+from pydantic import BaseModel, Field # Import Pydantic for request/response models
+from typing import Dict, Any, Literal # For type hinting
+
+from flow import create_feedback_flow # PocketFlow imports
+
+# --- Configuration ---
+app = FastAPI(title="Minimal Feedback Loop API")
+
+static_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), 'static'))
+if os.path.isdir(static_dir):
+    app.mount("/static", StaticFiles(directory=static_dir), name="static")
+else:
+    print(f"Warning: Static directory '{static_dir}' not found.")
+
+template_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), 'templates'))
+if os.path.isdir(template_dir):
+    templates = Jinja2Templates(directory=template_dir)
+else:
+    print(f"Warning: Template directory '{template_dir}' not found.")
+    templates = None
+
+# --- State Management (In-Memory - NOT FOR PRODUCTION) ---
+# Global dictionary to store task state. In production, use Redis, DB, etc.
+tasks: Dict[str, Dict[str, Any]] = {}
+# Structure: task_id -> {"shared": dict, "status": str, "task_obj": asyncio.Task | None}
+
+
+# --- Background Flow Runner ---
+# This function remains mostly the same, as it defines the work to be done.
+# It will be scheduled by FastAPI's BackgroundTasks now.
+async def run_flow_background(task_id: str, flow, shared: Dict[str, Any]):
+    """Runs the flow in background, uses queue in shared for SSE."""
+    # Check if task exists (might have been cancelled/deleted)
+    if task_id not in tasks:
+        print(f"Background task {task_id}: Task not found, aborting.")
+        return
+    queue = shared.get("sse_queue")
+    if not queue:
+        print(f"ERROR: Task {task_id} missing sse_queue in shared store!")
+        tasks[task_id]["status"] = "failed"
+        # Cannot report failure via SSE if queue is missing
+        return
+
+    tasks[task_id]["status"] = "running"
+    await queue.put({"status": "running"})
+    print(f"Task {task_id}: Background flow starting.")
+
+    final_status = "unknown"
+    error_message = None
+    try:
+        # Execute the potentially long-running PocketFlow
+        await flow.run_async(shared)
+
+        # Determine final status based on shared state after flow completion
+        if shared.get("final_result") is not None:
+            final_status = "completed"
+        else:
+            # If flow ends without setting final_result
+            final_status = "finished_incomplete"
+        print(f"Task {task_id}: Flow finished with status: {final_status}")
+
+    except Exception as e:
+        final_status = "failed"
+        error_message = str(e)
+        print(f"Task {task_id}: Flow execution failed: {e}")
+        # Consider logging traceback here in production
+    finally:
+        # Ensure task still exists before updating state
+        if task_id in tasks:
+            tasks[task_id]["status"] = final_status
+            final_update = {"status": final_status}
+            if final_status == "completed":
+                final_update["final_result"] = shared.get("final_result")
+            elif error_message:
+                final_update["error"] = error_message
+            # Put final status update onto the queue
+            await queue.put(final_update)
+
+        # Signal the end of the SSE stream by putting None
+        # Must happen regardless of whether task was deleted mid-run
+        if queue:
+           await queue.put(None)
+        print(f"Task {task_id}: Background task ended. Final update sentinel put on queue.")
+        # Remove the reference to the completed/failed asyncio Task object
+        if task_id in tasks:
+            tasks[task_id]["task_obj"] = None
+
+# --- Pydantic Models for Request/Response Validation ---
+class SubmitRequest(BaseModel):
+    data: str = Field(..., min_length=1, description="Input data for the task")
+
+class SubmitResponse(BaseModel):
+    message: str = "Task submitted"
+    task_id: str
+
+class FeedbackRequest(BaseModel):
+    feedback: Literal["approved", "rejected"] # Use Literal for specific choices
+
+class FeedbackResponse(BaseModel):
+    message: str
+
+# --- FastAPI Routes ---
+@app.get("/", response_class=HTMLResponse, include_in_schema=False)
+async def get_index(request: Request):
+    """Serves the main HTML frontend."""
+    if templates is None:
+        raise HTTPException(status_code=500, detail="Templates directory not configured.")
+    return templates.TemplateResponse("index.html", {"request": request})
+
+@app.post("/submit", response_model=SubmitResponse, status_code=status.HTTP_202_ACCEPTED)
+async def submit_task(
+    submit_request: SubmitRequest, # Use Pydantic model for validation
+    background_tasks: BackgroundTasks # Inject BackgroundTasks instance
+):
+    """
+    Submits a new task. The actual processing runs in the background.
+    Returns immediately with the task ID.
+    """
+    task_id = str(uuid.uuid4())
+    feedback_event = asyncio.Event()
+    status_queue = asyncio.Queue()
+
+    shared = {
+        "task_input": submit_request.data,
+        "processed_output": None,
+        "feedback": None,
+        "review_event": feedback_event,
+        "sse_queue": status_queue,
+        "final_result": None,
+        "task_id": task_id
+    }
+
+    flow = create_feedback_flow()
+
+    # Store task state BEFORE scheduling background task
+    tasks[task_id] = {
+        "shared": shared,
+        "status": "pending",
+        "task_obj": None # Placeholder for the asyncio Task created by BackgroundTasks
+    }
+
+    await status_queue.put({"status": "pending", "task_id": task_id})
+
+    # Schedule the flow execution using FastAPI's BackgroundTasks
+    # This runs AFTER the response has been sent
+    background_tasks.add_task(run_flow_background, task_id, flow, shared)
+    # Note: We don't get a direct reference to the asyncio Task object this way,
+    # which is fine for this minimal example. If cancellation were needed,
+    # managing asyncio.create_task manually would be necessary.
+
+    print(f"Task {task_id}: Submitted, scheduled for background execution.")
+    return SubmitResponse(task_id=task_id)
+
+
+@app.post("/feedback/{task_id}", response_model=FeedbackResponse)
+async def provide_feedback(task_id: str, feedback_request: FeedbackRequest):
+    """Provides feedback (approved/rejected) to potentially unblock a waiting task."""
+    if task_id not in tasks:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Task not found")
+
+    task_info = tasks[task_id]
+    shared = task_info["shared"]
+    queue = shared.get("sse_queue")
+    review_event = shared.get("review_event")
+
+    async def report_error(message, status_code=status.HTTP_400_BAD_REQUEST):
+        # Helper to log, put status on queue, and raise HTTP exception
+        print(f"Task {task_id}: Feedback error - {message}")
+        if queue: await queue.put({"status": "feedback_error", "error": message})
+        raise HTTPException(status_code=status_code, detail=message)
+
+    if not review_event:
+        # This indicates an internal setup error if the task exists but has no event
+        await report_error("Task not configured for feedback", status.HTTP_500_INTERNAL_SERVER_ERROR)
+    if review_event.is_set():
+        # Prevent processing feedback multiple times or if the task isn't waiting
+        await report_error("Task not awaiting feedback or feedback already sent", status.HTTP_409_CONFLICT)
+
+    feedback = feedback_request.feedback # Already validated by Pydantic
+    print(f"Task {task_id}: Received feedback via POST: {feedback}")
+
+    # Update status *before* setting the event, so client sees 'processing' first
+    if queue: await queue.put({"status": "processing_feedback", "feedback_value": feedback})
+    tasks[task_id]["status"] = "processing_feedback" # Update central status tracker
+
+    # Store feedback and signal the waiting ReviewNode
+    shared["feedback"] = feedback
+    review_event.set()
+
+    return FeedbackResponse(message=f"Feedback '{feedback}' received")
+
+
+# --- SSE Endpoint ---
+@app.get("/stream/{task_id}")
+async def stream_status(task_id: str):
+    """Streams status updates for a given task using Server-Sent Events."""
+    if task_id not in tasks or "sse_queue" not in tasks[task_id]["shared"]:
+         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Task or queue not found")
+
+    queue = tasks[task_id]["shared"]["sse_queue"]
+
+    async def event_generator():
+        """Yields SSE messages from the task's queue."""
+        print(f"SSE Stream: Client connected for {task_id}")
+        try:
+            while True:
+                # Wait for the next status update from the queue
+                update = await queue.get()
+                if update is None: # Sentinel value indicates end of stream
+                    print(f"SSE Stream: Sentinel received for {task_id}, closing stream.")
+                    yield f"data: {json.dumps({'status': 'stream_closed'})}\n\n"
+                    break
+
+                sse_data = json.dumps(update)
+                print(f"SSE Stream: Sending for {task_id}: {sse_data}")
+                yield f"data: {sse_data}\n\n" # SSE format: "data: <json>\n\n"
+                queue.task_done() # Acknowledge processing the queue item
+
+        except asyncio.CancelledError:
+            # This happens if the client disconnects
+            print(f"SSE Stream: Client disconnected for {task_id}.")
+        except Exception as e:
+            # Log unexpected errors during streaming
+            print(f"SSE Stream: Error in generator for {task_id}: {e}")
+            # Optionally send an error message to the client if possible
+            try:
+                yield f"data: {json.dumps({'status': 'stream_error', 'error': str(e)})}\n\n"
+            except Exception: # Catch errors if yield fails (e.g., connection already closed)
+                pass
+        finally:
+            print(f"SSE Stream: Generator finished for {task_id}.")
+            # Consider cleanup here (e.g., removing task if no longer needed)
+            # if task_id in tasks: del tasks[task_id]
+
+    # Use FastAPI/Starlette's StreamingResponse for SSE
+    headers = {'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'}
+    return StreamingResponse(event_generator(), media_type="text/event-stream", headers=headers)
+
+# --- Main Execution Guard (for running with uvicorn) ---
+if __name__ == "__main__":
+    print("Starting FastAPI server using Uvicorn is recommended:")
+    print("uvicorn server:app --reload --host 0.0.0.0 --port 8000")
+    # Example using uvicorn programmatically (less common than CLI)
+    # import uvicorn
+    # uvicorn.run(app, host="0.0.0.0", port=8000)

cookbook/pocketflow-fastapi-hitl/static/style.css

@@ -0,0 +1,137 @@
+body {
+    font-family: sans-serif;
+    margin: 0; /* Remove default body margin */
+    padding: 20px; /* Add some padding around the content */
+    background-color: #f8f9fa; /* Lighter grey background */
+    display: flex; /* Enable Flexbox */
+    flex-direction: column; /* Stack children vertically */
+    align-items: center; /* Center children horizontally */
+    min-height: 100vh; /* Ensure body takes at least full viewport height */
+    box-sizing: border-box; /* Include padding in height calculation */
+}
+
+h1 {
+    text-align: center; /* Center the main title */
+    color: #343a40;
+    margin-bottom: 25px;
+}
+
+/* Style the main containers */
+.container, .status-container {
+    background: #ffffff;
+    padding: 20px 25px; /* More padding */
+    border: 1px solid #dee2e6; /* Softer border */
+    margin-bottom: 20px;
+    border-radius: 6px; /* Slightly rounder corners */
+    width: 90%; /* Responsive width */
+    max-width: 650px; /* Max width for readability */
+    box-shadow: 0 2px 5px rgba(0,0,0,0.05); /* Subtle shadow */
+    box-sizing: border-box; /* Include padding/border in width */
+}
+
+textarea {
+    width: 100%; /* Take full width of parent container */
+    padding: 10px;
+    margin-bottom: 10px;
+    border: 1px solid #ced4da;
+    border-radius: 4px;
+    font-size: 1em;
+    min-height: 60px;
+    box-sizing: border-box;
+}
+
+button {
+    padding: 9px 15px; /* Slightly adjusted padding */
+    margin-right: 8px;
+    cursor: pointer;
+    border: none; /* Remove default border */
+    border-radius: 4px;
+    font-weight: 500;
+    transition: background-color 0.2s ease;
+}
+
+button:disabled {
+    cursor: not-allowed;
+    opacity: 0.6;
+}
+
+/* Specific button styling */
+#submit-button {
+    background-color: #0d6efd; /* Bootstrap primary blue */
+    color: white;
+}
+#submit-button:hover:not(:disabled) {
+    background-color: #0b5ed7;
+}
+
+.approve {
+    background-color: #198754; /* Bootstrap success green */
+    color: white;
+}
+.approve:hover:not(:disabled) {
+    background-color: #157347;
+}
+
+.reject {
+    background-color: #dc3545; /* Bootstrap danger red */
+    color: white;
+}
+.reject:hover:not(:disabled) {
+    background-color: #bb2d3b;
+}
+
+
+#task-id-display {
+    font-size: 0.9em;
+    color: #6c757d; /* Bootstrap secondary text color */
+    margin-bottom: 8px;
+    word-wrap: break-word;
+}
+
+#status-display {
+    font-weight: bold;
+    margin-bottom: 15px;
+    padding: 10px;
+    background-color: #e9ecef; /* Light grey background */
+    border: 1px solid #dee2e6;
+    border-radius: 4px;
+    color: #495057;
+}
+
+.hidden {
+    display: none;
+}
+
+/* Review/Result Box Styling */
+.review-box, .result-box {
+    border: 1px solid #dee2e6;
+    padding: 15px;
+    margin-top: 15px;
+    border-radius: 4px;
+    background-color: #f8f9fa; /* Very light background */
+}
+
+h2, h3 {
+    margin-top: 0; /* Remove default top margin */
+    margin-bottom: 15px;
+    color: #495057;
+}
+
+h3 {
+     border-bottom: 1px solid #eee;
+     padding-bottom: 8px;
+}
+
+pre {
+    background-color: #e9ecef;
+    padding: 12px;
+    border: 1px solid #ced4da;
+    border-radius: 4px;
+    white-space: pre-wrap;
+    word-wrap: break-word;
+    max-height: 250px; /* Adjusted height */
+    overflow-y: auto;
+    font-family: monospace;
+    font-size: 0.95em;
+    color: #212529;
+}

cookbook/pocketflow-fastapi-hitl/templates/index.html

@@ -0,0 +1,200 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>Pocket Flow Web Feedback</title>
+    <link rel="stylesheet" href="{{ url_for('static', path='style.css') }}">
+</head>
+<body>
+    <h1>Pocket Flow Web Feedback</h1>
+
+    <div class="container">
+        <textarea id="task-input" rows="3" placeholder="Enter text to process..."></textarea>
+        <button id="submit-button">Submit</button>
+    </div>
+
+    <div class="status-container">
+        <h2>Status</h2>
+        <div id="task-id-display">Task ID: N/A</div>
+        <div id="status-display">Submit a task.</div>
+
+        <div id="review-section" class="hidden review-box">
+            <h3>Review Output</h3>
+            <pre id="review-output"></pre>
+            <button id="approve-button" class="feedback-button approve">Approve</button>
+            <button id="reject-button" class="feedback-button reject">Reject</button>
+        </div>
+
+        <div id="result-section" class="hidden result-box">
+            <h3>Final Result</h3>
+            <pre id="final-result"></pre>
+        </div>
+    </div>
+
+    <script>
+        const taskInput = document.getElementById('task-input');
+        const submitButton = document.getElementById('submit-button');
+        const taskIdDisplay = document.getElementById('task-id-display');
+        const statusDisplay = document.getElementById('status-display');
+        const reviewSection = document.getElementById('review-section');
+        const reviewOutput = document.getElementById('review-output');
+        const approveButton = document.getElementById('approve-button');
+        const rejectButton = document.getElementById('reject-button');
+        const resultSection = document.getElementById('result-section');
+        const finalResult = document.getElementById('final-result');
+
+        let currentTaskId = null;
+        let eventSource = null;
+
+        submitButton.addEventListener('click', handleSubmit);
+        approveButton.addEventListener('click', () => handleFeedback('approved'));
+        rejectButton.addEventListener('click', () => handleFeedback('rejected'));
+
+        async function handleSubmit() {
+            const data = taskInput.value.trim();
+            if (!data) return alert('Input is empty.');
+
+            resetUI();
+            statusDisplay.textContent = 'Submitting...';
+            submitButton.disabled = true;
+
+            try {
+                const response = await fetch('/submit', {
+                    method: 'POST',
+                    headers: { 'Content-Type': 'application/json' },
+                    body: JSON.stringify({ data: data })
+                });
+                if (!response.ok) throw new Error(`Submit failed: ${response.status}`);
+                const result = await response.json();
+                currentTaskId = result.task_id;
+                taskIdDisplay.textContent = `Task ID: ${currentTaskId}`;
+                startSSEListener(currentTaskId);
+            } catch (error) {
+                console.error('Submit error:', error);
+                statusDisplay.textContent = `Submit Error: ${error.message}`;
+                resetUI();
+            } finally {
+                submitButton.disabled = false;
+            }
+        }
+
+        function startSSEListener(taskId) {
+            closeSSEListener(); // Close existing connection
+            eventSource = new EventSource(`/stream/${taskId}`);
+            eventSource.onmessage = handleSSEMessage;
+            eventSource.onerror = handleSSEError;
+            eventSource.onopen = () => console.log(`SSE connected for ${taskId}`);
+        }
+
+        function handleSSEMessage(event) {
+            console.log("SSE data:", event.data);
+            try {
+                const data = JSON.parse(event.data);
+                updateUI(data);
+            } catch (e) { console.error("SSE parse error:", e); }
+        }
+
+        function handleSSEError(error) {
+            console.error("SSE Error:", error);
+            statusDisplay.textContent = "Status stream error. Connection closed.";
+            closeSSEListener();
+        }
+
+        function closeSSEListener() {
+            if (eventSource) {
+                eventSource.close();
+                eventSource = null;
+                console.log("SSE connection closed.");
+            }
+        }
+
+        function updateUI(data) {
+             // Always update main status
+            statusDisplay.textContent = `Status: ${data.status || 'Unknown'}`;
+
+            // Hide sections, then show relevant one
+            reviewSection.classList.add('hidden');
+            resultSection.classList.add('hidden');
+            approveButton.disabled = false; // Re-enable by default
+            rejectButton.disabled = false;
+
+            switch(data.status) {
+                case 'waiting_for_review':
+                    reviewOutput.textContent = data.output_to_review || '';
+                    reviewSection.classList.remove('hidden');
+                    break;
+                case 'processing_feedback':
+                    approveButton.disabled = true; // Disable while processing
+                    rejectButton.disabled = true;
+                    break;
+                case 'completed':
+                    finalResult.textContent = data.final_result || '';
+                    resultSection.classList.remove('hidden');
+                    closeSSEListener();
+                    break;
+                case 'failed':
+                case 'feedback_error':
+                     statusDisplay.textContent = `Status: ${data.status} - ${data.error || 'Unknown error'}`;
+                     closeSSEListener();
+                     break;
+                case 'finished_incomplete':
+                     statusDisplay.textContent = `Status: Flow finished unexpectedly.`;
+                     closeSSEListener();
+                     break;
+                case 'stream_closed':
+                    // Server closed the stream gracefully (usually after completed/failed)
+                    if (!['completed', 'failed', 'finished_incomplete'].includes(tasks[currentTaskId]?.status)) {
+                         statusDisplay.textContent = "Status: Connection closed by server.";
+                    }
+                    closeSSEListener();
+                    break;
+                case 'pending':
+                case 'running':
+                     // Just update status text, wait for next message
+                     break;
+            }
+        }
+
+       async function handleFeedback(feedbackValue) {
+            if (!currentTaskId) return;
+            approveButton.disabled = true;
+            rejectButton.disabled = true;
+            statusDisplay.textContent = `Sending ${feedbackValue}...`; // Optimistic UI update
+
+            try {
+                const response = await fetch(`/feedback/${currentTaskId}`, {
+                    method: 'POST',
+                    headers: { 'Content-Type': 'application/json' },
+                    body: JSON.stringify({ feedback: feedbackValue })
+                });
+                if (!response.ok) { // Rely on SSE for status change or error reporting
+                     const errorData = await response.json().catch(()=>({error: `Feedback failed: ${response.status}`}));
+                     throw new Error(errorData.error);
+                }
+                console.log(`Feedback ${feedbackValue} POST successful.`);
+                // Successful POST - wait for SSE to update status to 'processing', then 'running' etc.
+            } catch (error) {
+                console.error('Feedback error:', error);
+                statusDisplay.textContent = `Feedback Error: ${error.message}`;
+                // Re-enable buttons if feedback POST failed
+                approveButton.disabled = false;
+                rejectButton.disabled = false;
+            }
+        }
+
+        function resetUI() {
+            closeSSEListener();
+            currentTaskId = null;
+            taskIdDisplay.textContent = 'Task ID: N/A';
+            statusDisplay.textContent = 'Submit a task.';
+            reviewSection.classList.add('hidden');
+            resultSection.classList.add('hidden');
+            taskInput.value = '';
+            submitButton.disabled = false;
+            approveButton.disabled = false;
+            rejectButton.disabled = false;
+        }
+    </script>
+</body>
+</html>

cookbook/pocketflow-fastapi-hitl/utils/process_task.py

@@ -0,0 +1,16 @@
+import time
+
+def process_task(input_data):
+    """Minimal simulation of processing the input data."""
+    print(f"Processing: '{input_data[:50]}...'")
+    
+    # Simulate work
+    time.sleep(2)
+
+    processed_result = f"Processed: {input_data}"
+    print(f"Finished processing.")
+    return processed_result
+
+# We don't need a separate utils/call_llm.py for this minimal example,
+# but you would add it here if ProcessNode used an LLM.
+