Upload folder using huggingface_hub

.dockerignore

@@ -0,0 +1,38 @@
+# Docker
+.git
+.gitignore
+venv/
+__pycache__/
+*.pyc
+*.pyo
+*.pyd
+.Python
+env/
+.env
+.venv/
+*.egg-info/
+.eggs/
+dist/
+build/
+
+# IDE
+.idea/
+.vscode/
+*.swp
+*.swo
+
+# Testing artifacts
+.pytest_cache/
+.coverage
+htmlcov/
+.tox/
+.nox/
+
+# Documentation
+*.md
+!README.md
+
+# Misc
+.DS_Store
+Thumbs.db
+*.log

.env.example

@@ -0,0 +1,18 @@
+# Example .env file
+# Copy this to .env and modify as needed
+
+# Application
+APP_NAME="Workflow Engine"
+APP_VERSION="1.0.0"
+DEBUG=true
+
+# Server
+HOST=0.0.0.0
+PORT=8000
+
+# Workflow Engine
+MAX_ITERATIONS=100
+EXECUTION_TIMEOUT=300
+
+# Logging
+LOG_LEVEL=INFO

.gitignore

@@ -0,0 +1,54 @@
+# Environment variables
+.env
+
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# Virtual environments
+venv/
+ENV/
+env/
+.venv/
+
+# IDE
+.idea/
+.vscode/
+*.swp
+*.swo
+*~
+
+# Testing
+.pytest_cache/
+.coverage
+htmlcov/
+.tox/
+.nox/
+
+# Type checking
+.mypy_cache/
+
+# Logs
+*.log
+
+# OS
+.DS_Store
+Thumbs.db

Dockerfile

@@ -0,0 +1,35 @@
+# Use Python 3.11 slim image
+FROM python:3.11-slim
+
+# Set working directory
+WORKDIR /app
+
+# Set environment variables
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PIP_NO_CACHE_DIR=1 \
+    PIP_DISABLE_PIP_VERSION_CHECK=1
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    gcc \
+    && rm -rf /var/lib/apt/lists/*
+
+# Copy requirements first for better caching
+COPY requirements.txt .
+
+# Install Python dependencies
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy application code
+COPY . .
+
+# Expose port (HuggingFace uses 7860 by default)
+EXPOSE 7860
+
+# Health check
+HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
+    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:7860/health')" || exit 1
+
+# Run the application
+CMD uvicorn app.main:app --host 0.0.0.0 --port 7860

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2024
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -1,10 +1,16 @@
 ---
-title: Flowgraph
-emoji: 🌖
-colorFrom: yellow
-colorTo: indigo
+title: FlowGraph
+emoji: 🔄
+colorFrom: blue
+colorTo: purple
 sdk: docker
 pinned: false
+license: mit
+app_port: 7860
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# FlowGraph
+
+A lightweight workflow orchestration engine for building agent pipelines.
+
+Check out the API docs at `/docs`

app/__init__.py

@@ -0,0 +1,9 @@
+"""
+FlowGraph - A lightweight, async-first workflow orchestration engine.
+
+Build agent pipelines with nodes, edges, conditional branching, and looping.
+Similar to LangGraph, but minimal and focused.
+"""
+
+__version__ = "1.0.0"
+__author__ = "AI Engineering Intern"

app/api/__init__.py

@@ -0,0 +1,7 @@
+"""
+API package - FastAPI routes and schemas.
+"""
+
+from app.api.routes import graph, tools, websocket
+
+__all__ = ["graph", "tools", "websocket"]

app/api/routes/__init__.py

@@ -0,0 +1,7 @@
+"""
+API Routes package.
+"""
+
+from app.api.routes import graph, tools, websocket
+
+__all__ = ["graph", "tools", "websocket"]

app/api/routes/graph.py

@@ -0,0 +1,644 @@
+"""
+Graph API Routes.
+
+Endpoints for creating, managing, and executing workflow graphs.
+"""
+
+from typing import Any, Dict, Optional
+from fastapi import APIRouter, HTTPException, BackgroundTasks, status
+from uuid import uuid4
+import logging
+
+from app.api.schemas import (
+    GraphCreateRequest,
+    GraphCreateResponse,
+    GraphRunRequest,
+    GraphRunResponse,
+    GraphInfoResponse,
+    GraphListResponse,
+    RunStateResponse,
+    RunListResponse,
+    ExecutionLogEntry,
+    ExecutionStatus,
+    ErrorResponse,
+)
+from app.engine.graph import Graph, END
+from app.engine.node import Node, get_registered_node
+from app.engine.executor import Executor, ExecutionResult
+from app.storage.memory import graph_storage, run_storage
+from app.tools.registry import tool_registry
+
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/graph", tags=["Graph"])
+
+
+# ============================================================
+# Condition Functions Registry
+# ============================================================
+
+# Built-in condition functions for routing
+_condition_registry: Dict[str, Any] = {}
+
+
+def register_condition(name: str):
+    """Decorator to register a condition function."""
+    def decorator(func):
+        _condition_registry[name] = func
+        return func
+    return decorator
+
+
+@register_condition("quality_check")
+def quality_check_condition(state: Dict[str, Any]) -> str:
+    """Route based on quality score vs threshold."""
+    quality_score = state.get("quality_score", 0)
+    threshold = state.get("quality_threshold", 7.0)
+    return "pass" if quality_score >= threshold else "fail"
+
+
+# Also register as quality_meets_threshold (used by code review workflow)
+@register_condition("quality_meets_threshold")
+def quality_meets_threshold(state: Dict[str, Any]) -> str:
+    """Route based on quality score vs threshold."""
+    quality_score = state.get("quality_score", 0)
+    threshold = state.get("quality_threshold", 7.0)
+    return "pass" if quality_score >= threshold else "fail"
+
+
+@register_condition("always_continue")
+def always_continue(state: Dict[str, Any]) -> str:
+    """Always returns 'continue' - for unconditional looping."""
+    return "continue"
+
+
+# Also register as always_loop (used by code review workflow)
+@register_condition("always_loop")
+def always_loop(state: Dict[str, Any]) -> str:
+    """Always returns 'continue' - for looping back."""
+    return "continue"
+
+
+@register_condition("always_end")
+def always_end(state: Dict[str, Any]) -> str:
+    """Always returns 'end' - for explicit termination."""
+    return "end"
+
+
+@register_condition("max_iterations_check")
+def max_iterations_check(state: Dict[str, Any]) -> str:
+    """Check if max iterations reached."""
+    iteration = state.get("_iteration", 0)
+    max_iter = state.get("_max_iterations", 3)
+    return "stop" if iteration >= max_iter else "continue"
+
+
+def get_condition(name: str):
+    """Get a condition function by name."""
+    return _condition_registry.get(name)
+
+
+# ============================================================
+# Graph CRUD Endpoints
+# ============================================================
+
+@router.post(
+    "/create",
+    response_model=GraphCreateResponse,
+    status_code=status.HTTP_201_CREATED,
+    responses={
+        400: {"model": ErrorResponse, "description": "Invalid graph definition"},
+        404: {"model": ErrorResponse, "description": "Handler not found"},
+    }
+)
+async def create_graph(request: GraphCreateRequest) -> GraphCreateResponse:
+    """
+    Create a new workflow graph.
+    
+    Define nodes with their handlers, edges for flow control,
+    and conditional edges for branching logic.
+    """
+    graph_id = str(uuid4())
+    
+    # Build the graph
+    graph = Graph(
+        graph_id=graph_id,
+        name=request.name,
+        description=request.description or "",
+        max_iterations=request.max_iterations,
+    )
+    
+    # Add nodes
+    for node_def in request.nodes:
+        # Find the handler function
+        handler = get_registered_node(node_def.handler)
+        if handler is None:
+            # Check tool registry as fallback
+            tool = tool_registry.get(node_def.handler)
+            if tool:
+                handler = _create_node_handler_from_tool(node_def.handler)
+            else:
+                raise HTTPException(
+                    status_code=404,
+                    detail=f"Handler '{node_def.handler}' not found. "
+                           f"Available handlers: {list(tool_registry.list_tools())}"
+                )
+        
+        graph.add_node(
+            name=node_def.name,
+            handler=handler,
+            description=node_def.description or "",
+        )
+    
+    # Add direct edges
+    for source, target in request.edges.items():
+        if source not in graph.nodes:
+            raise HTTPException(
+                status_code=400,
+                detail=f"Edge source '{source}' is not a valid node"
+            )
+        if target != END and target != "__END__" and target not in graph.nodes:
+            raise HTTPException(
+                status_code=400,
+                detail=f"Edge target '{target}' is not a valid node"
+            )
+        # Normalize END
+        target = END if target == "__END__" else target
+        graph.add_edge(source, target)
+    
+    # Add conditional edges
+    for source, cond_routes in request.conditional_edges.items():
+        if source not in graph.nodes:
+            raise HTTPException(
+                status_code=400,
+                detail=f"Conditional edge source '{source}' is not a valid node"
+            )
+        
+        # Get condition function
+        condition_func = get_condition(cond_routes.condition)
+        if condition_func is None:
+            raise HTTPException(
+                status_code=404,
+                detail=f"Condition '{cond_routes.condition}' not found. "
+                       f"Available: {list(_condition_registry.keys())}"
+            )
+        
+        # Normalize routes (handle __END__)
+        routes = {}
+        for key, target in cond_routes.routes.items():
+            if target == "__END__":
+                routes[key] = END
+            else:
+                if target not in graph.nodes:
+                    raise HTTPException(
+                        status_code=400,
+                        detail=f"Conditional route target '{target}' is not a valid node"
+                    )
+                routes[key] = target
+        
+        graph.add_conditional_edge(source, condition_func, routes)
+    
+    # Set entry point
+    if request.entry_point:
+        if request.entry_point not in graph.nodes:
+            raise HTTPException(
+                status_code=400,
+                detail=f"Entry point '{request.entry_point}' is not a valid node"
+            )
+        graph.set_entry_point(request.entry_point)
+    
+    # Validate graph
+    errors = graph.validate()
+    if errors:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Graph validation failed: {errors}"
+        )
+    
+    # Store the graph
+    await graph_storage.save(
+        graph_id=graph_id,
+        name=request.name,
+        definition=graph.to_dict(),
+    )
+    
+    logger.info(f"Created graph: {graph_id} ({request.name})")
+    
+    return GraphCreateResponse(
+        graph_id=graph_id,
+        name=request.name,
+        message="Graph created successfully",
+        node_count=len(graph.nodes),
+    )
+
+
+def _create_node_handler_from_tool(tool_name: str):
+    """Create a node handler that calls a tool and updates state."""
+    def handler(state: Dict[str, Any]) -> Dict[str, Any]:
+        tool = tool_registry.get(tool_name)
+        if not tool:
+            raise ValueError(f"Tool '{tool_name}' not found")
+        
+        # Check if the tool function expects a 'state' parameter (node handler style)
+        # or individual parameters (regular tool style)
+        import inspect
+        sig = inspect.signature(tool.func)
+        param_names = list(sig.parameters.keys())
+        
+        if len(param_names) == 1 and param_names[0] == 'state':
+            # This is a node handler - pass state directly
+            result = tool.func(state)
+        else:
+            # This is a regular tool - extract arguments from state
+            result = tool.func(**_extract_tool_args(tool, state))
+        
+        # Handle the result
+        if isinstance(result, dict):
+            # If the tool returns a full state, use it directly
+            # Check if it looks like a state update (has same keys or adds new ones)
+            if result is state:
+                return result
+            # Merge result into state
+            state.update(result)
+        
+        return state
+    
+    handler.__name__ = f"{tool_name}_handler"
+    return handler
+
+
+def _extract_tool_args(tool, state: Dict[str, Any]) -> Dict[str, Any]:
+    """Extract arguments for a tool from state."""
+    import inspect
+    sig = inspect.signature(tool.func)
+    args = {}
+    
+    for param_name, param in sig.parameters.items():
+        if param_name in state:
+            args[param_name] = state[param_name]
+        elif param.default != inspect.Parameter.empty:
+            pass  # Use default
+        # Skip missing optional params
+    
+    return args
+
+
+@router.get(
+    "/{graph_id}",
+    response_model=GraphInfoResponse,
+    responses={404: {"model": ErrorResponse}},
+)
+async def get_graph(graph_id: str) -> GraphInfoResponse:
+    """Get information about a specific graph."""
+    stored = await graph_storage.get(graph_id)
+    if not stored:
+        raise HTTPException(status_code=404, detail=f"Graph '{graph_id}' not found")
+    
+    definition = stored.definition
+    
+    # Generate mermaid diagram
+    mermaid = _generate_mermaid(definition)
+    
+    return GraphInfoResponse(
+        graph_id=stored.graph_id,
+        name=stored.name,
+        description=definition.get("description"),
+        node_count=len(definition.get("nodes", {})),
+        nodes=list(definition.get("nodes", {}).keys()),
+        entry_point=definition.get("entry_point"),
+        max_iterations=definition.get("max_iterations", 100),
+        created_at=stored.created_at.isoformat(),
+        mermaid_diagram=mermaid,
+    )
+
+
+def _generate_mermaid(definition: Dict[str, Any]) -> str:
+    """Generate a Mermaid diagram from graph definition."""
+    lines = ["graph TD"]
+    
+    nodes = definition.get("nodes", {})
+    edges = definition.get("edges", {})
+    cond_edges = definition.get("conditional_edges", {})
+    
+    # Add nodes
+    for name in nodes:
+        label = name.replace("_", " ").title()
+        lines.append(f'    {name}["{label}"]')
+    
+    # Check if END is used
+    has_end = END in edges.values()
+    for cond in cond_edges.values():
+        if END in cond.get("routes", {}).values():
+            has_end = True
+    
+    if has_end:
+        lines.append(f'    {END}(("END"))')
+    
+    # Add direct edges
+    for source, target in edges.items():
+        lines.append(f"    {source} --> {target}")
+    
+    # Add conditional edges
+    for source, cond in cond_edges.items():
+        for route_key, target in cond.get("routes", {}).items():
+            lines.append(f"    {source} -->|{route_key}| {target}")
+    
+    return "\n".join(lines)
+
+
+@router.get(
+    "/",
+    response_model=GraphListResponse,
+)
+async def list_graphs() -> GraphListResponse:
+    """List all available graphs."""
+    graphs = await graph_storage.list_all()
+    
+    graph_infos = []
+    for stored in graphs:
+        definition = stored.definition
+        graph_infos.append(GraphInfoResponse(
+            graph_id=stored.graph_id,
+            name=stored.name,
+            description=definition.get("description"),
+            node_count=len(definition.get("nodes", {})),
+            nodes=list(definition.get("nodes", {}).keys()),
+            entry_point=definition.get("entry_point"),
+            max_iterations=definition.get("max_iterations", 100),
+            created_at=stored.created_at.isoformat(),
+            mermaid_diagram=None,  # Skip for list view
+        ))
+    
+    return GraphListResponse(graphs=graph_infos, total=len(graph_infos))
+
+
+@router.delete(
+    "/{graph_id}",
+    status_code=status.HTTP_204_NO_CONTENT,
+    responses={404: {"model": ErrorResponse}},
+)
+async def delete_graph(graph_id: str):
+    """Delete a graph."""
+    deleted = await graph_storage.delete(graph_id)
+    if not deleted:
+        raise HTTPException(status_code=404, detail=f"Graph '{graph_id}' not found")
+    logger.info(f"Deleted graph: {graph_id}")
+
+
+# ============================================================
+# Execution Endpoints
+# ============================================================
+
+@router.post(
+    "/run",
+    response_model=GraphRunResponse,
+    responses={
+        404: {"model": ErrorResponse},
+        500: {"model": ErrorResponse, "description": "Execution failed"},
+    }
+)
+async def run_graph(
+    request: GraphRunRequest,
+    background_tasks: BackgroundTasks,
+) -> GraphRunResponse:
+    """
+    Execute a workflow graph with the given initial state.
+    
+    If `async_execution` is True, the workflow runs in the background
+    and you can poll the status using GET /graph/state/{run_id}.
+    """
+    # Get the graph
+    stored = await graph_storage.get(request.graph_id)
+    if not stored:
+        raise HTTPException(
+            status_code=404,
+            detail=f"Graph '{request.graph_id}' not found"
+        )
+    
+    # Rebuild the graph from definition
+    graph = await _rebuild_graph_from_definition(stored.definition)
+    
+    # Create run
+    run_id = str(uuid4())
+    await run_storage.create(run_id, request.graph_id, request.initial_state)
+    
+    if request.async_execution:
+        # Run in background
+        background_tasks.add_task(
+            _execute_in_background,
+            graph,
+            run_id,
+            request.initial_state,
+        )
+        
+        return GraphRunResponse(
+            run_id=run_id,
+            graph_id=request.graph_id,
+            status=ExecutionStatus.PENDING,
+            final_state={},
+            execution_log=[],
+            started_at=None,
+            completed_at=None,
+            total_duration_ms=None,
+            iterations=0,
+        )
+    
+    # Execute synchronously
+    try:
+        executor = Executor(
+            graph,
+            run_id=run_id,
+            on_step=lambda step, state: _update_run_state(run_id, step, state),
+        )
+        result = await executor.run(request.initial_state)
+        
+        # Update storage
+        if result.status.value == "completed":
+            await run_storage.complete(
+                run_id,
+                result.final_state,
+                [s.to_dict() for s in result.execution_log],
+            )
+        else:
+            await run_storage.fail(run_id, result.error or "Unknown error", result.final_state)
+        
+        return _result_to_response(result)
+        
+    except Exception as e:
+        logger.exception(f"Execution failed: {e}")
+        await run_storage.fail(run_id, str(e))
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+async def _rebuild_graph_from_definition(definition: Dict[str, Any]) -> Graph:
+    """Rebuild a Graph object from its stored definition."""
+    graph = Graph(
+        graph_id=definition.get("graph_id", str(uuid4())),
+        name=definition.get("name", "Unnamed"),
+        description=definition.get("description", ""),
+        max_iterations=definition.get("max_iterations", 100),
+    )
+    
+    # Add nodes
+    nodes_def = definition.get("nodes", {})
+    for node_name, node_info in nodes_def.items():
+        handler_name = node_info.get("handler", node_name)
+        handler = _create_node_handler_from_tool(handler_name)
+        graph.add_node(
+            name=node_name,
+            handler=handler,
+            description=node_info.get("description", ""),
+        )
+    
+    # Add direct edges
+    for source, target in definition.get("edges", {}).items():
+        graph.add_edge(source, target)
+    
+    # Add conditional edges
+    for source, cond_info in definition.get("conditional_edges", {}).items():
+        condition_name = cond_info.get("condition", "always_continue")
+        condition_func = get_condition(condition_name)
+        if condition_func is None:
+            condition_func = always_continue
+        
+        routes = cond_info.get("routes", {})
+        graph.add_conditional_edge(source, condition_func, routes)
+    
+    # Set entry point
+    if definition.get("entry_point"):
+        graph.set_entry_point(definition["entry_point"])
+    
+    return graph
+
+
+async def _execute_in_background(graph: Graph, run_id: str, initial_state: Dict[str, Any]):
+    """Execute a workflow in the background."""
+    try:
+        executor = Executor(
+            graph,
+            run_id=run_id,
+            on_step=lambda step, state: _update_run_state(run_id, step, state),
+        )
+        result = await executor.run(initial_state)
+        
+        if result.status.value == "completed":
+            await run_storage.complete(
+                run_id,
+                result.final_state,
+                [s.to_dict() for s in result.execution_log],
+            )
+        else:
+            await run_storage.fail(run_id, result.error or "Unknown error", result.final_state)
+            
+    except Exception as e:
+        logger.exception(f"Background execution failed: {e}")
+        await run_storage.fail(run_id, str(e))
+
+
+def _update_run_state(run_id: str, step, state: Dict[str, Any]):
+    """Update run state during execution (sync callback)."""
+    import asyncio
+    try:
+        loop = asyncio.get_event_loop()
+        if loop.is_running():
+            asyncio.create_task(
+                run_storage.update_state(run_id, state, step.node, step.iteration)
+            )
+    except Exception:
+        pass  # Ignore errors in callback
+
+
+def _result_to_response(result: ExecutionResult) -> GraphRunResponse:
+    """Convert ExecutionResult to API response."""
+    return GraphRunResponse(
+        run_id=result.run_id,
+        graph_id=result.graph_id,
+        status=ExecutionStatus(result.status.value),
+        final_state=result.final_state,
+        execution_log=[
+            ExecutionLogEntry(
+                step=s.step,
+                node=s.node,
+                started_at=s.started_at.isoformat(),
+                completed_at=s.completed_at.isoformat() if s.completed_at else None,
+                duration_ms=s.duration_ms,
+                iteration=s.iteration,
+                result=s.result,
+                error=s.error,
+                route_taken=s.route_taken,
+            )
+            for s in result.execution_log
+        ],
+        started_at=result.started_at.isoformat() if result.started_at else None,
+        completed_at=result.completed_at.isoformat() if result.completed_at else None,
+        total_duration_ms=result.total_duration_ms,
+        iterations=result.iterations,
+        error=result.error,
+    )
+
+
+# ============================================================
+# Run State Endpoints
+# ============================================================
+
+@router.get(
+    "/state/{run_id}",
+    response_model=RunStateResponse,
+    responses={404: {"model": ErrorResponse}},
+)
+async def get_run_state(run_id: str) -> RunStateResponse:
+    """
+    Get the current state of a workflow run.
+    
+    Use this to poll the status of async executions.
+    """
+    stored = await run_storage.get(run_id)
+    if not stored:
+        raise HTTPException(status_code=404, detail=f"Run '{run_id}' not found")
+    
+    return RunStateResponse(
+        run_id=stored.run_id,
+        graph_id=stored.graph_id,
+        status=ExecutionStatus(stored.status),
+        current_node=stored.current_node,
+        current_state=stored.current_state,
+        iteration=stored.iteration,
+        execution_log=[
+            ExecutionLogEntry(**entry) for entry in stored.execution_log
+        ],
+        started_at=stored.started_at.isoformat(),
+        completed_at=stored.completed_at.isoformat() if stored.completed_at else None,
+        error=stored.error,
+    )
+
+
+@router.get(
+    "/runs",
+    response_model=RunListResponse,
+)
+async def list_runs(graph_id: Optional[str] = None) -> RunListResponse:
+    """List all runs, optionally filtered by graph_id."""
+    if graph_id:
+        runs = await run_storage.list_by_graph(graph_id)
+    else:
+        runs = await run_storage.list_all()
+    
+    run_states = []
+    for stored in runs:
+        run_states.append(RunStateResponse(
+            run_id=stored.run_id,
+            graph_id=stored.graph_id,
+            status=ExecutionStatus(stored.status),
+            current_node=stored.current_node,
+            current_state=stored.current_state,
+            iteration=stored.iteration,
+            execution_log=[
+                ExecutionLogEntry(**entry) for entry in stored.execution_log
+            ],
+            started_at=stored.started_at.isoformat(),
+            completed_at=stored.completed_at.isoformat() if stored.completed_at else None,
+            error=stored.error,
+        ))
+    
+    return RunListResponse(runs=run_states, total=len(run_states))

app/api/routes/tools.py

@@ -0,0 +1,175 @@
+"""
+Tools API Routes.
+
+Endpoints for listing and managing registered tools.
+"""
+
+from typing import Any, Dict
+from fastapi import APIRouter, HTTPException, status
+import logging
+
+from app.api.schemas import (
+    ToolInfo,
+    ToolListResponse,
+    ToolRegisterRequest,
+    ToolRegisterResponse,
+    ErrorResponse,
+)
+from app.tools.registry import tool_registry
+
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/tools", tags=["Tools"])
+
+
+@router.get(
+    "/",
+    response_model=ToolListResponse,
+)
+async def list_tools() -> ToolListResponse:
+    """
+    List all registered tools.
+    
+    Tools are functions that workflow nodes can use during execution.
+    """
+    tools = tool_registry.list_tools()
+    
+    tool_infos = [
+        ToolInfo(
+            name=t["name"],
+            description=t["description"],
+            parameters=t["parameters"],
+        )
+        for t in tools
+    ]
+    
+    return ToolListResponse(tools=tool_infos, total=len(tool_infos))
+
+
+@router.get(
+    "/{tool_name}",
+    response_model=ToolInfo,
+    responses={404: {"model": ErrorResponse}},
+)
+async def get_tool(tool_name: str) -> ToolInfo:
+    """Get information about a specific tool."""
+    tool = tool_registry.get(tool_name)
+    if not tool:
+        raise HTTPException(
+            status_code=404,
+            detail=f"Tool '{tool_name}' not found"
+        )
+    
+    return ToolInfo(
+        name=tool.name,
+        description=tool.description,
+        parameters=tool.parameters,
+    )
+
+
+@router.post(
+    "/register",
+    response_model=ToolRegisterResponse,
+    status_code=status.HTTP_201_CREATED,
+    responses={
+        400: {"model": ErrorResponse, "description": "Invalid tool code"},
+        409: {"model": ErrorResponse, "description": "Tool already exists"},
+    }
+)
+async def register_tool(request: ToolRegisterRequest) -> ToolRegisterResponse:
+    """
+    Register a new tool dynamically.
+    
+    **Warning**: This endpoint executes Python code. Use with caution
+    and only in trusted environments.
+    
+    The code should define a function that:
+    - Takes parameters as needed
+    - Returns a dictionary with results
+    """
+    # Check if tool already exists
+    if tool_registry.has(request.name):
+        raise HTTPException(
+            status_code=409,
+            detail=f"Tool '{request.name}' already exists"
+        )
+    
+    # Try to compile and execute the code
+    try:
+        # Create a restricted namespace
+        namespace: Dict[str, Any] = {}
+        
+        # Execute the code to define the function
+        exec(request.code, namespace)
+        
+        # Find the function in the namespace
+        func = None
+        for name, value in namespace.items():
+            if callable(value) and not name.startswith("_"):
+                func = value
+                break
+        
+        if func is None:
+            raise HTTPException(
+                status_code=400,
+                detail="No callable function found in the provided code"
+            )
+        
+        # Register the tool
+        tool_registry.add(
+            func=func,
+            name=request.name,
+            description=request.description,
+        )
+        
+        logger.info(f"Registered dynamic tool: {request.name}")
+        
+        return ToolRegisterResponse(
+            name=request.name,
+            message=f"Tool '{request.name}' registered successfully",
+            warning="Dynamic tool registration executes code. Use responsibly.",
+        )
+        
+    except SyntaxError as e:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Syntax error in tool code: {e}"
+        )
+    except Exception as e:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Error registering tool: {e}"
+        )
+
+
+@router.delete(
+    "/{tool_name}",
+    status_code=status.HTTP_204_NO_CONTENT,
+    responses={404: {"model": ErrorResponse}},
+)
+async def delete_tool(tool_name: str):
+    """Delete a registered tool."""
+    # Protect built-in tools
+    builtin_tools = {
+        "extract_functions",
+        "calculate_complexity", 
+        "detect_issues",
+        "suggest_improvements",
+        "quality_check",
+    }
+    
+    if tool_name in builtin_tools:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Cannot delete built-in tool '{tool_name}'"
+        )
+    
+    deleted = tool_registry.remove(tool_name)
+    if not deleted:
+        raise HTTPException(
+            status_code=404,
+            detail=f"Tool '{tool_name}' not found"
+        )
+    
+    logger.info(f"Deleted tool: {tool_name}")

app/api/routes/websocket.py

@@ -0,0 +1,269 @@
+"""
+WebSocket Routes for Real-time Execution Streaming.
+
+Provides live updates during workflow execution.
+"""
+
+from typing import Any, Dict, Set
+from fastapi import APIRouter, WebSocket, WebSocketDisconnect
+from uuid import uuid4
+import asyncio
+import json
+import logging
+
+from app.engine.graph import Graph
+from app.engine.executor import Executor, ExecutionStep
+from app.storage.memory import graph_storage, run_storage
+
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(tags=["WebSocket"])
+
+
+class ConnectionManager:
+    """Manages WebSocket connections."""
+    
+    def __init__(self):
+        self.active_connections: Dict[str, Set[WebSocket]] = {}
+    
+    async def connect(self, websocket: WebSocket, run_id: str):
+        """Accept a new WebSocket connection."""
+        await websocket.accept()
+        if run_id not in self.active_connections:
+            self.active_connections[run_id] = set()
+        self.active_connections[run_id].add(websocket)
+        logger.info(f"WebSocket connected for run: {run_id}")
+    
+    def disconnect(self, websocket: WebSocket, run_id: str):
+        """Remove a WebSocket connection."""
+        if run_id in self.active_connections:
+            self.active_connections[run_id].discard(websocket)
+            if not self.active_connections[run_id]:
+                del self.active_connections[run_id]
+        logger.info(f"WebSocket disconnected for run: {run_id}")
+    
+    async def broadcast(self, run_id: str, message: Dict[str, Any]):
+        """Broadcast a message to all connections for a run."""
+        if run_id in self.active_connections:
+            disconnected = set()
+            for websocket in self.active_connections[run_id]:
+                try:
+                    await websocket.send_json(message)
+                except Exception:
+                    disconnected.add(websocket)
+            
+            # Clean up disconnected clients
+            for ws in disconnected:
+                self.active_connections[run_id].discard(ws)
+
+
+# Global connection manager
+manager = ConnectionManager()
+
+
+@router.websocket("/ws/run/{graph_id}")
+async def websocket_run(websocket: WebSocket, graph_id: str):
+    """
+    WebSocket endpoint for real-time workflow execution.
+    
+    Connect to this endpoint and send the initial state as JSON.
+    You'll receive step-by-step updates as the workflow executes.
+    
+    Message format (client -> server):
+    ```json
+    {"action": "start", "initial_state": {"code": "..."}}
+    ```
+    
+    Message format (server -> client):
+    ```json
+    {
+        "type": "step",
+        "step": 1,
+        "node": "extract",
+        "status": "completed",
+        "duration_ms": 15.5,
+        "state": {...}
+    }
+    ```
+    """
+    # Check if graph exists
+    stored = await graph_storage.get(graph_id)
+    if not stored:
+        await websocket.close(code=4004, reason=f"Graph '{graph_id}' not found")
+        return
+    
+    run_id = str(uuid4())
+    await manager.connect(websocket, run_id)
+    
+    try:
+        # Wait for start message
+        data = await websocket.receive_json()
+        
+        if data.get("action") != "start":
+            await websocket.send_json({
+                "type": "error",
+                "error": "Expected 'start' action"
+            })
+            return
+        
+        initial_state = data.get("initial_state", {})
+        
+        # Send acknowledgment
+        await websocket.send_json({
+            "type": "started",
+            "run_id": run_id,
+            "graph_id": graph_id,
+        })
+        
+        # Rebuild graph
+        graph = await _rebuild_graph(stored.definition)
+        
+        # Create run record
+        await run_storage.create(run_id, graph_id, initial_state)
+        
+        # Execute with streaming updates
+        async def on_step(step: ExecutionStep, state: Dict[str, Any]):
+            await manager.broadcast(run_id, {
+                "type": "step",
+                "step": step.step,
+                "node": step.node,
+                "status": step.result,
+                "duration_ms": step.duration_ms,
+                "iteration": step.iteration,
+                "route_taken": step.route_taken,
+                "error": step.error,
+                "state": state,
+            })
+        
+        executor = Executor(graph, run_id=run_id)
+        
+        # Run with step notifications
+        result = await _run_with_streaming(executor, initial_state, on_step)
+        
+        # Send completion
+        await websocket.send_json({
+            "type": "completed",
+            "run_id": run_id,
+            "status": result.status.value,
+            "final_state": result.final_state,
+            "total_duration_ms": result.total_duration_ms,
+            "iterations": result.iterations,
+            "error": result.error,
+        })
+        
+        # Update storage
+        if result.status.value == "completed":
+            await run_storage.complete(
+                run_id,
+                result.final_state,
+                [s.to_dict() for s in result.execution_log],
+            )
+        else:
+            await run_storage.fail(run_id, result.error or "Unknown error")
+        
+    except WebSocketDisconnect:
+        logger.info(f"Client disconnected from run {run_id}")
+    except Exception as e:
+        logger.exception(f"WebSocket error: {e}")
+        try:
+            await websocket.send_json({
+                "type": "error",
+                "error": str(e),
+            })
+        except Exception:
+            pass
+    finally:
+        manager.disconnect(websocket, run_id)
+
+
+async def _rebuild_graph(definition: Dict[str, Any]) -> Graph:
+    """Rebuild graph from definition (copied from graph.py to avoid circular import)."""
+    from app.api.routes.graph import _rebuild_graph_from_definition
+    return await _rebuild_graph_from_definition(definition)
+
+
+async def _run_with_streaming(
+    executor: Executor,
+    initial_state: Dict[str, Any],
+    on_step
+):
+    """Run executor with async step callbacks."""
+    from app.engine.graph import END
+    from app.engine.state import StateManager
+    import time
+    from datetime import datetime
+    
+    # Execute the workflow
+    result = await executor.run(initial_state)
+    
+    # Stream each step (already executed, but we notify)
+    for step in result.execution_log:
+        await on_step(step, result.final_state)
+        await asyncio.sleep(0.01)  # Small delay for streaming effect
+    
+    return result
+
+
+@router.websocket("/ws/subscribe/{run_id}")
+async def websocket_subscribe(websocket: WebSocket, run_id: str):
+    """
+    Subscribe to updates for an existing run.
+    
+    Use this to watch an async execution started via POST /graph/run.
+    """
+    # Check if run exists
+    stored = await run_storage.get(run_id)
+    if not stored:
+        await websocket.close(code=4004, reason=f"Run '{run_id}' not found")
+        return
+    
+    await manager.connect(websocket, run_id)
+    
+    try:
+        # Send current state
+        await websocket.send_json({
+            "type": "current_state",
+            "run_id": run_id,
+            "status": stored.status,
+            "current_node": stored.current_node,
+            "iteration": stored.iteration,
+            "state": stored.current_state,
+        })
+        
+        # Keep connection open and poll for updates
+        last_log_count = len(stored.execution_log)
+        
+        while True:
+            await asyncio.sleep(0.5)  # Poll interval
+            
+            stored = await run_storage.get(run_id)
+            if not stored:
+                break
+            
+            # Send new log entries
+            if len(stored.execution_log) > last_log_count:
+                for entry in stored.execution_log[last_log_count:]:
+                    await websocket.send_json({
+                        "type": "step",
+                        **entry,
+                    })
+                last_log_count = len(stored.execution_log)
+            
+            # Check if completed
+            if stored.status in ("completed", "failed", "cancelled"):
+                await websocket.send_json({
+                    "type": "completed",
+                    "run_id": run_id,
+                    "status": stored.status,
+                    "final_state": stored.final_state,
+                    "error": stored.error,
+                })
+                break
+                
+    except WebSocketDisconnect:
+        logger.info(f"Subscriber disconnected from run {run_id}")
+    except Exception as e:
+        logger.exception(f"WebSocket error: {e}")
+    finally:
+        manager.disconnect(websocket, run_id)

app/api/schemas.py

@@ -0,0 +1,322 @@
+"""
+Pydantic Schemas for API Request/Response Models.
+
+These schemas define the structure of data flowing through the API,
+providing automatic validation and documentation.
+"""
+
+from typing import Any, Dict, List, Optional, Union
+from pydantic import BaseModel, Field
+from datetime import datetime
+from enum import Enum
+
+
+# ============================================================
+# Enums
+# ============================================================
+
+class ExecutionStatus(str, Enum):
+    """Status of a workflow execution."""
+    PENDING = "pending"
+    RUNNING = "running"
+    COMPLETED = "completed"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+
+
+# ============================================================
+# Node Schemas
+# ============================================================
+
+class NodeDefinition(BaseModel):
+    """Definition of a node in the graph."""
+    name: str = Field(..., description="Unique name for the node")
+    handler: str = Field(..., description="Name of the handler function (must be registered)")
+    description: Optional[str] = Field(None, description="Human-readable description")
+    
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "name": "extract",
+                "handler": "extract_functions",
+                "description": "Extract function definitions from code"
+            }
+        }
+
+
+# ============================================================
+# Edge Schemas
+# ============================================================
+
+class ConditionalRoutes(BaseModel):
+    """Routes for a conditional edge."""
+    condition: str = Field(..., description="Name of the condition function")
+    routes: Dict[str, str] = Field(
+        ..., 
+        description="Mapping of condition results to target nodes"
+    )
+    
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "condition": "quality_check",
+                "routes": {
+                    "pass": "__END__",
+                    "fail": "improve"
+                }
+            }
+        }
+
+
+# ============================================================
+# Graph Schemas
+# ============================================================
+
+class GraphCreateRequest(BaseModel):
+    """Request to create a new workflow graph."""
+    name: str = Field(..., description="Name of the workflow")
+    description: Optional[str] = Field(None, description="Description of what this workflow does")
+    nodes: List[NodeDefinition] = Field(..., description="List of nodes in the graph")
+    edges: Dict[str, str] = Field(
+        default_factory=dict, 
+        description="Direct edges: source -> target"
+    )
+    conditional_edges: Dict[str, ConditionalRoutes] = Field(
+        default_factory=dict,
+        description="Conditional edges with routing logic"
+    )
+    entry_point: Optional[str] = Field(None, description="Entry node (defaults to first node)")
+    max_iterations: int = Field(100, description="Maximum loop iterations", ge=1, le=1000)
+    
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "name": "code_review_workflow",
+                "description": "Automated code review with quality checks",
+                "nodes": [
+                    {"name": "extract", "handler": "extract_functions"},
+                    {"name": "complexity", "handler": "calculate_complexity"},
+                    {"name": "issues", "handler": "detect_issues"},
+                    {"name": "improve", "handler": "suggest_improvements"}
+                ],
+                "edges": {
+                    "extract": "complexity",
+                    "complexity": "issues"
+                },
+                "conditional_edges": {
+                    "issues": {
+                        "condition": "quality_check",
+                        "routes": {"pass": "__END__", "fail": "improve"}
+                    },
+                    "improve": {
+                        "condition": "always_continue",
+                        "routes": {"continue": "issues"}
+                    }
+                },
+                "entry_point": "extract",
+                "max_iterations": 10
+            }
+        }
+
+
+class GraphCreateResponse(BaseModel):
+    """Response after creating a graph."""
+    graph_id: str = Field(..., description="Unique identifier for the created graph")
+    name: str = Field(..., description="Name of the workflow")
+    message: str = Field(default="Graph created successfully")
+    node_count: int = Field(..., description="Number of nodes in the graph")
+    
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "graph_id": "abc123-def456",
+                "name": "code_review_workflow",
+                "message": "Graph created successfully",
+                "node_count": 4
+            }
+        }
+
+
+class GraphInfoResponse(BaseModel):
+    """Response with graph information."""
+    graph_id: str
+    name: str
+    description: Optional[str]
+    node_count: int
+    nodes: List[str]
+    entry_point: Optional[str]
+    max_iterations: int
+    created_at: str
+    mermaid_diagram: Optional[str] = Field(None, description="Mermaid diagram of the graph")
+
+
+class GraphListResponse(BaseModel):
+    """Response listing all graphs."""
+    graphs: List[GraphInfoResponse]
+    total: int
+
+
+# ============================================================
+# Run Schemas
+# ============================================================
+
+class GraphRunRequest(BaseModel):
+    """Request to run a workflow graph."""
+    graph_id: str = Field(..., description="ID of the graph to run")
+    initial_state: Dict[str, Any] = Field(
+        ..., 
+        description="Initial state data for the workflow"
+    )
+    async_execution: bool = Field(
+        False, 
+        description="If true, run in background and return immediately"
+    )
+    
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "graph_id": "abc123-def456",
+                "initial_state": {
+                    "code": "def hello():\n    print('world')",
+                    "quality_threshold": 7.0
+                },
+                "async_execution": False
+            }
+        }
+
+
+class ExecutionLogEntry(BaseModel):
+    """A single entry in the execution log."""
+    step: int
+    node: str
+    started_at: str
+    completed_at: Optional[str]
+    duration_ms: Optional[float]
+    iteration: int
+    result: str
+    error: Optional[str]
+    route_taken: Optional[str]
+
+
+class GraphRunResponse(BaseModel):
+    """Response after running a graph."""
+    run_id: str = Field(..., description="Unique identifier for this run")
+    graph_id: str
+    status: ExecutionStatus
+    final_state: Dict[str, Any]
+    execution_log: List[ExecutionLogEntry]
+    started_at: Optional[str]
+    completed_at: Optional[str]
+    total_duration_ms: Optional[float]
+    iterations: int
+    error: Optional[str] = None
+    
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "run_id": "run-xyz789",
+                "graph_id": "abc123-def456",
+                "status": "completed",
+                "final_state": {
+                    "code": "def hello():\n    print('world')",
+                    "functions": [{"name": "hello"}],
+                    "quality_score": 8.5
+                },
+                "execution_log": [
+                    {
+                        "step": 1,
+                        "node": "extract",
+                        "started_at": "2024-01-01T12:00:00",
+                        "completed_at": "2024-01-01T12:00:01",
+                        "duration_ms": 15.5,
+                        "iteration": 0,
+                        "result": "success",
+                        "error": None,
+                        "route_taken": None
+                    }
+                ],
+                "started_at": "2024-01-01T12:00:00",
+                "completed_at": "2024-01-01T12:00:05",
+                "total_duration_ms": 5000.0,
+                "iterations": 1,
+                "error": None
+            }
+        }
+
+
+class RunStateResponse(BaseModel):
+    """Response with current run state."""
+    run_id: str
+    graph_id: str
+    status: ExecutionStatus
+    current_node: Optional[str]
+    current_state: Dict[str, Any]
+    iteration: int
+    execution_log: List[ExecutionLogEntry]
+    started_at: str
+    completed_at: Optional[str]
+    error: Optional[str]
+
+
+class RunListResponse(BaseModel):
+    """Response listing runs."""
+    runs: List[RunStateResponse]
+    total: int
+
+
+# ============================================================
+# Tool Schemas
+# ============================================================
+
+class ToolInfo(BaseModel):
+    """Information about a registered tool."""
+    name: str
+    description: str
+    parameters: Dict[str, str]
+
+
+class ToolListResponse(BaseModel):
+    """Response listing all registered tools."""
+    tools: List[ToolInfo]
+    total: int
+
+
+class ToolRegisterRequest(BaseModel):
+    """Request to register a new tool (for dynamic registration)."""
+    name: str = Field(..., description="Unique name for the tool")
+    description: str = Field("", description="Description of what the tool does")
+    code: str = Field(..., description="Python code for the tool function")
+    
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "name": "custom_validator",
+                "description": "Custom validation logic",
+                "code": "def custom_validator(data):\n    return {'valid': True}"
+            }
+        }
+
+
+class ToolRegisterResponse(BaseModel):
+    """Response after registering a tool."""
+    name: str
+    message: str
+    warning: Optional[str] = None
+
+
+# ============================================================
+# Error Schemas
+# ============================================================
+
+class ErrorResponse(BaseModel):
+    """Standard error response."""
+    error: str
+    detail: Optional[str] = None
+    status_code: int
+
+
+class ValidationErrorResponse(BaseModel):
+    """Validation error response."""
+    error: str = "Validation Error"
+    detail: List[Dict[str, Any]]
+    status_code: int = 422

app/config.py

@@ -0,0 +1,34 @@
+"""
+Configuration settings for the Workflow Engine.
+"""
+
+from pydantic_settings import BaseSettings
+from typing import Optional
+
+
+class Settings(BaseSettings):
+    """Application settings with environment variable support."""
+    
+    # Application
+    APP_NAME: str = "FlowGraph"
+    APP_VERSION: str = "1.0.0"
+    DEBUG: bool = True
+    
+    # Server
+    HOST: str = "0.0.0.0"
+    PORT: int = 8000
+    
+    # Workflow Engine
+    MAX_ITERATIONS: int = 100  # Default max loop iterations
+    EXECUTION_TIMEOUT: int = 300  # Seconds
+    
+    # Logging
+    LOG_LEVEL: str = "INFO"
+    
+    class Config:
+        env_file = ".env"
+        case_sensitive = True
+
+
+# Global settings instance
+settings = Settings()

app/engine/__init__.py

@@ -0,0 +1,18 @@
+"""
+Engine package - Core workflow orchestration components.
+"""
+
+from app.engine.state import WorkflowState, StateManager
+from app.engine.node import Node, node
+from app.engine.graph import Graph
+from app.engine.executor import Executor, ExecutionResult
+
+__all__ = [
+    "WorkflowState",
+    "StateManager", 
+    "Node",
+    "node",
+    "Graph",
+    "Executor",
+    "ExecutionResult",
+]

app/engine/executor.py

@@ -0,0 +1,362 @@
+"""
+Async Workflow Executor.
+
+The executor runs a workflow graph, managing state transitions,
+handling loops, and generating execution logs.
+"""
+
+from typing import Any, Callable, Dict, List, Optional
+from dataclasses import dataclass, field
+from datetime import datetime
+from enum import Enum
+import asyncio
+import uuid
+import time
+import logging
+
+from app.engine.graph import Graph, END
+from app.engine.state import WorkflowState, StateManager
+
+
+# Configure logging
+logger = logging.getLogger(__name__)
+
+
+class ExecutionStatus(str, Enum):
+    """Status of a workflow execution."""
+    PENDING = "pending"
+    RUNNING = "running"
+    COMPLETED = "completed"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+
+
+@dataclass
+class ExecutionStep:
+    """A single step in the execution log."""
+    step: int
+    node: str
+    started_at: datetime
+    completed_at: Optional[datetime] = None
+    duration_ms: Optional[float] = None
+    iteration: int = 0
+    result: str = "success"
+    error: Optional[str] = None
+    route_taken: Optional[str] = None
+    
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "step": self.step,
+            "node": self.node,
+            "started_at": self.started_at.isoformat(),
+            "completed_at": self.completed_at.isoformat() if self.completed_at else None,
+            "duration_ms": self.duration_ms,
+            "iteration": self.iteration,
+            "result": self.result,
+            "error": self.error,
+            "route_taken": self.route_taken,
+        }
+
+
+@dataclass
+class ExecutionResult:
+    """Result of a workflow execution."""
+    run_id: str
+    graph_id: str
+    status: ExecutionStatus
+    final_state: Dict[str, Any]
+    execution_log: List[ExecutionStep] = field(default_factory=list)
+    started_at: Optional[datetime] = None
+    completed_at: Optional[datetime] = None
+    total_duration_ms: Optional[float] = None
+    error: Optional[str] = None
+    iterations: int = 0
+    
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "run_id": self.run_id,
+            "graph_id": self.graph_id,
+            "status": self.status.value,
+            "final_state": self.final_state,
+            "execution_log": [step.to_dict() for step in self.execution_log],
+            "started_at": self.started_at.isoformat() if self.started_at else None,
+            "completed_at": self.completed_at.isoformat() if self.completed_at else None,
+            "total_duration_ms": self.total_duration_ms,
+            "error": self.error,
+            "iterations": self.iterations,
+        }
+
+
+class Executor:
+    """
+    Async workflow executor.
+    
+    Executes a graph with given initial state, handling:
+    - Sequential node execution
+    - Conditional branching
+    - Loop iterations with max limit
+    - Detailed execution logging
+    - Error handling
+    
+    Usage:
+        executor = Executor(graph)
+        result = await executor.run({"input": "data"})
+    """
+    
+    def __init__(
+        self,
+        graph: Graph,
+        run_id: Optional[str] = None,
+        on_step: Optional[Callable[[ExecutionStep, Dict[str, Any]], None]] = None
+    ):
+        """
+        Initialize the executor.
+        
+        Args:
+            graph: The workflow graph to execute
+            run_id: Optional run ID (generated if not provided)
+            on_step: Optional callback for each step (for WebSocket streaming)
+        """
+        self.graph = graph
+        self.run_id = run_id or str(uuid.uuid4())
+        self.on_step = on_step
+        
+        # Execution state
+        self._state_manager: Optional[StateManager] = None
+        self._execution_log: List[ExecutionStep] = []
+        self._step_counter = 0
+        self._status = ExecutionStatus.PENDING
+        self._cancelled = False
+    
+    @property
+    def status(self) -> ExecutionStatus:
+        """Get the current execution status."""
+        return self._status
+    
+    @property
+    def current_state(self) -> Optional[Dict[str, Any]]:
+        """Get the current state data."""
+        if self._state_manager and self._state_manager.current_state:
+            return self._state_manager.current_state.data
+        return None
+    
+    @property
+    def current_node(self) -> Optional[str]:
+        """Get the current node being executed."""
+        if self._state_manager and self._state_manager.current_state:
+            return self._state_manager.current_state.current_node
+        return None
+    
+    def cancel(self) -> None:
+        """Cancel the execution."""
+        self._cancelled = True
+        self._status = ExecutionStatus.CANCELLED
+    
+    async def run(self, initial_state: Dict[str, Any]) -> ExecutionResult:
+        """
+        Execute the workflow with the given initial state.
+        
+        Args:
+            initial_state: Initial state data
+            
+        Returns:
+            ExecutionResult with final state and logs
+        """
+        start_time = time.time()
+        self._status = ExecutionStatus.RUNNING
+        self._state_manager = StateManager(self.run_id)
+        
+        # Initialize state
+        state = self._state_manager.initialize(initial_state)
+        
+        # Validate graph
+        errors = self.graph.validate()
+        if errors:
+            return self._create_error_result(
+                f"Graph validation failed: {errors}",
+                start_time
+            )
+        
+        current_node = self.graph.entry_point
+        iteration = 0
+        visited_in_iteration: set = set()
+        
+        try:
+            while current_node and current_node != END:
+                # Check cancellation
+                if self._cancelled:
+                    logger.info(f"Execution cancelled at node '{current_node}'")
+                    break
+                
+                # Check max iterations
+                if iteration >= self.graph.max_iterations:
+                    return self._create_error_result(
+                        f"Max iterations ({self.graph.max_iterations}) exceeded",
+                        start_time
+                    )
+                
+                # Get the node
+                node = self.graph.nodes.get(current_node)
+                if not node:
+                    return self._create_error_result(
+                        f"Node '{current_node}' not found in graph",
+                        start_time
+                    )
+                
+                # Execute the node
+                step = await self._execute_node(node, state, iteration)
+                
+                # Handle error
+                if step.result == "error":
+                    return self._create_error_result(
+                        step.error or "Unknown error",
+                        start_time
+                    )
+                
+                # Update state from state manager
+                state = self._state_manager.current_state
+                
+                # Get next node
+                next_node = self.graph.get_next_node(current_node, state.data)
+                
+                # Track route for conditional edges
+                if current_node in self.graph.conditional_edges:
+                    cond_edge = self.graph.conditional_edges[current_node]
+                    route_key = cond_edge.condition(state.data)
+                    step.route_taken = route_key
+                    logger.debug(f"Conditional route: {route_key} -> {next_node}")
+                
+                # Detect loops and increment iteration
+                if next_node in visited_in_iteration:
+                    iteration += 1
+                    visited_in_iteration.clear()
+                    state = state.increment_iteration()
+                    logger.debug(f"Loop detected, iteration: {iteration}")
+                
+                visited_in_iteration.add(current_node)
+                current_node = next_node
+            
+            # Finalize
+            self._status = ExecutionStatus.COMPLETED
+            final_state = self._state_manager.finalize()
+            
+            return ExecutionResult(
+                run_id=self.run_id,
+                graph_id=self.graph.graph_id,
+                status=self._status,
+                final_state=final_state.data,
+                execution_log=self._execution_log,
+                started_at=final_state.started_at,
+                completed_at=final_state.completed_at,
+                total_duration_ms=(time.time() - start_time) * 1000,
+                iterations=iteration + 1,
+            )
+            
+        except Exception as e:
+            logger.exception(f"Execution failed: {e}")
+            return self._create_error_result(str(e), start_time)
+    
+    async def _execute_node(
+        self,
+        node,
+        state: WorkflowState,
+        iteration: int
+    ) -> ExecutionStep:
+        """Execute a single node and update state."""
+        self._step_counter += 1
+        step_start = datetime.now()
+        node_start_time = time.time()
+        
+        step = ExecutionStep(
+            step=self._step_counter,
+            node=node.name,
+            started_at=step_start,
+            iteration=iteration,
+        )
+        
+        logger.info(f"Executing node: {node.name} (step {self._step_counter})")
+        
+        try:
+            # Execute node handler
+            result_data = await node.execute(state.data)
+            
+            # Update state
+            new_state = state.update(result_data).mark_visited(node.name)
+            self._state_manager.update(new_state, node.name)
+            
+            # Complete step
+            step.completed_at = datetime.now()
+            step.duration_ms = (time.time() - node_start_time) * 1000
+            step.result = "success"
+            
+        except Exception as e:
+            logger.error(f"Node {node.name} failed: {e}")
+            step.completed_at = datetime.now()
+            step.duration_ms = (time.time() - node_start_time) * 1000
+            step.result = "error"
+            step.error = str(e)
+        
+        # Add to log
+        self._execution_log.append(step)
+        
+        # Notify callback
+        if self.on_step:
+            try:
+                self.on_step(step, self._state_manager.current_state.data)
+            except Exception as e:
+                logger.warning(f"Step callback failed: {e}")
+        
+        return step
+    
+    def _create_error_result(
+        self,
+        error: str,
+        start_time: float
+    ) -> ExecutionResult:
+        """Create an error result."""
+        self._status = ExecutionStatus.FAILED
+        return ExecutionResult(
+            run_id=self.run_id,
+            graph_id=self.graph.graph_id,
+            status=ExecutionStatus.FAILED,
+            final_state=self.current_state or {},
+            execution_log=self._execution_log,
+            started_at=datetime.now(),
+            completed_at=datetime.now(),
+            total_duration_ms=(time.time() - start_time) * 1000,
+            error=error,
+        )
+    
+    def get_execution_summary(self) -> Dict[str, Any]:
+        """Get a summary of the current execution."""
+        return {
+            "run_id": self.run_id,
+            "graph_id": self.graph.graph_id,
+            "status": self._status.value,
+            "current_node": self.current_node,
+            "current_state": self.current_state,
+            "step_count": self._step_counter,
+            "iteration": self._state_manager.current_state.iteration if self._state_manager and self._state_manager.current_state else 0,
+        }
+
+
+async def execute_graph(
+    graph: Graph,
+    initial_state: Dict[str, Any],
+    run_id: Optional[str] = None,
+    on_step: Optional[Callable] = None
+) -> ExecutionResult:
+    """
+    Convenience function to execute a graph.
+    
+    Args:
+        graph: The workflow graph
+        initial_state: Initial state data
+        run_id: Optional run ID
+        on_step: Optional step callback
+        
+    Returns:
+        ExecutionResult
+    """
+    executor = Executor(graph, run_id, on_step)
+    return await executor.run(initial_state)

app/engine/graph.py

@@ -0,0 +1,360 @@
+"""
+Graph Definition for Workflow Engine.
+
+The Graph is the core structure that defines the workflow - nodes, edges,
+conditional routing, and execution flow.
+"""
+
+from typing import Any, Callable, Dict, List, Optional, Set, Union
+from dataclasses import dataclass, field
+from enum import Enum
+import uuid
+
+from app.engine.node import Node, NodeType, get_registered_node, create_node_from_function
+
+
+# Special node names
+END = "__END__"
+START = "__START__"
+
+
+class EdgeType(str, Enum):
+    """Types of edges between nodes."""
+    DIRECT = "direct"           # Always follow this edge
+    CONDITIONAL = "conditional"  # Choose based on condition
+
+
+@dataclass
+class Edge:
+    """An edge connecting two nodes."""
+    source: str
+    target: str
+    edge_type: EdgeType = EdgeType.DIRECT
+    
+    def to_dict(self) -> Dict[str, str]:
+        return {
+            "source": self.source,
+            "target": self.target,
+            "type": self.edge_type.value
+        }
+
+
+@dataclass
+class ConditionalEdge:
+    """
+    A conditional edge that routes to different nodes based on a condition.
+    
+    The condition function receives the current state and returns a route key.
+    The routes dict maps route keys to target node names.
+    """
+    source: str
+    condition: Callable[[Dict[str, Any]], str]
+    routes: Dict[str, str]  # route_key -> target_node_name
+    
+    def evaluate(self, state_data: Dict[str, Any]) -> str:
+        """Evaluate the condition and return the target node name."""
+        route_key = self.condition(state_data)
+        if route_key not in self.routes:
+            raise ValueError(
+                f"Condition returned unknown route '{route_key}'. "
+                f"Available routes: {list(self.routes.keys())}"
+            )
+        return self.routes[route_key]
+    
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "source": self.source,
+            "condition": self.condition.__name__ if hasattr(self.condition, '__name__') else str(self.condition),
+            "routes": self.routes
+        }
+
+
+@dataclass
+class Graph:
+    """
+    A workflow graph consisting of nodes and edges.
+    
+    The graph defines the structure of a workflow:
+    - Nodes: Processing units that transform state
+    - Edges: Connections between nodes
+    - Conditional Edges: Branching logic based on state
+    
+    Attributes:
+        graph_id: Unique identifier for this graph
+        name: Human-readable name
+        nodes: Dict of node_name -> Node
+        edges: List of direct edges
+        conditional_edges: Dict of source_node -> ConditionalEdge
+        entry_point: Name of the first node to execute
+        max_iterations: Maximum loop iterations allowed
+    """
+    
+    graph_id: str = field(default_factory=lambda: str(uuid.uuid4()))
+    name: str = "Unnamed Workflow"
+    nodes: Dict[str, Node] = field(default_factory=dict)
+    edges: Dict[str, str] = field(default_factory=dict)  # source -> target for direct edges
+    conditional_edges: Dict[str, ConditionalEdge] = field(default_factory=dict)
+    entry_point: Optional[str] = None
+    max_iterations: int = 100
+    description: str = ""
+    metadata: Dict[str, Any] = field(default_factory=dict)
+    
+    def add_node(
+        self,
+        name: str,
+        handler: Optional[Callable] = None,
+        node_type: NodeType = NodeType.STANDARD,
+        description: str = ""
+    ) -> "Graph":
+        """
+        Add a node to the graph.
+        
+        If handler is not provided, attempts to find a registered node
+        with the given name.
+        
+        Args:
+            name: Unique name for the node
+            handler: Function to execute (optional if registered)
+            node_type: Type of node
+            description: Human-readable description
+            
+        Returns:
+            Self for chaining
+        """
+        if handler is None:
+            # Try to find a registered handler
+            handler = get_registered_node(name)
+            if handler is None:
+                raise ValueError(
+                    f"No handler provided for node '{name}' and no registered "
+                    f"node found with that name"
+                )
+        
+        if name in self.nodes:
+            raise ValueError(f"Node '{name}' already exists in the graph")
+        
+        node = create_node_from_function(handler, name, node_type, description)
+        self.nodes[name] = node
+        
+        # Set as entry point if it's the first node or marked as entry
+        if self.entry_point is None or node_type == NodeType.ENTRY:
+            self.entry_point = name
+        
+        return self
+    
+    def add_edge(self, source: str, target: str) -> "Graph":
+        """
+        Add a direct edge from source to target.
+        
+        Args:
+            source: Source node name
+            target: Target node name (or END)
+            
+        Returns:
+            Self for chaining
+        """
+        if source not in self.nodes:
+            raise ValueError(f"Source node '{source}' not found in graph")
+        if target != END and target not in self.nodes:
+            raise ValueError(f"Target node '{target}' not found in graph")
+        
+        # Check for conflicts with conditional edges
+        if source in self.conditional_edges:
+            raise ValueError(
+                f"Node '{source}' already has a conditional edge. "
+                f"Cannot add a direct edge."
+            )
+        
+        self.edges[source] = target
+        return self
+    
+    def add_conditional_edge(
+        self,
+        source: str,
+        condition: Callable[[Dict[str, Any]], str],
+        routes: Dict[str, str]
+    ) -> "Graph":
+        """
+        Add a conditional edge from source node.
+        
+        The condition function receives state and returns a route key.
+        
+        Args:
+            source: Source node name
+            condition: Function that returns route key
+            routes: Dict mapping route keys to target nodes
+            
+        Returns:
+            Self for chaining
+        """
+        if source not in self.nodes:
+            raise ValueError(f"Source node '{source}' not found in graph")
+        
+        # Validate all targets
+        for route_key, target in routes.items():
+            if target != END and target not in self.nodes:
+                raise ValueError(
+                    f"Target node '{target}' for route '{route_key}' not found in graph"
+                )
+        
+        # Check for conflicts with direct edges
+        if source in self.edges:
+            raise ValueError(
+                f"Node '{source}' already has a direct edge. "
+                f"Cannot add a conditional edge."
+            )
+        
+        self.conditional_edges[source] = ConditionalEdge(
+            source=source,
+            condition=condition,
+            routes=routes
+        )
+        return self
+    
+    def set_entry_point(self, node_name: str) -> "Graph":
+        """Set the entry point of the graph."""
+        if node_name not in self.nodes:
+            raise ValueError(f"Node '{node_name}' not found in graph")
+        self.entry_point = node_name
+        return self
+    
+    def get_next_node(self, current_node: str, state_data: Dict[str, Any]) -> Optional[str]:
+        """
+        Get the next node to execute based on edges and state.
+        
+        Args:
+            current_node: Current node name
+            state_data: Current state data
+            
+        Returns:
+            Next node name, END, or None if no edge defined
+        """
+        # Check for conditional edge first
+        if current_node in self.conditional_edges:
+            conditional = self.conditional_edges[current_node]
+            return conditional.evaluate(state_data)
+        
+        # Check for direct edge
+        if current_node in self.edges:
+            return self.edges[current_node]
+        
+        # No edge defined - implicit end
+        return None
+    
+    def validate(self) -> List[str]:
+        """
+        Validate the graph structure.
+        
+        Returns:
+            List of validation errors (empty if valid)
+        """
+        errors = []
+        
+        # Must have at least one node
+        if not self.nodes:
+            errors.append("Graph must have at least one node")
+            return errors
+        
+        # Must have an entry point
+        if not self.entry_point:
+            errors.append("Graph must have an entry point")
+        elif self.entry_point not in self.nodes:
+            errors.append(f"Entry point '{self.entry_point}' not found in nodes")
+        
+        # Check for orphan nodes (not reachable from entry point)
+        reachable = self._get_reachable_nodes()
+        orphans = set(self.nodes.keys()) - reachable
+        if orphans:
+            errors.append(f"Orphan nodes (not reachable): {orphans}")
+        
+        # Check that nodes without outgoing edges make sense
+        for node_name in self.nodes:
+            if node_name not in self.edges and node_name not in self.conditional_edges:
+                # This is an implicit end node - that's okay
+                pass
+        
+        return errors
+    
+    def _get_reachable_nodes(self) -> Set[str]:
+        """Get all nodes reachable from the entry point."""
+        if not self.entry_point:
+            return set()
+        
+        reachable = set()
+        to_visit = [self.entry_point]
+        
+        while to_visit:
+            node = to_visit.pop()
+            if node in reachable or node == END:
+                continue
+            
+            reachable.add(node)
+            
+            # Add direct edge target
+            if node in self.edges:
+                to_visit.append(self.edges[node])
+            
+            # Add conditional edge targets
+            if node in self.conditional_edges:
+                for target in self.conditional_edges[node].routes.values():
+                    to_visit.append(target)
+        
+        return reachable
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """Serialize the graph to a dictionary."""
+        return {
+            "graph_id": self.graph_id,
+            "name": self.name,
+            "description": self.description,
+            "nodes": {name: node.to_dict() for name, node in self.nodes.items()},
+            "edges": self.edges,
+            "conditional_edges": {
+                name: edge.to_dict() 
+                for name, edge in self.conditional_edges.items()
+            },
+            "entry_point": self.entry_point,
+            "max_iterations": self.max_iterations,
+            "metadata": self.metadata,
+        }
+    
+    def to_mermaid(self) -> str:
+        """Generate a Mermaid diagram of the graph."""
+        lines = ["graph TD"]
+        
+        # Add nodes
+        for name, node in self.nodes.items():
+            label = name.replace("_", " ").title()
+            if node.node_type == NodeType.ENTRY:
+                lines.append(f'    {name}["{label} 🚀"]')
+            elif node.node_type == NodeType.EXIT:
+                lines.append(f'    {name}["{label} 🏁"]')
+            else:
+                lines.append(f'    {name}["{label}"]')
+        
+        # Add END node if used
+        has_end = END in self.edges.values()
+        for cond in self.conditional_edges.values():
+            if END in cond.routes.values():
+                has_end = True
+                break
+        
+        if has_end:
+            lines.append(f'    {END}(("END"))')
+        
+        # Add direct edges
+        for source, target in self.edges.items():
+            lines.append(f"    {source} --> {target}")
+        
+        # Add conditional edges
+        for source, cond in self.conditional_edges.items():
+            for route_key, target in cond.routes.items():
+                lines.append(f"    {source} -->|{route_key}| {target}")
+        
+        return "\n".join(lines)
+    
+    def __repr__(self) -> str:
+        return (
+            f"Graph(name='{self.name}', nodes={list(self.nodes.keys())}, "
+            f"entry='{self.entry_point}')"
+        )

app/engine/node.py

@@ -0,0 +1,196 @@
+"""
+Node Definition for Workflow Engine.
+
+Nodes are the building blocks of a workflow. Each node is a function
+that receives state, performs some operation, and returns modified state.
+"""
+
+from typing import Any, Callable, Dict, Optional, Union
+from dataclasses import dataclass, field
+from enum import Enum
+import asyncio
+import inspect
+import functools
+
+
+class NodeType(str, Enum):
+    """Types of nodes in the workflow."""
+    STANDARD = "standard"      # Regular processing node
+    CONDITIONAL = "conditional"  # Branching decision node
+    ENTRY = "entry"            # Entry point
+    EXIT = "exit"              # Exit point
+
+
+@dataclass
+class Node:
+    """
+    A node in the workflow graph.
+    
+    Each node has a name and a handler function. The handler receives
+    the current state data (as a dict) and returns modified state data.
+    
+    Attributes:
+        name: Unique identifier for the node
+        handler: Function that processes state (sync or async)
+        node_type: Type of node (standard, conditional, etc.)
+        description: Human-readable description
+        metadata: Additional node metadata
+    """
+    
+    name: str
+    handler: Callable[[Dict[str, Any]], Union[Dict[str, Any], Any]]
+    node_type: NodeType = NodeType.STANDARD
+    description: str = ""
+    metadata: Dict[str, Any] = field(default_factory=dict)
+    
+    def __post_init__(self):
+        """Validate the node after initialization."""
+        if not self.name:
+            raise ValueError("Node name cannot be empty")
+        if not callable(self.handler):
+            raise ValueError(f"Handler for node '{self.name}' must be callable")
+    
+    @property
+    def is_async(self) -> bool:
+        """Check if the handler is an async function."""
+        return asyncio.iscoroutinefunction(self.handler)
+    
+    async def execute(self, state_data: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Execute the node handler with the given state data.
+        
+        Handles both sync and async handlers transparently.
+        
+        Args:
+            state_data: The current state data dictionary
+            
+        Returns:
+            Modified state data dictionary
+        """
+        try:
+            if self.is_async:
+                result = await self.handler(state_data)
+            else:
+                # Run sync handler in executor to not block
+                loop = asyncio.get_event_loop()
+                result = await loop.run_in_executor(
+                    None, 
+                    functools.partial(self.handler, state_data)
+                )
+            
+            # If handler returns None, return original state
+            if result is None:
+                return state_data
+            
+            # If handler returns a dict, use it as the new state
+            if isinstance(result, dict):
+                return result
+            
+            # Otherwise, something unexpected happened
+            raise ValueError(
+                f"Node '{self.name}' handler must return a dict or None, "
+                f"got {type(result).__name__}"
+            )
+            
+        except Exception as e:
+            # Add context to the error
+            raise RuntimeError(f"Error in node '{self.name}': {str(e)}") from e
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """Serialize the node to a dictionary."""
+        return {
+            "name": self.name,
+            "type": self.node_type.value,
+            "description": self.description,
+            "handler": self.handler.__name__ if hasattr(self.handler, '__name__') else str(self.handler),
+            "metadata": self.metadata,
+        }
+
+
+# Registry to hold decorated node functions
+_node_registry: Dict[str, Callable] = {}
+
+
+def node(
+    name: Optional[str] = None,
+    node_type: NodeType = NodeType.STANDARD,
+    description: str = ""
+) -> Callable:
+    """
+    Decorator to register a function as a workflow node.
+    
+    Usage:
+        @node(name="extract_functions", description="Extract functions from code")
+        def extract_functions(state: dict) -> dict:
+            # ... process state
+            return state
+    
+    Args:
+        name: Node name (defaults to function name)
+        node_type: Type of node
+        description: Human-readable description
+        
+    Returns:
+        Decorated function
+    """
+    def decorator(func: Callable) -> Callable:
+        node_name = name or func.__name__
+        
+        # Store metadata on the function
+        func._node_metadata = {
+            "name": node_name,
+            "type": node_type,
+            "description": description or func.__doc__ or "",
+        }
+        
+        # Register in global registry
+        _node_registry[node_name] = func
+        
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs):
+            return func(*args, **kwargs)
+        
+        wrapper._node_metadata = func._node_metadata
+        return wrapper
+    
+    return decorator
+
+
+def get_registered_node(name: str) -> Optional[Callable]:
+    """Get a registered node function by name."""
+    return _node_registry.get(name)
+
+
+def list_registered_nodes() -> Dict[str, Dict[str, Any]]:
+    """List all registered nodes and their metadata."""
+    return {
+        name: func._node_metadata 
+        for name, func in _node_registry.items()
+        if hasattr(func, '_node_metadata')
+    }
+
+
+def create_node_from_function(
+    func: Callable,
+    name: Optional[str] = None,
+    node_type: NodeType = NodeType.STANDARD,
+    description: str = ""
+) -> Node:
+    """
+    Create a Node instance from a function.
+    
+    Args:
+        func: The handler function
+        name: Node name (defaults to function name)
+        node_type: Type of node
+        description: Human-readable description
+        
+    Returns:
+        A Node instance
+    """
+    return Node(
+        name=name or func.__name__,
+        handler=func,
+        node_type=node_type,
+        description=description or func.__doc__ or "",
+    )

app/engine/state.py

@@ -0,0 +1,154 @@
+"""
+State Management for Workflow Engine.
+
+This module provides the state management system that flows through the workflow.
+State is immutable - each node receives state and returns a new modified state.
+"""
+
+from typing import Any, Dict, List, Optional
+from pydantic import BaseModel, Field
+from datetime import datetime
+from copy import deepcopy
+import uuid
+
+
+class StateSnapshot(BaseModel):
+    """A snapshot of state at a specific point in execution."""
+    
+    timestamp: datetime = Field(default_factory=datetime.now)
+    node_name: str
+    state_data: Dict[str, Any]
+    iteration: int = 0
+
+
+class WorkflowState(BaseModel):
+    """
+    The shared state that flows through the workflow.
+    
+    This is a flexible container that holds all data being processed
+    by the workflow nodes. Each node can read from and write to this state.
+    
+    Attributes:
+        data: The actual workflow data (flexible dictionary)
+        metadata: Execution metadata (iteration count, visited nodes, etc.)
+    """
+    
+    # The actual data being processed
+    data: Dict[str, Any] = Field(default_factory=dict)
+    
+    # Execution metadata
+    current_node: Optional[str] = None
+    iteration: int = 0
+    visited_nodes: List[str] = Field(default_factory=list)
+    started_at: Optional[datetime] = None
+    completed_at: Optional[datetime] = None
+    
+    class Config:
+        arbitrary_types_allowed = True
+    
+    def get(self, key: str, default: Any = None) -> Any:
+        """Get a value from the state data."""
+        return self.data.get(key, default)
+    
+    def set(self, key: str, value: Any) -> "WorkflowState":
+        """Set a value in state data and return a new state (immutable pattern)."""
+        new_data = deepcopy(self.data)
+        new_data[key] = value
+        return self.model_copy(update={"data": new_data})
+    
+    def update(self, updates: Dict[str, Any]) -> "WorkflowState":
+        """Update multiple values and return a new state."""
+        new_data = deepcopy(self.data)
+        new_data.update(updates)
+        return self.model_copy(update={"data": new_data})
+    
+    def mark_visited(self, node_name: str) -> "WorkflowState":
+        """Mark a node as visited."""
+        new_visited = self.visited_nodes + [node_name]
+        return self.model_copy(update={
+            "visited_nodes": new_visited,
+            "current_node": node_name
+        })
+    
+    def increment_iteration(self) -> "WorkflowState":
+        """Increment the iteration counter."""
+        return self.model_copy(update={"iteration": self.iteration + 1})
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert state to a plain dictionary."""
+        return {
+            "data": self.data,
+            "current_node": self.current_node,
+            "iteration": self.iteration,
+            "visited_nodes": self.visited_nodes,
+            "started_at": self.started_at.isoformat() if self.started_at else None,
+            "completed_at": self.completed_at.isoformat() if self.completed_at else None,
+        }
+    
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "WorkflowState":
+        """Create a WorkflowState from a dictionary."""
+        if "data" in data:
+            return cls(**data)
+        # If it's just raw data, wrap it
+        return cls(data=data)
+
+
+class StateManager:
+    """
+    Manages state history and snapshots for a workflow run.
+    
+    This provides debugging capabilities by tracking state changes
+    throughout the workflow execution.
+    """
+    
+    def __init__(self, run_id: Optional[str] = None):
+        self.run_id = run_id or str(uuid.uuid4())
+        self.history: List[StateSnapshot] = []
+        self._current_state: Optional[WorkflowState] = None
+    
+    @property
+    def current_state(self) -> Optional[WorkflowState]:
+        """Get the current state."""
+        return self._current_state
+    
+    def initialize(self, initial_data: Dict[str, Any]) -> WorkflowState:
+        """Initialize the state manager with initial data."""
+        self._current_state = WorkflowState(
+            data=initial_data,
+            started_at=datetime.now()
+        )
+        return self._current_state
+    
+    def update(self, new_state: WorkflowState, node_name: str) -> None:
+        """Update the current state and record a snapshot."""
+        # Record snapshot
+        snapshot = StateSnapshot(
+            node_name=node_name,
+            state_data=deepcopy(new_state.data),
+            iteration=new_state.iteration
+        )
+        self.history.append(snapshot)
+        
+        # Update current state
+        self._current_state = new_state
+    
+    def finalize(self) -> WorkflowState:
+        """Mark the workflow as complete."""
+        if self._current_state:
+            self._current_state = self._current_state.model_copy(
+                update={"completed_at": datetime.now()}
+            )
+        return self._current_state
+    
+    def get_history(self) -> List[Dict[str, Any]]:
+        """Get the state history as a list of dictionaries."""
+        return [
+            {
+                "timestamp": s.timestamp.isoformat(),
+                "node": s.node_name,
+                "iteration": s.iteration,
+                "state": s.state_data
+            }
+            for s in self.history
+        ]

app/main.py

@@ -0,0 +1,141 @@
+"""
+FlowGraph - FastAPI Application Entry Point.
+
+A lightweight, async-first workflow orchestration engine for building agent pipelines.
+"""
+
+from contextlib import asynccontextmanager
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+import logging
+
+from app.config import settings
+from app.api.routes import graph, tools, websocket
+from app.workflows.code_review import register_code_review_workflow
+
+# Import builtin tools to register them
+import app.tools.builtin  # noqa: F401
+
+
+# Configure logging
+logging.basicConfig(
+    level=getattr(logging, settings.LOG_LEVEL),
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """Application lifespan handler."""
+    # Startup
+    logger.info(f"Starting {settings.APP_NAME} v{settings.APP_VERSION}")
+    
+    # Register the demo workflow
+    await register_code_review_workflow()
+    
+    yield
+    
+    # Shutdown
+    logger.info("Shutting down...")
+
+
+# Create FastAPI application
+app = FastAPI(
+    title=settings.APP_NAME,
+    description="""
+## Workflow Engine API
+
+A minimal but powerful workflow/graph engine for building agent workflows.
+
+### Features
+- **Nodes**: Python functions that read and modify shared state
+- **Edges**: Define execution flow between nodes
+- **Branching**: Conditional routing based on state values
+- **Looping**: Support for iterative workflows
+- **Real-time Updates**: WebSocket support for live execution streaming
+
+### Quick Start
+1. List available tools: `GET /tools`
+2. Create a graph: `POST /graph/create`
+3. Run the graph: `POST /graph/run`
+4. Check execution state: `GET /graph/state/{run_id}`
+
+### Demo Workflow
+A pre-registered Code Review workflow is available with ID: `code-review-demo`
+    """,
+    version=settings.APP_VERSION,
+    docs_url="/docs",
+    redoc_url="/redoc",
+    lifespan=lifespan,
+)
+
+
+# Add CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+# Include routers
+app.include_router(graph.router)
+app.include_router(tools.router)
+app.include_router(websocket.router)
+
+
+# ============================================================
+# Root Endpoints
+# ============================================================
+
+@app.get("/", tags=["Root"])
+async def root():
+    """API root - returns basic info and links."""
+    return {
+        "name": settings.APP_NAME,
+        "version": settings.APP_VERSION,
+        "description": "A minimal workflow/graph engine for agent workflows",
+        "docs": "/docs",
+        "redoc": "/redoc",
+        "endpoints": {
+            "graphs": "/graph",
+            "tools": "/tools",
+            "websocket_run": "/ws/run/{graph_id}",
+            "websocket_subscribe": "/ws/subscribe/{run_id}",
+        },
+        "demo_workflow": "code-review-demo",
+    }
+
+
+@app.get("/health", tags=["Root"])
+async def health():
+    """Health check endpoint."""
+    from app.storage.memory import graph_storage, run_storage
+    
+    return {
+        "status": "healthy",
+        "version": settings.APP_VERSION,
+        "graphs_count": len(graph_storage),
+        "runs_count": len(run_storage),
+    }
+
+
+# ============================================================
+# Error Handlers
+# ============================================================
+
+@app.exception_handler(Exception)
+async def global_exception_handler(request, exc):
+    """Global exception handler for unhandled errors."""
+    logger.exception(f"Unhandled error: {exc}")
+    return JSONResponse(
+        status_code=500,
+        content={
+            "error": "Internal Server Error",
+            "detail": str(exc) if settings.DEBUG else "An unexpected error occurred",
+        },
+    )

app/storage/__init__.py

@@ -0,0 +1,17 @@
+"""
+Storage package - In-memory storage for graphs and runs.
+"""
+
+from app.storage.memory import (
+    GraphStorage,
+    RunStorage,
+    graph_storage,
+    run_storage,
+)
+
+__all__ = [
+    "GraphStorage",
+    "RunStorage",
+    "graph_storage",
+    "run_storage",
+]

app/storage/memory.py

@@ -0,0 +1,271 @@
+"""
+In-Memory Storage for Workflow Engine.
+
+Provides thread-safe storage for graphs and execution runs.
+Can be easily replaced with a database implementation.
+"""
+
+from typing import Any, Dict, List, Optional
+from datetime import datetime
+import asyncio
+from dataclasses import dataclass, field
+
+
+@dataclass
+class StoredGraph:
+    """A stored graph definition."""
+    graph_id: str
+    name: str
+    definition: Dict[str, Any]
+    created_at: datetime = field(default_factory=datetime.now)
+    updated_at: datetime = field(default_factory=datetime.now)
+    
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "graph_id": self.graph_id,
+            "name": self.name,
+            "definition": self.definition,
+            "created_at": self.created_at.isoformat(),
+            "updated_at": self.updated_at.isoformat(),
+        }
+
+
+@dataclass
+class StoredRun:
+    """A stored execution run."""
+    run_id: str
+    graph_id: str
+    status: str
+    initial_state: Dict[str, Any]
+    current_state: Dict[str, Any] = field(default_factory=dict)
+    final_state: Optional[Dict[str, Any]] = None
+    execution_log: List[Dict[str, Any]] = field(default_factory=list)
+    current_node: Optional[str] = None
+    iteration: int = 0
+    started_at: datetime = field(default_factory=datetime.now)
+    completed_at: Optional[datetime] = None
+    error: Optional[str] = None
+    
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "run_id": self.run_id,
+            "graph_id": self.graph_id,
+            "status": self.status,
+            "initial_state": self.initial_state,
+            "current_state": self.current_state,
+            "final_state": self.final_state,
+            "execution_log": self.execution_log,
+            "current_node": self.current_node,
+            "iteration": self.iteration,
+            "started_at": self.started_at.isoformat(),
+            "completed_at": self.completed_at.isoformat() if self.completed_at else None,
+            "error": self.error,
+        }
+
+
+class GraphStorage:
+    """
+    Thread-safe in-memory storage for workflow graphs.
+    
+    Stores graph definitions by their ID, allowing creation,
+    retrieval, update, and deletion operations.
+    """
+    
+    def __init__(self):
+        self._graphs: Dict[str, StoredGraph] = {}
+        self._lock = asyncio.Lock()
+    
+    async def save(self, graph_id: str, name: str, definition: Dict[str, Any]) -> StoredGraph:
+        """
+        Save a graph definition.
+        
+        Args:
+            graph_id: Unique graph identifier
+            name: Graph name
+            definition: Graph definition dict
+            
+        Returns:
+            The stored graph
+        """
+        async with self._lock:
+            stored = StoredGraph(
+                graph_id=graph_id,
+                name=name,
+                definition=definition,
+            )
+            self._graphs[graph_id] = stored
+            return stored
+    
+    async def get(self, graph_id: str) -> Optional[StoredGraph]:
+        """Get a graph by ID."""
+        async with self._lock:
+            return self._graphs.get(graph_id)
+    
+    async def update(self, graph_id: str, definition: Dict[str, Any]) -> Optional[StoredGraph]:
+        """Update a graph definition."""
+        async with self._lock:
+            if graph_id not in self._graphs:
+                return None
+            stored = self._graphs[graph_id]
+            stored.definition = definition
+            stored.updated_at = datetime.now()
+            return stored
+    
+    async def delete(self, graph_id: str) -> bool:
+        """Delete a graph."""
+        async with self._lock:
+            if graph_id in self._graphs:
+                del self._graphs[graph_id]
+                return True
+            return False
+    
+    async def list_all(self) -> List[StoredGraph]:
+        """List all stored graphs."""
+        async with self._lock:
+            return list(self._graphs.values())
+    
+    async def exists(self, graph_id: str) -> bool:
+        """Check if a graph exists."""
+        async with self._lock:
+            return graph_id in self._graphs
+    
+    def __len__(self) -> int:
+        return len(self._graphs)
+
+
+class RunStorage:
+    """
+    Thread-safe in-memory storage for execution runs.
+    
+    Stores run state, allowing real-time updates and queries
+    for ongoing and completed runs.
+    """
+    
+    def __init__(self):
+        self._runs: Dict[str, StoredRun] = {}
+        self._lock = asyncio.Lock()
+    
+    async def create(
+        self,
+        run_id: str,
+        graph_id: str,
+        initial_state: Dict[str, Any]
+    ) -> StoredRun:
+        """
+        Create a new run.
+        
+        Args:
+            run_id: Unique run identifier
+            graph_id: Associated graph ID
+            initial_state: Initial state data
+            
+        Returns:
+            The stored run
+        """
+        async with self._lock:
+            stored = StoredRun(
+                run_id=run_id,
+                graph_id=graph_id,
+                status="pending",
+                initial_state=initial_state,
+                current_state=initial_state.copy(),
+            )
+            self._runs[run_id] = stored
+            return stored
+    
+    async def get(self, run_id: str) -> Optional[StoredRun]:
+        """Get a run by ID."""
+        async with self._lock:
+            return self._runs.get(run_id)
+    
+    async def update_state(
+        self,
+        run_id: str,
+        current_state: Dict[str, Any],
+        current_node: Optional[str] = None,
+        iteration: Optional[int] = None
+    ) -> Optional[StoredRun]:
+        """Update the current state of a run."""
+        async with self._lock:
+            if run_id not in self._runs:
+                return None
+            stored = self._runs[run_id]
+            stored.current_state = current_state
+            stored.status = "running"
+            if current_node is not None:
+                stored.current_node = current_node
+            if iteration is not None:
+                stored.iteration = iteration
+            return stored
+    
+    async def add_log_entry(
+        self,
+        run_id: str,
+        entry: Dict[str, Any]
+    ) -> Optional[StoredRun]:
+        """Add an entry to the execution log."""
+        async with self._lock:
+            if run_id not in self._runs:
+                return None
+            self._runs[run_id].execution_log.append(entry)
+            return self._runs[run_id]
+    
+    async def complete(
+        self,
+        run_id: str,
+        final_state: Dict[str, Any],
+        execution_log: List[Dict[str, Any]]
+    ) -> Optional[StoredRun]:
+        """Mark a run as completed."""
+        async with self._lock:
+            if run_id not in self._runs:
+                return None
+            stored = self._runs[run_id]
+            stored.status = "completed"
+            stored.final_state = final_state
+            stored.execution_log = execution_log
+            stored.completed_at = datetime.now()
+            return stored
+    
+    async def fail(
+        self,
+        run_id: str,
+        error: str,
+        final_state: Optional[Dict[str, Any]] = None
+    ) -> Optional[StoredRun]:
+        """Mark a run as failed."""
+        async with self._lock:
+            if run_id not in self._runs:
+                return None
+            stored = self._runs[run_id]
+            stored.status = "failed"
+            stored.error = error
+            stored.final_state = final_state
+            stored.completed_at = datetime.now()
+            return stored
+    
+    async def list_all(self) -> List[StoredRun]:
+        """List all runs."""
+        async with self._lock:
+            return list(self._runs.values())
+    
+    async def list_by_graph(self, graph_id: str) -> List[StoredRun]:
+        """List all runs for a specific graph."""
+        async with self._lock:
+            return [r for r in self._runs.values() if r.graph_id == graph_id]
+    
+    async def delete(self, run_id: str) -> bool:
+        """Delete a run."""
+        async with self._lock:
+            if run_id in self._runs:
+                del self._runs[run_id]
+                return True
+            return False
+    
+    def __len__(self) -> int:
+        return len(self._runs)
+
+
+# Global storage instances
+graph_storage = GraphStorage()
+run_storage = RunStorage()

app/tools/__init__.py

@@ -0,0 +1,12 @@
+"""
+Tools package - Tool registry and built-in tools.
+"""
+
+from app.tools.registry import ToolRegistry, tool_registry, register_tool, get_tool
+
+__all__ = [
+    "ToolRegistry",
+    "tool_registry",
+    "register_tool",
+    "get_tool",
+]

app/tools/builtin.py

@@ -0,0 +1,387 @@
+"""
+Built-in Tools for the Code Review Workflow.
+
+These tools implement the functionality needed for the sample
+Code Review workflow demonstration.
+"""
+
+import re
+import ast
+from typing import Any, Dict, List, Optional
+from app.tools.registry import register_tool
+
+
+@register_tool(
+    name="extract_functions",
+    description="Extract function definitions from Python code"
+)
+def extract_functions(code: str) -> Dict[str, Any]:
+    """
+    Extract function names and basic info from Python code.
+    
+    Args:
+        code: Python source code string
+        
+    Returns:
+        Dict with 'functions' list containing function info
+    """
+    functions = []
+    
+    try:
+        tree = ast.parse(code)
+        
+        for node in ast.walk(tree):
+            if isinstance(node, ast.FunctionDef):
+                func_info = {
+                    "name": node.name,
+                    "lineno": node.lineno,
+                    "args": [arg.arg for arg in node.args.args],
+                    "has_docstring": (
+                        ast.get_docstring(node) is not None
+                    ),
+                    "decorators": [
+                        ast.unparse(d) if hasattr(ast, 'unparse') else str(d)
+                        for d in node.decorator_list
+                    ],
+                    "line_count": (
+                        node.end_lineno - node.lineno + 1
+                        if hasattr(node, 'end_lineno') and node.end_lineno
+                        else 0
+                    ),
+                }
+                functions.append(func_info)
+                
+    except SyntaxError as e:
+        return {
+            "functions": [],
+            "error": f"Syntax error in code: {e}",
+            "parse_success": False,
+        }
+    
+    return {
+        "functions": functions,
+        "function_count": len(functions),
+        "parse_success": True,
+    }
+
+
+@register_tool(
+    name="calculate_complexity",
+    description="Calculate complexity metrics for code"
+)
+def calculate_complexity(code: str, functions: Optional[List[Dict]] = None) -> Dict[str, Any]:
+    """
+    Calculate simple complexity metrics for Python code.
+    
+    Metrics:
+    - Lines of code (LOC)
+    - Cyclomatic complexity (simplified)
+    - Nesting depth
+    - Function count
+    
+    Args:
+        code: Python source code
+        functions: Optional pre-extracted function list
+        
+    Returns:
+        Dict with complexity metrics
+    """
+    lines = code.split('\n')
+    loc = len([l for l in lines if l.strip() and not l.strip().startswith('#')])
+    
+    # Simple cyclomatic complexity: count decision points
+    complexity_keywords = ['if', 'elif', 'for', 'while', 'and', 'or', 'except', 'with']
+    complexity = 1  # Base complexity
+    
+    for line in lines:
+        stripped = line.strip()
+        for keyword in complexity_keywords:
+            if re.match(rf'\b{keyword}\b', stripped):
+                complexity += 1
+    
+    # Calculate max nesting depth
+    max_depth = 0
+    current_depth = 0
+    for line in lines:
+        stripped = line.strip()
+        if stripped:
+            # Count leading spaces
+            indent = len(line) - len(line.lstrip())
+            depth = indent // 4  # Assume 4-space indentation
+            max_depth = max(max_depth, depth)
+    
+    # Calculate function count
+    func_count = len(functions) if functions else code.count('def ')
+    
+    # Generate a simple complexity score (1-10 scale)
+    # Lower is better
+    score = 10
+    if complexity > 10:
+        score -= 2
+    if complexity > 20:
+        score -= 2
+    if max_depth > 4:
+        score -= 1
+    if max_depth > 6:
+        score -= 1
+    if loc > 200:
+        score -= 1
+    if func_count > 10:
+        score -= 1
+    if functions:
+        long_funcs = [f for f in functions if f.get('line_count', 0) > 50]
+        score -= len(long_funcs)
+    
+    score = max(1, score)  # Minimum score of 1
+    
+    return {
+        "lines_of_code": loc,
+        "cyclomatic_complexity": complexity,
+        "max_nesting_depth": max_depth,
+        "function_count": func_count,
+        "complexity_score": score,
+    }
+
+
+@register_tool(
+    name="detect_issues",
+    description="Detect code quality issues and smells"
+)
+def detect_issues(
+    code: str,
+    functions: Optional[List[Dict]] = None,
+    complexity_score: Optional[int] = None
+) -> Dict[str, Any]:
+    """
+    Detect common code quality issues.
+    
+    Checks for:
+    - Missing docstrings
+    - Long functions
+    - Deep nesting
+    - Magic numbers
+    - TODO/FIXME comments
+    - Print statements (in production code)
+    - Unused imports (basic check)
+    
+    Args:
+        code: Python source code
+        functions: Optional pre-extracted functions
+        complexity_score: Optional pre-calculated complexity
+        
+    Returns:
+        Dict with issues list and summary
+    """
+    issues = []
+    lines = code.split('\n')
+    
+    # Check for missing docstrings
+    if functions:
+        for func in functions:
+            if not func.get('has_docstring'):
+                issues.append({
+                    "type": "missing_docstring",
+                    "severity": "warning",
+                    "message": f"Function '{func['name']}' lacks a docstring",
+                    "line": func.get('lineno'),
+                })
+    
+    # Check for long functions
+    if functions:
+        for func in functions:
+            line_count = func.get('line_count', 0)
+            if line_count > 50:
+                issues.append({
+                    "type": "long_function",
+                    "severity": "warning",
+                    "message": f"Function '{func['name']}' is too long ({line_count} lines)",
+                    "line": func.get('lineno'),
+                })
+    
+    # Check for TODO/FIXME
+    for i, line in enumerate(lines, 1):
+        if 'TODO' in line or 'FIXME' in line or 'XXX' in line:
+            issues.append({
+                "type": "todo_comment",
+                "severity": "info",
+                "message": f"Found TODO/FIXME comment",
+                "line": i,
+            })
+    
+    # Check for print statements
+    for i, line in enumerate(lines, 1):
+        stripped = line.strip()
+        if stripped.startswith('print(') or 'print(' in stripped:
+            issues.append({
+                "type": "print_statement",
+                "severity": "info",
+                "message": "Print statement found (consider using logging)",
+                "line": i,
+            })
+    
+    # Check for magic numbers
+    magic_number_pattern = r'\b(?<![\'".])\d{2,}\b(?![\'"])'
+    for i, line in enumerate(lines, 1):
+        # Skip comments and string assignments
+        stripped = line.strip()
+        if not stripped.startswith('#'):
+            matches = re.findall(magic_number_pattern, line)
+            for match in matches:
+                if int(match) not in (0, 1, 2, 100):  # Common acceptable values
+                    issues.append({
+                        "type": "magic_number",
+                        "severity": "info",
+                        "message": f"Magic number {match} found (consider using a constant)",
+                        "line": i,
+                    })
+                    break  # One per line is enough
+    
+    # Calculate quality score based on issues
+    quality_score = 10
+    for issue in issues:
+        if issue['severity'] == 'error':
+            quality_score -= 2
+        elif issue['severity'] == 'warning':
+            quality_score -= 1
+        else:
+            quality_score -= 0.5
+    
+    # Factor in complexity score if provided
+    if complexity_score:
+        quality_score = (quality_score + complexity_score) / 2
+    
+    quality_score = max(1, min(10, quality_score))
+    
+    return {
+        "issues": issues,
+        "issue_count": len(issues),
+        "quality_score": round(quality_score, 1),
+        "issues_by_severity": {
+            "error": len([i for i in issues if i['severity'] == 'error']),
+            "warning": len([i for i in issues if i['severity'] == 'warning']),
+            "info": len([i for i in issues if i['severity'] == 'info']),
+        }
+    }
+
+
+@register_tool(
+    name="suggest_improvements",
+    description="Generate improvement suggestions based on detected issues"
+)
+def suggest_improvements(
+    issues: List[Dict],
+    functions: Optional[List[Dict]] = None,
+    quality_score: Optional[float] = None
+) -> Dict[str, Any]:
+    """
+    Generate actionable improvement suggestions.
+    
+    Args:
+        issues: List of detected issues
+        functions: Optional function info
+        quality_score: Current quality score
+        
+    Returns:
+        Dict with suggestions and priority ranking
+    """
+    suggestions = []
+    
+    # Group issues by type
+    issue_types = {}
+    for issue in issues:
+        issue_type = issue.get('type', 'unknown')
+        if issue_type not in issue_types:
+            issue_types[issue_type] = []
+        issue_types[issue_type].append(issue)
+    
+    # Generate suggestions based on issue types
+    if 'missing_docstring' in issue_types:
+        count = len(issue_types['missing_docstring'])
+        suggestions.append({
+            "priority": "high",
+            "category": "documentation",
+            "suggestion": f"Add docstrings to {count} function(s)",
+            "details": "Good docstrings improve code maintainability and enable automatic documentation generation.",
+            "affected_functions": [i.get('message', '').split("'")[1] for i in issue_types['missing_docstring'] if "'" in i.get('message', '')],
+        })
+    
+    if 'long_function' in issue_types:
+        count = len(issue_types['long_function'])
+        suggestions.append({
+            "priority": "high",
+            "category": "refactoring",
+            "suggestion": f"Refactor {count} long function(s) into smaller units",
+            "details": "Functions over 50 lines are harder to understand and test. Consider extracting helper functions.",
+        })
+    
+    if 'print_statement' in issue_types:
+        count = len(issue_types['print_statement'])
+        suggestions.append({
+            "priority": "medium",
+            "category": "logging",
+            "suggestion": f"Replace {count} print statement(s) with proper logging",
+            "details": "Use the logging module for better control over log levels and output.",
+        })
+    
+    if 'magic_number' in issue_types:
+        count = len(issue_types['magic_number'])
+        suggestions.append({
+            "priority": "medium",
+            "category": "readability",
+            "suggestion": f"Extract {count} magic number(s) into named constants",
+            "details": "Named constants improve readability and make the code easier to modify.",
+        })
+    
+    if 'todo_comment' in issue_types:
+        count = len(issue_types['todo_comment'])
+        suggestions.append({
+            "priority": "low",
+            "category": "maintenance",
+            "suggestion": f"Address {count} TODO/FIXME comment(s)",
+            "details": "Consider creating issues or tasks to track these items.",
+        })
+    
+    # Add general suggestions if quality is low
+    if quality_score and quality_score < 5:
+        suggestions.append({
+            "priority": "high",
+            "category": "general",
+            "suggestion": "Consider a comprehensive code review",
+            "details": "The overall quality score is low. A thorough review may reveal structural improvements.",
+        })
+    
+    # Sort by priority
+    priority_order = {"high": 0, "medium": 1, "low": 2}
+    suggestions.sort(key=lambda x: priority_order.get(x['priority'], 3))
+    
+    # Calculate new expected quality score after improvements
+    potential_improvement = len(suggestions) * 0.5
+    new_quality_score = min(10, (quality_score or 5) + potential_improvement)
+    
+    return {
+        "suggestions": suggestions,
+        "suggestion_count": len(suggestions),
+        "current_quality_score": quality_score,
+        "potential_quality_score": round(new_quality_score, 1),
+        "categories": list(set(s['category'] for s in suggestions)),
+    }
+
+
+@register_tool(
+    name="quality_check",
+    description="Check if quality meets the threshold"
+)
+def quality_check(quality_score: float, quality_threshold: float = 7.0) -> str:
+    """
+    Simple routing function to check if quality meets threshold.
+    
+    Args:
+        quality_score: Current quality score (1-10)
+        quality_threshold: Minimum acceptable score
+        
+    Returns:
+        "pass" if quality meets threshold, "fail" otherwise
+    """
+    if quality_score >= quality_threshold:
+        return "pass"
+    return "fail"

app/tools/registry.py

@@ -0,0 +1,218 @@
+"""
+Tool Registry for Workflow Engine.
+
+The tool registry maintains a collection of callable tools that
+workflow nodes can use. Tools are simple Python functions that
+perform specific operations.
+"""
+
+from typing import Any, Callable, Dict, List, Optional
+from dataclasses import dataclass, field
+import functools
+import inspect
+import logging
+
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class Tool:
+    """
+    A registered tool.
+    
+    Attributes:
+        name: Unique identifier for the tool
+        func: The callable function
+        description: Human-readable description
+        parameters: Parameter descriptions
+    """
+    name: str
+    func: Callable
+    description: str = ""
+    parameters: Dict[str, str] = field(default_factory=dict)
+    
+    def __call__(self, *args, **kwargs) -> Any:
+        """Call the tool function."""
+        return self.func(*args, **kwargs)
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """Serialize tool metadata."""
+        return {
+            "name": self.name,
+            "description": self.description,
+            "parameters": self.parameters,
+        }
+
+
+class ToolRegistry:
+    """
+    Registry for workflow tools.
+    
+    Tools are simple Python functions that nodes can call to perform
+    specific operations. The registry allows dynamic registration
+    and lookup of tools.
+    
+    Usage:
+        registry = ToolRegistry()
+        
+        @registry.register("my_tool")
+        def my_tool(data: str) -> dict:
+            return {"result": data.upper()}
+        
+        # Later
+        tool = registry.get("my_tool")
+        result = tool("hello")
+    """
+    
+    def __init__(self):
+        self._tools: Dict[str, Tool] = {}
+    
+    def register(
+        self,
+        name: Optional[str] = None,
+        description: str = "",
+        parameters: Optional[Dict[str, str]] = None
+    ) -> Callable:
+        """
+        Decorator to register a function as a tool.
+        
+        Args:
+            name: Tool name (defaults to function name)
+            description: Tool description (defaults to docstring)
+            parameters: Parameter descriptions
+            
+        Returns:
+            Decorator function
+        """
+        def decorator(func: Callable) -> Callable:
+            tool_name = name or func.__name__
+            tool_desc = description or func.__doc__ or ""
+            
+            # Extract parameters from signature if not provided
+            params = parameters or {}
+            if not params:
+                sig = inspect.signature(func)
+                for param_name, param in sig.parameters.items():
+                    if param_name not in ("self", "cls"):
+                        params[param_name] = str(param.annotation) if param.annotation != inspect.Parameter.empty else "Any"
+            
+            # Create and store tool
+            tool = Tool(
+                name=tool_name,
+                func=func,
+                description=tool_desc.strip(),
+                parameters=params,
+            )
+            self._tools[tool_name] = tool
+            
+            logger.debug(f"Registered tool: {tool_name}")
+            
+            @functools.wraps(func)
+            def wrapper(*args, **kwargs):
+                return func(*args, **kwargs)
+            
+            return wrapper
+        
+        return decorator
+    
+    def add(
+        self,
+        func: Callable,
+        name: Optional[str] = None,
+        description: str = "",
+        parameters: Optional[Dict[str, str]] = None
+    ) -> None:
+        """
+        Directly add a function as a tool (non-decorator version).
+        
+        Args:
+            func: The function to register
+            name: Tool name (defaults to function name)
+            description: Tool description
+            parameters: Parameter descriptions
+        """
+        tool_name = name or func.__name__
+        tool_desc = description or func.__doc__ or ""
+        
+        tool = Tool(
+            name=tool_name,
+            func=func,
+            description=tool_desc.strip(),
+            parameters=parameters or {},
+        )
+        self._tools[tool_name] = tool
+        logger.debug(f"Added tool: {tool_name}")
+    
+    def get(self, name: str) -> Optional[Tool]:
+        """Get a tool by name."""
+        return self._tools.get(name)
+    
+    def call(self, name: str, *args, **kwargs) -> Any:
+        """
+        Call a tool by name.
+        
+        Args:
+            name: Tool name
+            *args: Positional arguments
+            **kwargs: Keyword arguments
+            
+        Returns:
+            Tool result
+            
+        Raises:
+            KeyError: If tool not found
+        """
+        tool = self.get(name)
+        if not tool:
+            raise KeyError(f"Tool '{name}' not found in registry")
+        return tool(*args, **kwargs)
+    
+    def remove(self, name: str) -> bool:
+        """Remove a tool from the registry."""
+        if name in self._tools:
+            del self._tools[name]
+            return True
+        return False
+    
+    def list_tools(self) -> List[Dict[str, Any]]:
+        """List all registered tools with their metadata."""
+        return [tool.to_dict() for tool in self._tools.values()]
+    
+    def has(self, name: str) -> bool:
+        """Check if a tool is registered."""
+        return name in self._tools
+    
+    def __contains__(self, name: str) -> bool:
+        return self.has(name)
+    
+    def __len__(self) -> int:
+        return len(self._tools)
+    
+    def __iter__(self):
+        return iter(self._tools.values())
+
+
+# Global tool registry instance
+tool_registry = ToolRegistry()
+
+
+def register_tool(
+    name: Optional[str] = None,
+    description: str = "",
+    parameters: Optional[Dict[str, str]] = None
+) -> Callable:
+    """
+    Convenience decorator to register a tool in the global registry.
+    
+    Usage:
+        @register_tool("my_tool", description="Does something cool")
+        def my_tool(data: str) -> dict:
+            return {"result": data}
+    """
+    return tool_registry.register(name, description, parameters)
+
+
+def get_tool(name: str) -> Optional[Tool]:
+    """Get a tool from the global registry."""
+    return tool_registry.get(name)

app/workflows/__init__.py

@@ -0,0 +1,10 @@
+"""
+Workflows package - Sample workflow implementations.
+"""
+
+from app.workflows.code_review import create_code_review_workflow, register_code_review_workflow
+
+__all__ = [
+    "create_code_review_workflow",
+    "register_code_review_workflow",
+]

app/workflows/code_review.py

@@ -0,0 +1,338 @@
+"""
+Code Review Workflow Implementation.
+
+This is the sample workflow demonstrating the workflow engine capabilities:
+1. Extract functions from code
+2. Check complexity
+3. Detect issues
+4. Suggest improvements
+5. Loop until quality_score >= threshold
+"""
+
+from typing import Any, Dict
+import logging
+
+from app.engine.graph import Graph, END
+from app.engine.node import node, NodeType
+from app.tools.builtin import (
+    extract_functions,
+    calculate_complexity,
+    detect_issues,
+    suggest_improvements,
+    quality_check,
+)
+from app.tools.registry import tool_registry
+
+
+logger = logging.getLogger(__name__)
+
+
+# ============================================================
+# Node Handlers (using the @node decorator)
+# ============================================================
+
+@node(name="extract_node", description="Extract functions from the input code")
+def extract_node(state: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Extract function definitions from the code.
+    
+    Input state requires:
+    - code: str - The Python source code to analyze
+    
+    Updates state with:
+    - functions: List[dict] - Extracted function information
+    - function_count: int - Number of functions found
+    """
+    code = state.get("code", "")
+    result = extract_functions(code)
+    state.update(result)
+    logger.info(f"Extracted {result.get('function_count', 0)} functions")
+    return state
+
+
+@node(name="complexity_node", description="Calculate code complexity metrics")
+def complexity_node(state: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Calculate complexity metrics for the code.
+    
+    Uses state:
+    - code: str - Source code
+    - functions: List[dict] - Previously extracted functions
+    
+    Updates state with:
+    - lines_of_code: int
+    - cyclomatic_complexity: int
+    - complexity_score: int (1-10)
+    """
+    code = state.get("code", "")
+    functions = state.get("functions", [])
+    result = calculate_complexity(code, functions)
+    state.update(result)
+    logger.info(f"Complexity score: {result.get('complexity_score', 0)}")
+    return state
+
+
+@node(name="issues_node", description="Detect code quality issues")
+def issues_node(state: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Detect code quality issues and calculate quality score.
+    
+    Uses state:
+    - code: str - Source code
+    - functions: List[dict] - Extracted functions
+    - complexity_score: int - From complexity check
+    
+    Updates state with:
+    - issues: List[dict] - Detected issues
+    - issue_count: int
+    - quality_score: float (1-10)
+    """
+    code = state.get("code", "")
+    functions = state.get("functions", [])
+    complexity_score = state.get("complexity_score")
+    
+    result = detect_issues(code, functions, complexity_score)
+    state.update(result)
+    
+    logger.info(
+        f"Found {result.get('issue_count', 0)} issues, "
+        f"quality score: {result.get('quality_score', 0)}"
+    )
+    return state
+
+
+@node(name="improve_node", description="Generate improvement suggestions")
+def improve_node(state: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Generate improvement suggestions based on detected issues.
+    
+    Uses state:
+    - issues: List[dict] - Detected issues
+    - functions: List[dict] - Extracted functions
+    - quality_score: float - Current quality score
+    
+    Updates state with:
+    - suggestions: List[dict] - Improvement suggestions
+    - suggestion_count: int
+    - potential_quality_score: float - Score after improvements
+    """
+    issues = state.get("issues", [])
+    functions = state.get("functions", [])
+    quality_score = state.get("quality_score", 5.0)
+    
+    result = suggest_improvements(issues, functions, quality_score)
+    state.update(result)
+    
+    # Simulate improvement by slightly increasing quality score
+    # In a real scenario, this would involve actual code modifications
+    improvement = min(0.5, result.get("suggestion_count", 0) * 0.2)
+    state["quality_score"] = min(10, quality_score + improvement)
+    
+    logger.info(
+        f"Generated {result.get('suggestion_count', 0)} suggestions, "
+        f"quality improved to {state['quality_score']}"
+    )
+    return state
+
+
+# Register node handlers as tools so they can be retrieved when rebuilding from storage
+def _wrapper_handler(handler_func):
+    """Create a wrapper that works with tool registry."""
+    def wrapper(state: Dict[str, Any]) -> Dict[str, Any]:
+        return handler_func(state)
+    wrapper.__name__ = handler_func.__name__
+    wrapper.__doc__ = handler_func.__doc__
+    return wrapper
+
+tool_registry.add(_wrapper_handler(extract_node), name="extract_node", description="Extract functions from code")
+tool_registry.add(_wrapper_handler(complexity_node), name="complexity_node", description="Calculate complexity")
+tool_registry.add(_wrapper_handler(issues_node), name="issues_node", description="Detect quality issues")
+tool_registry.add(_wrapper_handler(improve_node), name="improve_node", description="Suggest improvements")
+
+
+# ============================================================
+# Condition Functions
+# ============================================================
+
+def quality_meets_threshold(state: Dict[str, Any]) -> str:
+    """
+    Routing condition: check if quality meets threshold.
+    
+    Returns:
+    - "pass" if quality_score >= quality_threshold
+    - "fail" if more improvement needed
+    """
+    quality_score = state.get("quality_score", 0)
+    threshold = state.get("quality_threshold", 7.0)
+    
+    if quality_score >= threshold:
+        logger.info(f"Quality {quality_score} meets threshold {threshold}")
+        return "pass"
+    else:
+        logger.info(f"Quality {quality_score} below threshold {threshold}")
+        return "fail"
+
+
+def always_loop(state: Dict[str, Any]) -> str:
+    """Always return to issues check after improvement."""
+    return "continue"
+
+
+# ============================================================
+# Workflow Factory
+# ============================================================
+
+def create_code_review_workflow(
+    max_iterations: int = 5,
+    quality_threshold: float = 7.0
+) -> Graph:
+    """
+    Create a Code Review workflow graph.
+    
+    Workflow flow:
+    ```
+    extract → complexity → issues ─┬─→ END (if pass)
+                                   │
+                                   └─→ improve → issues (loop if fail)
+    ```
+    
+    Args:
+        max_iterations: Maximum improvement loops
+        quality_threshold: Minimum quality score to pass
+        
+    Returns:
+        Configured Graph instance
+    """
+    graph = Graph(
+        name="Code Review Workflow",
+        description=(
+            "Analyzes Python code for quality issues and suggests improvements. "
+            f"Loops until quality score >= {quality_threshold} or max {max_iterations} iterations."
+        ),
+        max_iterations=max_iterations,
+    )
+    
+    # Add nodes
+    graph.add_node("extract", handler=extract_node, description="Extract functions from code")
+    graph.add_node("complexity", handler=complexity_node, description="Calculate complexity")
+    graph.add_node("issues", handler=issues_node, description="Detect quality issues")
+    graph.add_node("improve", handler=improve_node, description="Suggest improvements")
+    
+    # Add edges
+    graph.add_edge("extract", "complexity")
+    graph.add_edge("complexity", "issues")
+    
+    # Conditional edge: issues → END or improve
+    graph.add_conditional_edge(
+        "issues",
+        quality_meets_threshold,
+        {"pass": END, "fail": "improve"}
+    )
+    
+    # Loop back from improve to issues
+    graph.add_conditional_edge(
+        "improve",
+        always_loop,
+        {"continue": "issues"}
+    )
+    
+    # Set entry point
+    graph.set_entry_point("extract")
+    
+    return graph
+
+
+async def register_code_review_workflow():
+    """
+    Register a pre-built Code Review workflow in storage.
+    
+    This makes the workflow available immediately via the API
+    without needing to create it first.
+    """
+    from app.storage.memory import graph_storage
+    
+    workflow = create_code_review_workflow()
+    
+    await graph_storage.save(
+        graph_id="code-review-demo",
+        name="Code Review Demo",
+        definition=workflow.to_dict(),
+    )
+    
+    logger.info("Registered Code Review workflow with ID: code-review-demo")
+    return workflow
+
+
+# ============================================================
+# Example Usage
+# ============================================================
+
+async def run_code_review_demo():
+    """
+    Demo function showing how to run the code review workflow.
+    
+    Usage:
+        import asyncio
+        from app.workflows.code_review import run_code_review_demo
+        asyncio.run(run_code_review_demo())
+    """
+    from app.engine.executor import execute_graph
+    
+    # Sample code to review
+    sample_code = '''
+def calculate_total(items):
+    total = 0
+    for item in items:
+        if item.price > 0:
+            if item.quantity > 0:
+                if item.discount:
+                    total += item.price * item.quantity * (1 - item.discount)
+                else:
+                    total += item.price * item.quantity
+    return total
+
+def process_data(data):
+    result = []
+    for i in range(len(data)):
+        if data[i] > 100:
+            result.append(data[i] * 2)
+        else:
+            result.append(data[i])
+    print(result)
+    return result
+
+
+def helper():
+    x = 42
+    return x * 1000
+'''
+    
+    # Create workflow
+    workflow = create_code_review_workflow(max_iterations=3, quality_threshold=6.0)
+    
+    # Initial state
+    initial_state = {
+        "code": sample_code,
+        "quality_threshold": 6.0,
+    }
+    
+    # Execute
+    print("Starting Code Review...")
+    result = await execute_graph(workflow, initial_state)
+    
+    # Print results
+    print(f"\nExecution Status: {result.status.value}")
+    print(f"Total Duration: {result.total_duration_ms:.2f}ms")
+    print(f"Iterations: {result.iterations}")
+    print(f"\nFinal Quality Score: {result.final_state.get('quality_score', 'N/A')}")
+    print(f"Issues Found: {result.final_state.get('issue_count', 'N/A')}")
+    print(f"\nSuggestions:")
+    for suggestion in result.final_state.get("suggestions", []):
+        print(f"  - [{suggestion['priority']}] {suggestion['suggestion']}")
+    
+    return result
+
+
+if __name__ == "__main__":
+    import asyncio
+    asyncio.run(run_code_review_demo())

docker-compose.yml

@@ -0,0 +1,39 @@
+services:
+  workflow-engine:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    container_name: workflow-engine
+    ports:
+      - "8000:8000"
+    environment:
+      - APP_NAME=FlowGraph
+      - APP_VERSION=1.0.0
+      - DEBUG=true
+      - HOST=0.0.0.0
+      - PORT=8000
+      - MAX_ITERATIONS=100
+      - LOG_LEVEL=INFO
+    volumes:
+      # Mount for development (hot reload)
+      - .:/app
+    command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload
+    healthcheck:
+      test: [ "CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')" ]
+      interval: 30s
+      timeout: 10s
+      retries: 3
+      start_period: 10s
+    restart: unless-stopped
+
+  # Optional: Run tests in a separate container
+  tests:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    container_name: workflow-engine-tests
+    command: pytest tests/ -v
+    profiles:
+      - test
+    depends_on:
+      - workflow-engine

pytest.ini

@@ -0,0 +1,6 @@
+[pytest]
+asyncio_mode = auto
+testpaths = tests
+python_files = test_*.py
+python_functions = test_*
+addopts = -v --tb=short

requirements.txt

@@ -0,0 +1,16 @@
+# Core
+fastapi>=0.104.0
+uvicorn[standard]>=0.24.0
+pydantic>=2.5.0
+pydantic-settings>=2.1.0
+
+# Async support
+asyncio-throttle>=1.0.2
+
+# Testing
+pytest>=7.4.0
+pytest-asyncio>=0.21.0
+httpx>=0.25.0
+
+# Optional - for better logging
+rich>=13.7.0

run.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python3
+"""
+Simple run script for the Workflow Engine.
+
+Usage:
+    python run.py
+    
+Or with custom settings:
+    HOST=127.0.0.1 PORT=8080 python run.py
+"""
+
+import uvicorn
+import os
+
+
+def main():
+    """Run the FastAPI application."""
+    host = os.getenv("HOST", "0.0.0.0")
+    port = int(os.getenv("PORT", "8000"))
+    reload = os.getenv("RELOAD", "true").lower() == "true"
+    
+    print(f"""
+╔═══════════════════════════════════════════════════════════════╗
+║                      FlowGraph 🔄                             ║
+║                                                               ║
+║  A lightweight workflow orchestration engine                  ║
+╠═══════════════════════════════════════════════════════════════╣
+║  Server:    http://{host}:{port}                                 ║
+║  API Docs:  http://{host}:{port}/docs                            ║
+║  ReDoc:     http://{host}:{port}/redoc                           ║
+╠═══════════════════════════════════════════════════════════════╣
+║  Demo workflow ID: code-review-demo                           ║
+╚═══════════════════════════════════════════════════════════════╝
+    """)
+    
+    uvicorn.run(
+        "app.main:app",
+        host=host,
+        port=port,
+        reload=reload,
+        log_level="info",
+    )
+
+
+if __name__ == "__main__":
+    main()

tests/__init__.py

@@ -0,0 +1,3 @@
+"""
+Tests package.
+"""

tests/test_api.py

@@ -0,0 +1,207 @@
+"""
+Tests for the FastAPI endpoints.
+"""
+
+import pytest
+from fastapi.testclient import TestClient
+from httpx import AsyncClient, ASGITransport
+
+from app.main import app
+
+
+# ============================================================
+# Sync Test Client (for simple tests)
+# ============================================================
+
+client = TestClient(app)
+
+
+class TestRootEndpoints:
+    """Tests for root endpoints."""
+    
+    def test_root(self):
+        """Test root endpoint."""
+        response = client.get("/")
+        assert response.status_code == 200
+        
+        data = response.json()
+        assert "name" in data
+        assert "version" in data
+        assert "endpoints" in data
+    
+    def test_health(self):
+        """Test health endpoint."""
+        response = client.get("/health")
+        assert response.status_code == 200
+        
+        data = response.json()
+        assert data["status"] == "healthy"
+
+
+class TestToolsEndpoints:
+    """Tests for tools endpoints."""
+    
+    def test_list_tools(self):
+        """Test listing tools."""
+        response = client.get("/tools/")
+        assert response.status_code == 200
+        
+        data = response.json()
+        assert "tools" in data
+        assert "total" in data
+        assert data["total"] > 0
+        
+        # Check that built-in tools are present
+        tool_names = [t["name"] for t in data["tools"]]
+        assert "extract_functions" in tool_names
+        assert "calculate_complexity" in tool_names
+    
+    def test_get_tool(self):
+        """Test getting a specific tool."""
+        response = client.get("/tools/extract_functions")
+        assert response.status_code == 200
+        
+        data = response.json()
+        assert data["name"] == "extract_functions"
+        assert "description" in data
+    
+    def test_get_nonexistent_tool(self):
+        """Test getting a tool that doesn't exist."""
+        response = client.get("/tools/nonexistent_tool")
+        assert response.status_code == 404
+
+
+class TestGraphEndpoints:
+    """Tests for graph endpoints."""
+    
+    def test_list_graphs(self):
+        """Test listing graphs."""
+        response = client.get("/graph/")
+        assert response.status_code == 200
+        
+        data = response.json()
+        assert "graphs" in data
+        assert "total" in data
+    
+    def test_get_demo_workflow(self):
+        """Test getting the demo workflow."""
+        response = client.get("/graph/code-review-demo")
+        assert response.status_code == 200
+        
+        data = response.json()
+        assert data["graph_id"] == "code-review-demo"
+        assert data["name"] == "Code Review Demo"
+        assert "mermaid_diagram" in data
+    
+    def test_create_graph(self):
+        """Test creating a new graph."""
+        graph_data = {
+            "name": "test_workflow",
+            "description": "A test workflow",
+            "nodes": [
+                {"name": "start", "handler": "extract_functions"},
+                {"name": "end", "handler": "calculate_complexity"}
+            ],
+            "edges": {
+                "start": "end"
+            },
+            "entry_point": "start"
+        }
+        
+        response = client.post("/graph/create", json=graph_data)
+        assert response.status_code == 201
+        
+        data = response.json()
+        assert "graph_id" in data
+        assert data["name"] == "test_workflow"
+        assert data["node_count"] == 2
+    
+    def test_create_graph_invalid_handler(self):
+        """Test creating a graph with invalid handler."""
+        graph_data = {
+            "name": "invalid_workflow",
+            "nodes": [
+                {"name": "bad", "handler": "nonexistent_handler"}
+            ],
+            "edges": {}
+        }
+        
+        response = client.post("/graph/create", json=graph_data)
+        assert response.status_code == 404
+
+
+# ============================================================
+# Async Tests (for async endpoints)
+# ============================================================
+
+@pytest.fixture
+def anyio_backend():
+    return "asyncio"
+
+
+@pytest.mark.asyncio
+async def test_run_demo_workflow():
+    """Test running the demo workflow."""
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as ac:
+        run_data = {
+            "graph_id": "code-review-demo",
+            "initial_state": {
+                "code": "def hello():\n    print('world')",
+                "quality_threshold": 5.0
+            },
+            "async_execution": False
+        }
+        
+        response = await ac.post("/graph/run", json=run_data)
+        assert response.status_code == 200
+        
+        data = response.json()
+        assert "run_id" in data
+        assert data["status"] in ["completed", "failed"]
+        assert "execution_log" in data
+
+
+@pytest.mark.asyncio
+async def test_async_execution():
+    """Test async execution mode."""
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as ac:
+        run_data = {
+            "graph_id": "code-review-demo",
+            "initial_state": {
+                "code": "def test(): pass",
+                "quality_threshold": 5.0
+            },
+            "async_execution": True
+        }
+        
+        response = await ac.post("/graph/run", json=run_data)
+        assert response.status_code == 200
+        
+        data = response.json()
+        assert "run_id" in data
+        assert data["status"] == "pending"
+        
+        # Check run state
+        run_id = data["run_id"]
+        state_response = await ac.get(f"/graph/state/{run_id}")
+        assert state_response.status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_run_nonexistent_graph():
+    """Test running a graph that doesn't exist."""
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as ac:
+        run_data = {
+            "graph_id": "nonexistent-graph",
+            "initial_state": {}
+        }
+        
+        response = await ac.post("/graph/run", json=run_data)
+        assert response.status_code == 404
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])

tests/test_engine.py

@@ -0,0 +1,413 @@
+"""
+Tests for the Workflow Engine core components.
+"""
+
+import pytest
+import asyncio
+from typing import Dict, Any
+
+from app.engine.state import WorkflowState, StateManager
+from app.engine.node import Node, NodeType, node, create_node_from_function
+from app.engine.graph import Graph, END
+from app.engine.executor import Executor, ExecutionStatus, execute_graph
+
+
+# ============================================================
+# State Tests
+# ============================================================
+
+class TestWorkflowState:
+    """Tests for WorkflowState."""
+    
+    def test_create_empty_state(self):
+        """Test creating an empty state."""
+        state = WorkflowState()
+        assert state.data == {}
+        assert state.iteration == 0
+        assert state.visited_nodes == []
+    
+    def test_create_state_with_data(self):
+        """Test creating state with initial data."""
+        state = WorkflowState(data={"key": "value"})
+        assert state.get("key") == "value"
+        assert state.get("missing") is None
+        assert state.get("missing", "default") == "default"
+    
+    def test_state_immutability(self):
+        """Test that state updates return new instances."""
+        state1 = WorkflowState(data={"a": 1})
+        state2 = state1.set("b", 2)
+        
+        assert state1.get("b") is None
+        assert state2.get("b") == 2
+        assert state1 is not state2
+    
+    def test_state_update_multiple(self):
+        """Test updating multiple values at once."""
+        state = WorkflowState(data={"a": 1})
+        new_state = state.update({"b": 2, "c": 3})
+        
+        assert new_state.get("a") == 1
+        assert new_state.get("b") == 2
+        assert new_state.get("c") == 3
+    
+    def test_state_mark_visited(self):
+        """Test marking nodes as visited."""
+        state = WorkflowState()
+        state = state.mark_visited("node1")
+        state = state.mark_visited("node2")
+        
+        assert "node1" in state.visited_nodes
+        assert "node2" in state.visited_nodes
+        assert state.current_node == "node2"
+    
+    def test_state_to_from_dict(self):
+        """Test serialization and deserialization."""
+        state = WorkflowState(data={"test": 123})
+        state_dict = state.to_dict()
+        
+        assert "data" in state_dict
+        assert state_dict["data"]["test"] == 123
+        
+        restored = WorkflowState.from_dict(state_dict)
+        assert restored.get("test") == 123
+
+
+class TestStateManager:
+    """Tests for StateManager."""
+    
+    def test_initialize(self):
+        """Test state manager initialization."""
+        manager = StateManager()
+        state = manager.initialize({"input": "test"})
+        
+        assert manager.current_state is not None
+        assert manager.current_state.get("input") == "test"
+        assert manager.current_state.started_at is not None
+    
+    def test_update_and_history(self):
+        """Test state updates create history."""
+        manager = StateManager()
+        state = manager.initialize({"count": 0})
+        
+        new_state = state.set("count", 1)
+        manager.update(new_state, "node1")
+        
+        assert len(manager.history) == 1
+        assert manager.history[0].node_name == "node1"
+        assert manager.current_state.get("count") == 1
+
+
+# ============================================================
+# Node Tests
+# ============================================================
+
+class TestNode:
+    """Tests for Node class."""
+    
+    def test_create_node(self):
+        """Test creating a node."""
+        def handler(state):
+            return state
+        
+        n = Node(name="test_node", handler=handler)
+        
+        assert n.name == "test_node"
+        assert n.handler == handler
+        assert n.node_type == NodeType.STANDARD
+    
+    def test_node_validation(self):
+        """Test node validation."""
+        with pytest.raises(ValueError, match="name cannot be empty"):
+            Node(name="", handler=lambda x: x)
+        
+        with pytest.raises(ValueError, match="must be callable"):
+            Node(name="test", handler="not a function")
+    
+    @pytest.mark.asyncio
+    async def test_sync_node_execution(self):
+        """Test executing a sync node."""
+        def handler(state):
+            state["processed"] = True
+            return state
+        
+        n = Node(name="test", handler=handler)
+        result = await n.execute({"input": "data"})
+        
+        assert result["processed"] is True
+        assert result["input"] == "data"
+    
+    @pytest.mark.asyncio
+    async def test_async_node_execution(self):
+        """Test executing an async node."""
+        async def async_handler(state):
+            await asyncio.sleep(0.01)
+            state["async_processed"] = True
+            return state
+        
+        n = Node(name="async_test", handler=async_handler)
+        assert n.is_async is True
+        
+        result = await n.execute({"input": "data"})
+        assert result["async_processed"] is True
+    
+    def test_node_decorator(self):
+        """Test the @node decorator."""
+        @node(name="decorated_node", description="A test node")
+        def my_handler(state):
+            return state
+        
+        assert hasattr(my_handler, "_node_metadata")
+        assert my_handler._node_metadata["name"] == "decorated_node"
+
+
+# ============================================================
+# Graph Tests
+# ============================================================
+
+class TestGraph:
+    """Tests for Graph class."""
+    
+    def test_create_graph(self):
+        """Test creating a graph."""
+        graph = Graph(name="Test Graph")
+        assert graph.name == "Test Graph"
+        assert len(graph.nodes) == 0
+    
+    def test_add_nodes(self):
+        """Test adding nodes to a graph."""
+        graph = Graph()
+        graph.add_node("node1", handler=lambda s: s)
+        graph.add_node("node2", handler=lambda s: s)
+        
+        assert "node1" in graph.nodes
+        assert "node2" in graph.nodes
+        assert graph.entry_point == "node1"  # First node is entry
+    
+    def test_add_edges(self):
+        """Test adding edges."""
+        graph = Graph()
+        graph.add_node("a", handler=lambda s: s)
+        graph.add_node("b", handler=lambda s: s)
+        graph.add_edge("a", "b")
+        
+        assert graph.edges["a"] == "b"
+    
+    def test_add_edge_to_end(self):
+        """Test adding edge to END."""
+        graph = Graph()
+        graph.add_node("a", handler=lambda s: s)
+        graph.add_edge("a", END)
+        
+        assert graph.edges["a"] == END
+    
+    def test_invalid_edge(self):
+        """Test adding invalid edges raises error."""
+        graph = Graph()
+        graph.add_node("a", handler=lambda s: s)
+        
+        with pytest.raises(ValueError, match="not found"):
+            graph.add_edge("a", "nonexistent")
+    
+    def test_conditional_edge(self):
+        """Test conditional edges."""
+        graph = Graph()
+        graph.add_node("check", handler=lambda s: s)
+        graph.add_node("yes", handler=lambda s: s)
+        graph.add_node("no", handler=lambda s: s)
+        
+        def condition(state):
+            return "yes" if state.get("value") else "no"
+        
+        graph.add_conditional_edge("check", condition, {"yes": "yes", "no": "no"})
+        
+        # Test routing
+        assert graph.get_next_node("check", {"value": True}) == "yes"
+        assert graph.get_next_node("check", {"value": False}) == "no"
+    
+    def test_graph_validation(self):
+        """Test graph validation."""
+        graph = Graph()
+        
+        # Empty graph should fail
+        errors = graph.validate()
+        assert len(errors) > 0
+        
+        # Valid graph
+        graph.add_node("start", handler=lambda s: s)
+        graph.add_edge("start", END)
+        
+        errors = graph.validate()
+        assert len(errors) == 0
+    
+    def test_mermaid_generation(self):
+        """Test Mermaid diagram generation."""
+        graph = Graph()
+        graph.add_node("a", handler=lambda s: s)
+        graph.add_node("b", handler=lambda s: s)
+        graph.add_edge("a", "b")
+        graph.add_edge("b", END)
+        
+        mermaid = graph.to_mermaid()
+        
+        assert "graph TD" in mermaid
+        assert "a" in mermaid
+        assert "b" in mermaid
+
+
+# ============================================================
+# Executor Tests
+# ============================================================
+
+class TestExecutor:
+    """Tests for the Executor."""
+    
+    @pytest.mark.asyncio
+    async def test_simple_execution(self):
+        """Test executing a simple graph."""
+        graph = Graph()
+        graph.add_node("double", handler=lambda s: {**s, "value": s["value"] * 2})
+        graph.add_edge("double", END)
+        
+        result = await execute_graph(graph, {"value": 5})
+        
+        assert result.status == ExecutionStatus.COMPLETED
+        assert result.final_state["value"] == 10
+    
+    @pytest.mark.asyncio
+    async def test_multi_node_execution(self):
+        """Test executing multiple nodes."""
+        graph = Graph()
+        graph.add_node("add1", handler=lambda s: {**s, "value": s["value"] + 1})
+        graph.add_node("add2", handler=lambda s: {**s, "value": s["value"] + 2})
+        graph.add_edge("add1", "add2")
+        graph.add_edge("add2", END)
+        
+        result = await execute_graph(graph, {"value": 0})
+        
+        assert result.status == ExecutionStatus.COMPLETED
+        assert result.final_state["value"] == 3
+        assert len(result.execution_log) == 2
+    
+    @pytest.mark.asyncio
+    async def test_conditional_execution(self):
+        """Test conditional branching."""
+        graph = Graph()
+        graph.add_node("start", handler=lambda s: s)
+        graph.add_node("high", handler=lambda s: {**s, "path": "high"})
+        graph.add_node("low", handler=lambda s: {**s, "path": "low"})
+        
+        def route(state):
+            return "high" if state["value"] > 5 else "low"
+        
+        graph.add_conditional_edge("start", route, {"high": "high", "low": "low"})
+        graph.add_edge("high", END)
+        graph.add_edge("low", END)
+        
+        # Test high path
+        result = await execute_graph(graph, {"value": 10})
+        assert result.final_state["path"] == "high"
+        
+        # Test low path
+        result = await execute_graph(graph, {"value": 2})
+        assert result.final_state["path"] == "low"
+    
+    @pytest.mark.asyncio
+    async def test_loop_execution(self):
+        """Test looping execution."""
+        graph = Graph(max_iterations=10)
+        
+        def increment(state):
+            return {**state, "count": state["count"] + 1}
+        
+        def check_count(state):
+            return "done" if state["count"] >= 3 else "continue"
+        
+        graph.add_node("increment", handler=increment)
+        graph.add_conditional_edge("increment", check_count, {"done": END, "continue": "increment"})
+        
+        result = await execute_graph(graph, {"count": 0})
+        
+        assert result.status == ExecutionStatus.COMPLETED
+        assert result.final_state["count"] == 3
+    
+    @pytest.mark.asyncio
+    async def test_max_iterations(self):
+        """Test max iterations limit."""
+        graph = Graph(max_iterations=3)
+        
+        # Infinite loop
+        graph.add_node("loop", handler=lambda s: s)
+        graph.add_conditional_edge("loop", lambda s: "continue", {"continue": "loop"})
+        
+        result = await execute_graph(graph, {})
+        
+        assert result.status == ExecutionStatus.FAILED
+        assert "Max iterations" in result.error
+    
+    @pytest.mark.asyncio
+    async def test_error_handling(self):
+        """Test error handling during execution."""
+        def failing_handler(state):
+            raise ValueError("Intentional error")
+        
+        graph = Graph()
+        graph.add_node("fail", handler=failing_handler)
+        
+        result = await execute_graph(graph, {})
+        
+        assert result.status == ExecutionStatus.FAILED
+        assert "Intentional error" in result.error
+    
+    @pytest.mark.asyncio
+    async def test_execution_log(self):
+        """Test that execution log is properly generated."""
+        graph = Graph()
+        graph.add_node("step1", handler=lambda s: s)
+        graph.add_node("step2", handler=lambda s: s)
+        graph.add_edge("step1", "step2")
+        graph.add_edge("step2", END)
+        
+        result = await execute_graph(graph, {})
+        
+        assert len(result.execution_log) == 2
+        assert result.execution_log[0].node == "step1"
+        assert result.execution_log[1].node == "step2"
+        assert all(s.duration_ms > 0 for s in result.execution_log)
+
+
+# ============================================================
+# Integration Tests
+# ============================================================
+
+class TestCodeReviewWorkflow:
+    """Integration tests for the Code Review workflow."""
+    
+    @pytest.mark.asyncio
+    async def test_code_review_workflow(self):
+        """Test the full code review workflow."""
+        from app.workflows.code_review import create_code_review_workflow
+        
+        sample_code = '''
+def hello():
+    """Says hello."""
+    print("Hello, World!")
+
+def add(a, b):
+    return a + b
+'''
+        
+        workflow = create_code_review_workflow(max_iterations=3, quality_threshold=5.0)
+        result = await execute_graph(workflow, {
+            "code": sample_code,
+            "quality_threshold": 5.0,
+        })
+        
+        assert result.status == ExecutionStatus.COMPLETED
+        assert "functions" in result.final_state
+        assert "quality_score" in result.final_state
+        assert len(result.execution_log) > 0
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])