Upload 17 files

.env.example

@@ -0,0 +1,7 @@
+# Environment variables for Tritan
+
+# LLM API Keys (add your keys here)
+GROQ_API_KEY=
+OPENROUTER_API_KEY=
+CEREBRAS_API_KEY=
+GEMINI_API_KEY=

.gitignore

@@ -0,0 +1,14 @@
+# Environment
+.env
+.venv
+venv/
+__pycache__/
+*.pyc
+
+# IDE
+.vscode/
+.idea/
+
+# OS
+.DS_Store
+Thumbs.db

Dockerfile

@@ -0,0 +1,16 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+# Install dependencies
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy app
+COPY . .
+
+# HuggingFace Spaces uses port 7860
+EXPOSE 7860
+
+# Run with uvicorn
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,11 +1,40 @@
----
-title: Tritan Api
-emoji: 🔥
-colorFrom: gray
-colorTo: gray
-sdk: docker
-pinned: false
-license: apache-2.0
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+---
+title: Tritan API
+emoji: 🔱
+colorFrom: purple
+colorTo: indigo
+sdk: docker
+pinned: false
+license: mit
+app_port: 7860
+---
+
+# Tritan API - AI-Native Workflow Engine
+
+Backend API for the Tritan workflow engine.
+
+## Endpoints
+
+- `GET /` - API info
+- `GET /health` - Health check
+- `GET /docs` - Swagger documentation
+- `POST /api/planner/generate` - Generate workflow from task description
+- `POST /api/workflows` - Create workflow
+- `GET /api/workflows` - List workflows
+- `POST /api/execute/{workflow_id}` - Execute workflow
+
+## Environment Variables
+
+Set these as Secrets in your HuggingFace Space:
+
+- `GROQ_API_KEY` - Your Groq API key for LLM calls
+- `OPENROUTER_API_KEY` - (Optional) OpenRouter API key
+- `GEMINI_API_KEY` - (Optional) Google Gemini API key
+
+## Usage
+
+```bash
+curl -X POST https://YOUR-SPACE.hf.space/api/planner/generate \
+  -H "Content-Type: application/json" \
+  -d '{"task": "Monitor Bitcoin price and alert when it drops 5%"}'
+```

api/__init__.py

@@ -0,0 +1 @@
+# API package

api/execute.py

@@ -0,0 +1,47 @@
+"""
+Workflow execution endpoints
+"""
+from fastapi import APIRouter, HTTPException
+from fastapi.responses import StreamingResponse
+from models.workflow import Workflow
+from models.execution import ExecutionState, ExecutionResult, ExecutionStatus, NodeResult
+from core.engine import WorkflowEngine
+from typing import AsyncGenerator
+import json
+
+router = APIRouter()
+engine = WorkflowEngine()
+
+
+@router.post("/")
+async def execute_workflow(workflow: Workflow):
+    """Execute a workflow and return results"""
+    try:
+        result = await engine.execute(workflow)
+        return result
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@router.post("/stream")
+async def execute_workflow_stream(workflow: Workflow):
+    """Execute a workflow with streaming results"""
+    
+    async def generate() -> AsyncGenerator[str, None]:
+        async for event in engine.execute_stream(workflow):
+            yield f"data: {json.dumps(event)}\n\n"
+    
+    return StreamingResponse(
+        generate(),
+        media_type="text/event-stream"
+    )
+
+
+@router.post("/node/{node_type}")
+async def execute_single_node(node_type: str, config: dict):
+    """Execute a single node (for testing)"""
+    try:
+        result = await engine.execute_node(node_type, config)
+        return result
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))

api/planner.py

@@ -0,0 +1,146 @@
+"""
+Auto-Planner API - Generates workflows from natural language descriptions
+"""
+import os
+import json
+import httpx
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel
+from typing import Optional
+from dotenv import load_dotenv
+
+load_dotenv()
+
+router = APIRouter()
+
+GROQ_API_KEY = os.getenv("GROQ_API_KEY")
+GROQ_API_URL = "https://api.groq.com/openai/v1/chat/completions"
+
+SYSTEM_PROMPT = """You are a workflow planner for Tritan, an AI-native workflow engine.
+Given a task description, create a workflow using these node types:
+
+- trigger: Starting point of the workflow (ALWAYS required as first node)
+- action: Execute a generic task
+- condition: Branch based on a condition (has 'condition' field like "value > 10")
+- loop: Iterate over items
+- llm: AI language model call (has 'prompt', 'provider', 'model' fields)
+- http: HTTP API request (has 'url', 'method' fields)
+- code: Custom JavaScript code execution (has 'code' field)
+- transform: Data transformation
+
+Return ONLY valid JSON in this exact format, no markdown, no explanation:
+{
+  "name": "Descriptive Workflow Name",
+  "nodes": [
+    {"id": "n1", "type": "trigger", "x": 100, "y": 150, "data": {}},
+    {"id": "n2", "type": "llm", "x": 350, "y": 150, "data": {"prompt": "Your prompt here", "provider": "groq", "model": "llama-3.3-70b-versatile"}},
+    {"id": "n3", "type": "http", "x": 600, "y": 150, "data": {"url": "https://api.example.com", "method": "POST"}}
+  ],
+  "connections": [
+    {"from": "n1", "to": "n2"},
+    {"from": "n2", "to": "n3"}
+  ]
+}
+
+Rules:
+- Position nodes horizontally: x increases by 250 for each step
+- Keep y around 100-200 for a clean layout
+- Always start with a trigger node
+- Make prompts specific and actionable
+- Use realistic URLs for HTTP nodes
+- For conditions, write clear boolean expressions"""
+
+
+class PlannerRequest(BaseModel):
+    task: str
+    provider: Optional[str] = "groq"
+    model: Optional[str] = "llama-3.3-70b-versatile"
+
+
+class PlannerResponse(BaseModel):
+    name: str
+    nodes: list
+    connections: list
+
+
+@router.post("/generate", response_model=PlannerResponse)
+async def generate_workflow(request: PlannerRequest):
+    """Generate a workflow from a natural language task description."""
+    
+    if not request.task.strip():
+        raise HTTPException(status_code=400, detail="Task description is required")
+    
+    if not GROQ_API_KEY:
+        raise HTTPException(status_code=500, detail="GROQ_API_KEY not configured in backend")
+    
+    try:
+        async with httpx.AsyncClient(timeout=60.0) as client:
+            response = await client.post(
+                GROQ_API_URL,
+                headers={
+                    "Content-Type": "application/json",
+                    "Authorization": f"Bearer {GROQ_API_KEY}"
+                },
+                json={
+                    "model": request.model,
+                    "messages": [
+                        {"role": "system", "content": SYSTEM_PROMPT},
+                        {"role": "user", "content": f"Create a workflow for this task: {request.task}"}
+                    ],
+                    "temperature": 0.3,
+                    "max_tokens": 2000
+                }
+            )
+            
+            if response.status_code != 200:
+                error_data = response.json()
+                raise HTTPException(
+                    status_code=response.status_code,
+                    detail=error_data.get("error", {}).get("message", "API error")
+                )
+            
+            data = response.json()
+            content = data["choices"][0]["message"]["content"]
+            
+            # Extract JSON from response (handle markdown code blocks)
+            content = content.strip()
+            if content.startswith("```"):
+                content = content.split("```")[1]
+                if content.startswith("json"):
+                    content = content[4:]
+            
+            # Parse the workflow JSON
+            try:
+                workflow = json.loads(content)
+            except json.JSONDecodeError as e:
+                raise HTTPException(status_code=500, detail=f"Failed to parse workflow JSON: {str(e)}")
+            
+            # Validate required fields
+            if "nodes" not in workflow:
+                raise HTTPException(status_code=500, detail="Generated workflow missing 'nodes'")
+            if "connections" not in workflow:
+                workflow["connections"] = []
+            if "name" not in workflow:
+                workflow["name"] = "Generated Workflow"
+            
+            # Ensure all nodes have required fields
+            for i, node in enumerate(workflow["nodes"]):
+                if "id" not in node:
+                    node["id"] = f"node_{i}"
+                if "data" not in node:
+                    node["data"] = {}
+                if "x" not in node:
+                    node["x"] = 100 + (i * 250)
+                if "y" not in node:
+                    node["y"] = 150
+            
+            return PlannerResponse(
+                name=workflow["name"],
+                nodes=workflow["nodes"],
+                connections=workflow["connections"]
+            )
+            
+    except httpx.TimeoutException:
+        raise HTTPException(status_code=504, detail="Request to LLM timed out")
+    except httpx.RequestError as e:
+        raise HTTPException(status_code=500, detail=f"Request error: {str(e)}")

api/workflows.py

@@ -0,0 +1,89 @@
+"""
+Workflow CRUD endpoints
+"""
+from fastapi import APIRouter, HTTPException
+from models.workflow import Workflow, Node, Edge
+from typing import Optional
+
+router = APIRouter()
+
+# In-memory storage for MVP (frontend uses localStorage, this is for API testing)
+workflows_db: dict[str, Workflow] = {}
+
+
+@router.get("/")
+async def list_workflows():
+    """List all workflows"""
+    return {"workflows": list(workflows_db.values())}
+
+
+@router.get("/{workflow_id}")
+async def get_workflow(workflow_id: str):
+    """Get a specific workflow by ID"""
+    if workflow_id not in workflows_db:
+        raise HTTPException(status_code=404, detail="Workflow not found")
+    return workflows_db[workflow_id]
+
+
+@router.post("/")
+async def create_workflow(workflow: Workflow):
+    """Create a new workflow"""
+    workflows_db[workflow.id] = workflow
+    return {"message": "Workflow created", "workflow": workflow}
+
+
+@router.put("/{workflow_id}")
+async def update_workflow(workflow_id: str, workflow: Workflow):
+    """Update an existing workflow"""
+    if workflow_id not in workflows_db:
+        raise HTTPException(status_code=404, detail="Workflow not found")
+    workflow.id = workflow_id  # Ensure ID matches
+    workflows_db[workflow_id] = workflow
+    return {"message": "Workflow updated", "workflow": workflow}
+
+
+@router.delete("/{workflow_id}")
+async def delete_workflow(workflow_id: str):
+    """Delete a workflow"""
+    if workflow_id not in workflows_db:
+        raise HTTPException(status_code=404, detail="Workflow not found")
+    del workflows_db[workflow_id]
+    return {"message": "Workflow deleted"}
+
+
+@router.post("/validate")
+async def validate_workflow(workflow: Workflow):
+    """Validate a workflow structure (Guardian pre-check)"""
+    errors = []
+    warnings = []
+    
+    # Check for trigger node
+    trigger_nodes = [n for n in workflow.nodes if n.type == "trigger"]
+    if len(trigger_nodes) == 0:
+        errors.append("Workflow must have at least one trigger node")
+    elif len(trigger_nodes) > 1:
+        warnings.append("Multiple trigger nodes found - only first will be used")
+    
+    # Check for orphan nodes (no connections)
+    connected_nodes = set()
+    for edge in workflow.edges:
+        connected_nodes.add(edge.source)
+        connected_nodes.add(edge.target)
+    
+    for node in workflow.nodes:
+        if node.id not in connected_nodes and node.type != "trigger":
+            warnings.append(f"Node '{node.data.label}' is not connected to any other node")
+    
+    # Check LLM nodes have required fields
+    for node in workflow.nodes:
+        if node.type == "llm":
+            if not node.data.provider:
+                errors.append(f"LLM node '{node.data.label}' must have a provider")
+            if not node.data.model:
+                errors.append(f"LLM node '{node.data.label}' must have a model")
+    
+    return {
+        "valid": len(errors) == 0,
+        "errors": errors,
+        "warnings": warnings
+    }

core/__init__.py

@@ -0,0 +1,6 @@
+# Core package
+from .engine import WorkflowEngine
+from .nodes import NodeExecutor
+from .guardian import Guardian
+
+__all__ = ["WorkflowEngine", "NodeExecutor", "Guardian"]

core/engine.py

@@ -0,0 +1,212 @@
+"""
+Workflow Execution Engine
+"""
+from models.workflow import Workflow, Node, NodeType
+from models.execution import ExecutionState, ExecutionResult, ExecutionStatus, NodeResult
+from .nodes import NodeExecutor
+from .guardian import Guardian
+from datetime import datetime
+from typing import Any, AsyncGenerator
+import asyncio
+
+
+class WorkflowEngine:
+    """
+    Main workflow execution engine.
+    Handles node execution, parallel branching, and state management.
+    """
+    
+    def __init__(self):
+        self.node_executor = NodeExecutor()
+        self.guardian = Guardian()
+    
+    async def execute(self, workflow: Workflow) -> ExecutionResult:
+        """Execute a workflow and return the final result"""
+        start_time = datetime.now()
+        
+        # Initialize execution state
+        state = ExecutionState(
+            workflow_id=workflow.id,
+            status=ExecutionStatus.RUNNING,
+            started_at=start_time.isoformat()
+        )
+        
+        try:
+            # Build execution order (topological sort)
+            execution_order = self._build_execution_order(workflow)
+            
+            # Execute nodes in order
+            for node_id in execution_order:
+                node = self._get_node_by_id(workflow, node_id)
+                if not node:
+                    continue
+                
+                state.current_node = node_id
+                node_result = await self._execute_node(node, state)
+                state.node_results[node_id] = node_result
+                
+                # Check for failures
+                if node_result.status == ExecutionStatus.FAILED:
+                    state.status = ExecutionStatus.FAILED
+                    break
+            
+            if state.status != ExecutionStatus.FAILED:
+                state.status = ExecutionStatus.COMPLETED
+            
+        except Exception as e:
+            state.status = ExecutionStatus.FAILED
+            state.node_results["_error"] = NodeResult(
+                node_id="_error",
+                status=ExecutionStatus.FAILED,
+                error=str(e)
+            )
+        
+        end_time = datetime.now()
+        state.completed_at = end_time.isoformat()
+        
+        return ExecutionResult(
+            execution_id=state.execution_id,
+            workflow_id=workflow.id,
+            status=state.status,
+            node_results=list(state.node_results.values()),
+            final_output=self._get_final_output(state),
+            total_duration_ms=(end_time - start_time).total_seconds() * 1000,
+            started_at=state.started_at,
+            completed_at=state.completed_at
+        )
+    
+    async def execute_stream(self, workflow: Workflow) -> AsyncGenerator[dict, None]:
+        """Execute workflow with streaming events"""
+        start_time = datetime.now()
+        
+        yield {"type": "start", "workflow_id": workflow.id, "timestamp": start_time.isoformat()}
+        
+        state = ExecutionState(
+            workflow_id=workflow.id,
+            status=ExecutionStatus.RUNNING,
+            started_at=start_time.isoformat()
+        )
+        
+        try:
+            execution_order = self._build_execution_order(workflow)
+            
+            for node_id in execution_order:
+                node = self._get_node_by_id(workflow, node_id)
+                if not node:
+                    continue
+                
+                yield {"type": "node_start", "node_id": node_id, "node_type": node.type}
+                
+                node_result = await self._execute_node(node, state)
+                state.node_results[node_id] = node_result
+                
+                yield {
+                    "type": "node_complete",
+                    "node_id": node_id,
+                    "status": node_result.status,
+                    "output": node_result.output,
+                    "error": node_result.error
+                }
+                
+                if node_result.status == ExecutionStatus.FAILED:
+                    state.status = ExecutionStatus.FAILED
+                    break
+            
+            if state.status != ExecutionStatus.FAILED:
+                state.status = ExecutionStatus.COMPLETED
+                
+        except Exception as e:
+            state.status = ExecutionStatus.FAILED
+            yield {"type": "error", "message": str(e)}
+        
+        end_time = datetime.now()
+        yield {
+            "type": "complete",
+            "status": state.status,
+            "duration_ms": (end_time - start_time).total_seconds() * 1000
+        }
+    
+    async def execute_node(self, node_type: str, config: dict) -> dict:
+        """Execute a single node (for testing)"""
+        return await self.node_executor.execute(node_type, config, {})
+    
+    async def _execute_node(self, node: Node, state: ExecutionState) -> NodeResult:
+        """Execute a single node and return result"""
+        start_time = datetime.now()
+        
+        try:
+            output = await self.node_executor.execute(
+                node.type,
+                node.data.model_dump(),
+                state.variables
+            )
+            
+            end_time = datetime.now()
+            
+            # Store output in variables for next nodes
+            state.variables[node.id] = output
+            
+            return NodeResult(
+                node_id=node.id,
+                status=ExecutionStatus.COMPLETED,
+                output=output,
+                started_at=start_time.isoformat(),
+                completed_at=end_time.isoformat(),
+                duration_ms=(end_time - start_time).total_seconds() * 1000
+            )
+            
+        except Exception as e:
+            end_time = datetime.now()
+            return NodeResult(
+                node_id=node.id,
+                status=ExecutionStatus.FAILED,
+                error=str(e),
+                started_at=start_time.isoformat(),
+                completed_at=end_time.isoformat(),
+                duration_ms=(end_time - start_time).total_seconds() * 1000
+            )
+    
+    def _build_execution_order(self, workflow: Workflow) -> list[str]:
+        """Build topological order for node execution"""
+        # Build adjacency list
+        graph: dict[str, list[str]] = {node.id: [] for node in workflow.nodes}
+        in_degree: dict[str, int] = {node.id: 0 for node in workflow.nodes}
+        
+        for edge in workflow.edges:
+            graph[edge.source].append(edge.target)
+            in_degree[edge.target] += 1
+        
+        # Find trigger nodes (in_degree = 0)
+        queue = [node_id for node_id, degree in in_degree.items() if degree == 0]
+        result = []
+        
+        while queue:
+            node_id = queue.pop(0)
+            result.append(node_id)
+            
+            for neighbor in graph[node_id]:
+                in_degree[neighbor] -= 1
+                if in_degree[neighbor] == 0:
+                    queue.append(neighbor)
+        
+        return result
+    
+    def _get_node_by_id(self, workflow: Workflow, node_id: str) -> Node | None:
+        """Get a node by its ID"""
+        for node in workflow.nodes:
+            if node.id == node_id:
+                return node
+        return None
+    
+    def _get_final_output(self, state: ExecutionState) -> Any:
+        """Get the final output from the last executed node"""
+        if not state.node_results:
+            return None
+        
+        # Get the last completed node's output
+        for node_id in reversed(list(state.node_results.keys())):
+            result = state.node_results[node_id]
+            if result.status == ExecutionStatus.COMPLETED and result.output:
+                return result.output
+        
+        return None

core/guardian.py

@@ -0,0 +1,282 @@
+"""
+Guardian System - AI-powered workflow monitoring and validation
+"""
+from models.workflow import Workflow, Node, NodeType
+from models.execution import ExecutionState, GuardianAlert
+from typing import Optional
+import os
+import httpx
+
+
+class Guardian:
+    """
+    The Guardian is an AI agent that:
+    1. Validates workflows before execution
+    2. Monitors execution in real-time
+    3. Detects problems (infinite loops, missing resources)
+    4. Suggests improvements and fixes
+    """
+    
+    def __init__(self):
+        self.alerts: list[GuardianAlert] = []
+    
+    async def validate(self, workflow: Workflow) -> dict:
+        """
+        Validate a workflow before execution.
+        Returns validation result with errors and warnings.
+        """
+        errors = []
+        warnings = []
+        suggestions = []
+        
+        # 1. Check for trigger node
+        trigger_nodes = [n for n in workflow.nodes if n.type == NodeType.TRIGGER]
+        if len(trigger_nodes) == 0:
+            errors.append({
+                "type": "missing_trigger",
+                "message": "Workflow must have at least one trigger node",
+                "suggestion": "Add a Trigger node to start your workflow"
+            })
+        elif len(trigger_nodes) > 1:
+            warnings.append({
+                "type": "multiple_triggers",
+                "message": "Multiple trigger nodes found",
+                "suggestion": "Consider using only one trigger for clarity"
+            })
+        
+        # 2. Check for orphan nodes
+        connected_nodes = set()
+        for edge in workflow.edges:
+            connected_nodes.add(edge.source)
+            connected_nodes.add(edge.target)
+        
+        for node in workflow.nodes:
+            if node.id not in connected_nodes and node.type != NodeType.TRIGGER:
+                warnings.append({
+                    "type": "orphan_node",
+                    "node_id": node.id,
+                    "message": f"Node '{node.data.label}' is not connected",
+                    "suggestion": "Connect this node or remove it"
+                })
+        
+        # 3. Check for cycles (infinite loops)
+        if self._detect_cycle(workflow):
+            errors.append({
+                "type": "cycle_detected",
+                "message": "Workflow contains a cycle which may cause infinite loops",
+                "suggestion": "Remove the cyclic connection or add a loop limit"
+            })
+        
+        # 4. Validate LLM nodes
+        for node in workflow.nodes:
+            if node.type == NodeType.LLM:
+                if not node.data.provider:
+                    errors.append({
+                        "type": "missing_provider",
+                        "node_id": node.id,
+                        "message": f"LLM node '{node.data.label}' must have a provider",
+                        "suggestion": "Select a provider (groq, openrouter, cerebras, gemini)"
+                    })
+                if not node.data.model:
+                    errors.append({
+                        "type": "missing_model",
+                        "node_id": node.id,
+                        "message": f"LLM node '{node.data.label}' must have a model",
+                        "suggestion": "Select a model for this provider"
+                    })
+                if not node.data.prompt:
+                    warnings.append({
+                        "type": "empty_prompt",
+                        "node_id": node.id,
+                        "message": f"LLM node '{node.data.label}' has no prompt",
+                        "suggestion": "Add a prompt or use variables"
+                    })
+        
+        # 5. Check for empty workflow
+        if len(workflow.nodes) == 0:
+            errors.append({
+                "type": "empty_workflow",
+                "message": "Workflow has no nodes",
+                "suggestion": "Add nodes to create your workflow"
+            })
+        
+        # 6. Generate AI suggestions if Groq API key is available
+        if os.getenv("GROQ_API_KEY") and len(errors) == 0:
+            ai_suggestions = await self._get_ai_suggestions(workflow)
+            suggestions.extend(ai_suggestions)
+        
+        return {
+            "valid": len(errors) == 0,
+            "errors": errors,
+            "warnings": warnings,
+            "suggestions": suggestions,
+            "node_count": len(workflow.nodes),
+            "edge_count": len(workflow.edges)
+        }
+    
+    async def analyze_error(self, error: str, context: dict) -> dict:
+        """
+        Analyze an execution error and suggest fixes.
+        Uses AI to understand the error and provide solutions.
+        """
+        api_key = os.getenv("GROQ_API_KEY")
+        if not api_key:
+            return {
+                "analysis": "Unable to analyze - no API key configured",
+                "suggestions": [
+                    "Check the error message for clues",
+                    "Verify all node configurations are correct",
+                    "Ensure required variables are available"
+                ]
+            }
+        
+        prompt = f"""Analyze this workflow execution error and suggest fixes:
+
+Error: {error}
+
+Context:
+- Node Type: {context.get('node_type', 'unknown')}
+- Node Config: {context.get('config', {})}
+- Variables Available: {list(context.get('variables', {}).keys())}
+
+Provide:
+1. Brief analysis of what went wrong
+2. 2-3 specific suggestions to fix it
+
+Be concise and actionable."""
+
+        try:
+            async with httpx.AsyncClient() as client:
+                response = await client.post(
+                    "https://api.groq.com/openai/v1/chat/completions",
+                    headers={
+                        "Authorization": f"Bearer {api_key}",
+                        "Content-Type": "application/json"
+                    },
+                    json={
+                        "model": "llama-3.3-70b-versatile",
+                        "messages": [{"role": "user", "content": prompt}],
+                        "temperature": 0.3
+                    },
+                    timeout=30.0
+                )
+                response.raise_for_status()
+                data = response.json()
+                analysis = data["choices"][0]["message"]["content"]
+                
+                return {
+                    "analysis": analysis,
+                    "ai_powered": True
+                }
+        except Exception as e:
+            return {
+                "analysis": f"Error analyzing: {str(e)}",
+                "suggestions": ["Check your API configuration"]
+            }
+    
+    async def _get_ai_suggestions(self, workflow: Workflow) -> list[dict]:
+        """Get AI-powered suggestions for workflow improvement"""
+        api_key = os.getenv("GROQ_API_KEY")
+        if not api_key:
+            return []
+        
+        # Build workflow description
+        nodes_desc = ", ".join([f"{n.data.label} ({n.type})" for n in workflow.nodes])
+        
+        prompt = f"""Analyze this workflow and suggest 1-2 improvements:
+
+Workflow: {workflow.name}
+Nodes: {nodes_desc}
+Connections: {len(workflow.edges)}
+
+Focus on:
+- Performance optimizations
+- Error handling improvements
+- Best practices
+
+Be very concise (1-2 sentences per suggestion)."""
+
+        try:
+            async with httpx.AsyncClient() as client:
+                response = await client.post(
+                    "https://api.groq.com/openai/v1/chat/completions",
+                    headers={
+                        "Authorization": f"Bearer {api_key}",
+                        "Content-Type": "application/json"
+                    },
+                    json={
+                        "model": "llama-3.1-8b-instant",
+                        "messages": [{"role": "user", "content": prompt}],
+                        "temperature": 0.5,
+                        "max_tokens": 200
+                    },
+                    timeout=15.0
+                )
+                response.raise_for_status()
+                data = response.json()
+                content = data["choices"][0]["message"]["content"]
+                
+                return [{
+                    "type": "ai_suggestion",
+                    "message": content,
+                    "ai_powered": True
+                }]
+        except:
+            return []
+    
+    def _detect_cycle(self, workflow: Workflow) -> bool:
+        """Detect if workflow has a cycle using DFS"""
+        # Build adjacency list
+        graph: dict[str, list[str]] = {node.id: [] for node in workflow.nodes}
+        for edge in workflow.edges:
+            if edge.source in graph:
+                graph[edge.source].append(edge.target)
+        
+        visited = set()
+        rec_stack = set()
+        
+        def dfs(node: str) -> bool:
+            visited.add(node)
+            rec_stack.add(node)
+            
+            for neighbor in graph.get(node, []):
+                if neighbor not in visited:
+                    if dfs(neighbor):
+                        return True
+                elif neighbor in rec_stack:
+                    return True
+            
+            rec_stack.remove(node)
+            return False
+        
+        for node_id in graph:
+            if node_id not in visited:
+                if dfs(node_id):
+                    return True
+        
+        return False
+    
+    def create_alert(
+        self,
+        level: str,
+        message: str,
+        node_id: Optional[str] = None,
+        suggestion: Optional[str] = None
+    ) -> GuardianAlert:
+        """Create a new Guardian alert"""
+        alert = GuardianAlert(
+            level=level,
+            message=message,
+            node_id=node_id,
+            suggestion=suggestion
+        )
+        self.alerts.append(alert)
+        return alert
+    
+    def get_alerts(self) -> list[GuardianAlert]:
+        """Get all active alerts"""
+        return self.alerts
+    
+    def clear_alerts(self):
+        """Clear all alerts"""
+        self.alerts = []

core/nodes.py

@@ -0,0 +1,254 @@
+"""
+Node Executors - Handle execution of different node types
+"""
+from typing import Any
+import httpx
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+class NodeExecutor:
+    """
+    Executes individual nodes based on their type.
+    Supports: trigger, action, condition, loop, llm, http, code, transform
+    """
+    
+    def __init__(self):
+        # LLM provider configurations
+        self.llm_providers = {
+            "groq": {
+                "base_url": "https://api.groq.com/openai/v1",
+                "api_key_env": "GROQ_API_KEY",
+                "models": ["llama-3.3-70b-versatile", "llama-3.1-8b-instant", "mixtral-8x7b-32768"]
+            },
+            "openrouter": {
+                "base_url": "https://openrouter.ai/api/v1",
+                "api_key_env": "OPENROUTER_API_KEY",
+                "models": ["anthropic/claude-3.5-sonnet", "openai/gpt-4o", "google/gemini-2.0-flash-exp:free"]
+            },
+            "cerebras": {
+                "base_url": "https://api.cerebras.ai/v1",
+                "api_key_env": "CEREBRAS_API_KEY",
+                "models": ["llama3.1-8b", "llama3.1-70b"]
+            },
+            "gemini": {
+                "base_url": "https://generativelanguage.googleapis.com/v1beta",
+                "api_key_env": "GEMINI_API_KEY",
+                "models": ["gemini-2.0-flash-exp", "gemini-1.5-pro"]
+            }
+        }
+    
+    async def execute(self, node_type: str, config: dict, variables: dict) -> Any:
+        """Execute a node and return its output"""
+        executor = getattr(self, f"_execute_{node_type}", None)
+        if not executor:
+            raise ValueError(f"Unknown node type: {node_type}")
+        
+        return await executor(config, variables)
+    
+    async def _execute_trigger(self, config: dict, variables: dict) -> dict:
+        """Trigger node - starts the workflow"""
+        return {
+            "triggered": True,
+            "label": config.get("label", "Start"),
+            "timestamp": __import__("datetime").datetime.now().isoformat()
+        }
+    
+    async def _execute_action(self, config: dict, variables: dict) -> dict:
+        """Generic action node"""
+        action_type = config.get("config", {}).get("action_type", "log")
+        
+        if action_type == "log":
+            message = config.get("config", {}).get("message", "Action executed")
+            print(f"[ACTION] {message}")
+            return {"logged": message}
+        
+        return {"action": action_type, "executed": True}
+    
+    async def _execute_condition(self, config: dict, variables: dict) -> dict:
+        """Condition node - evaluates a condition"""
+        condition = config.get("config", {}).get("condition", "true")
+        
+        # Simple condition evaluation (be careful with eval in production!)
+        try:
+            # Replace variable references with actual values
+            for var_name, var_value in variables.items():
+                condition = condition.replace(f"${{{var_name}}}", str(var_value))
+            
+            result = eval(condition, {"__builtins__": {}}, variables)
+            return {"condition": condition, "result": bool(result)}
+        except Exception as e:
+            return {"condition": condition, "result": False, "error": str(e)}
+    
+    async def _execute_loop(self, config: dict, variables: dict) -> dict:
+        """Loop node - iterates over data"""
+        items = config.get("config", {}).get("items", [])
+        if isinstance(items, str) and items.startswith("${"):
+            # Reference to variable
+            var_name = items[2:-1]
+            items = variables.get(var_name, [])
+        
+        return {"items": items, "count": len(items) if isinstance(items, list) else 0}
+    
+    async def _execute_llm(self, config: dict, variables: dict) -> dict:
+        """LLM node - calls AI model"""
+        provider = config.get("provider", "groq")
+        model = config.get("model", "llama-3.3-70b-versatile")
+        prompt = config.get("prompt", "Hello!")
+        temperature = config.get("temperature", 0.7)
+        
+        # Replace variables in prompt
+        for var_name, var_value in variables.items():
+            if isinstance(var_value, dict):
+                var_value = str(var_value)
+            prompt = prompt.replace(f"${{{var_name}}}", str(var_value))
+        
+        provider_config = self.llm_providers.get(provider)
+        if not provider_config:
+            raise ValueError(f"Unknown LLM provider: {provider}")
+        
+        api_key = os.getenv(provider_config["api_key_env"])
+        if not api_key:
+            # Return mock response if no API key
+            return {
+                "provider": provider,
+                "model": model,
+                "response": f"[Mock response - set {provider_config['api_key_env']} for real responses]",
+                "prompt": prompt
+            }
+        
+        # Call LLM API
+        if provider == "gemini":
+            response = await self._call_gemini(api_key, model, prompt, temperature)
+        else:
+            response = await self._call_openai_compatible(
+                provider_config["base_url"],
+                api_key,
+                model,
+                prompt,
+                temperature
+            )
+        
+        return {
+            "provider": provider,
+            "model": model,
+            "response": response,
+            "prompt": prompt
+        }
+    
+    async def _call_openai_compatible(
+        self,
+        base_url: str,
+        api_key: str,
+        model: str,
+        prompt: str,
+        temperature: float
+    ) -> str:
+        """Call OpenAI-compatible API (Groq, OpenRouter, Cerebras)"""
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                f"{base_url}/chat/completions",
+                headers={
+                    "Authorization": f"Bearer {api_key}",
+                    "Content-Type": "application/json"
+                },
+                json={
+                    "model": model,
+                    "messages": [{"role": "user", "content": prompt}],
+                    "temperature": temperature
+                },
+                timeout=60.0
+            )
+            response.raise_for_status()
+            data = response.json()
+            return data["choices"][0]["message"]["content"]
+    
+    async def _call_gemini(
+        self,
+        api_key: str,
+        model: str,
+        prompt: str,
+        temperature: float
+    ) -> str:
+        """Call Gemini API"""
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                f"https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent",
+                params={"key": api_key},
+                json={
+                    "contents": [{"parts": [{"text": prompt}]}],
+                    "generationConfig": {"temperature": temperature}
+                },
+                timeout=60.0
+            )
+            response.raise_for_status()
+            data = response.json()
+            return data["candidates"][0]["content"]["parts"][0]["text"]
+    
+    async def _execute_http(self, config: dict, variables: dict) -> dict:
+        """HTTP request node"""
+        method = config.get("config", {}).get("method", "GET")
+        url = config.get("config", {}).get("url", "")
+        headers = config.get("config", {}).get("headers", {})
+        body = config.get("config", {}).get("body", None)
+        
+        # Replace variables
+        for var_name, var_value in variables.items():
+            url = url.replace(f"${{{var_name}}}", str(var_value))
+        
+        async with httpx.AsyncClient() as client:
+            response = await client.request(
+                method=method,
+                url=url,
+                headers=headers,
+                json=body if body else None,
+                timeout=30.0
+            )
+            
+            try:
+                data = response.json()
+            except:
+                data = response.text
+            
+            return {
+                "status_code": response.status_code,
+                "data": data
+            }
+    
+    async def _execute_code(self, config: dict, variables: dict) -> dict:
+        """Custom code execution node (sandboxed)"""
+        code = config.get("config", {}).get("code", "return {}")
+        
+        # Very basic sandboxed execution
+        # In production, use a proper sandbox like RestrictedPython
+        try:
+            # Create a function from the code
+            local_vars = {"variables": variables, "result": None}
+            exec(f"result = {code}", {"__builtins__": {}}, local_vars)
+            return {"result": local_vars.get("result")}
+        except Exception as e:
+            return {"error": str(e)}
+    
+    async def _execute_transform(self, config: dict, variables: dict) -> dict:
+        """Data transformation node"""
+        transform_type = config.get("config", {}).get("transform", "passthrough")
+        input_data = config.get("config", {}).get("input", None)
+        
+        # Resolve input from variables
+        if isinstance(input_data, str) and input_data.startswith("${"):
+            var_name = input_data[2:-1]
+            input_data = variables.get(var_name)
+        
+        if transform_type == "passthrough":
+            return {"data": input_data}
+        elif transform_type == "json_parse":
+            import json
+            return {"data": json.loads(input_data) if isinstance(input_data, str) else input_data}
+        elif transform_type == "uppercase":
+            return {"data": str(input_data).upper()}
+        elif transform_type == "lowercase":
+            return {"data": str(input_data).lower()}
+        
+        return {"data": input_data}

main.py

@@ -0,0 +1,41 @@
+"""
+Tritan API - AI-Native Workflow Engine
+"""
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from api import workflows, execute, planner
+
+app = FastAPI(
+    title="Tritan API",
+    description="AI-Native Workflow Engine",
+    version="0.1.0"
+)
+
+# CORS for frontend (GitHub Pages)
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # Will restrict to GitHub Pages URL later
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Include routers
+app.include_router(workflows.router, prefix="/api/workflows", tags=["workflows"])
+app.include_router(execute.router, prefix="/api/execute", tags=["execute"])
+app.include_router(planner.router, prefix="/api/planner", tags=["planner"])
+
+
+@app.get("/")
+async def root():
+    return {
+        "name": "Tritan API",
+        "version": "0.1.0",
+        "status": "running",
+        "docs": "/docs"
+    }
+
+
+@app.get("/health")
+async def health():
+    return {"status": "healthy", "version": "0.1.0"}

models/__init__.py

@@ -0,0 +1,8 @@
+# Models package
+from .workflow import Workflow, Node, Edge, NodeType
+from .execution import ExecutionState, ExecutionResult, NodeResult
+
+__all__ = [
+    "Workflow", "Node", "Edge", "NodeType",
+    "ExecutionState", "ExecutionResult", "NodeResult"
+]

models/execution.py

@@ -0,0 +1,63 @@
+"""
+Execution state and result models
+"""
+from pydantic import BaseModel, Field
+from typing import Optional, Any
+from enum import Enum
+from datetime import datetime
+import uuid
+
+
+class ExecutionStatus(str, Enum):
+    """Status of workflow execution"""
+    PENDING = "pending"
+    RUNNING = "running"
+    COMPLETED = "completed"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+
+
+class NodeResult(BaseModel):
+    """Result of executing a single node"""
+    node_id: str
+    status: ExecutionStatus
+    output: Optional[Any] = None
+    error: Optional[str] = None
+    started_at: Optional[str] = None
+    completed_at: Optional[str] = None
+    duration_ms: Optional[float] = None
+
+
+class ExecutionState(BaseModel):
+    """Current state of workflow execution"""
+    execution_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
+    workflow_id: str
+    status: ExecutionStatus = ExecutionStatus.PENDING
+    current_node: Optional[str] = None
+    node_results: dict[str, NodeResult] = Field(default_factory=dict)
+    variables: dict[str, Any] = Field(default_factory=dict)  # Shared state
+    started_at: Optional[str] = None
+    completed_at: Optional[str] = None
+
+
+class ExecutionResult(BaseModel):
+    """Final result of workflow execution"""
+    execution_id: str
+    workflow_id: str
+    status: ExecutionStatus
+    node_results: list[NodeResult]
+    final_output: Optional[Any] = None
+    error: Optional[str] = None
+    total_duration_ms: Optional[float] = None
+    started_at: str
+    completed_at: str
+
+
+class GuardianAlert(BaseModel):
+    """Alert from the Guardian system"""
+    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
+    level: str  # info, warning, error
+    message: str
+    node_id: Optional[str] = None
+    suggestion: Optional[str] = None
+    timestamp: str = Field(default_factory=lambda: datetime.now().isoformat())

models/workflow.py

@@ -0,0 +1,65 @@
+"""
+Workflow data models
+"""
+from pydantic import BaseModel, Field
+from typing import Optional, Any
+from enum import Enum
+import uuid
+
+
+class NodeType(str, Enum):
+    """Types of nodes available in Tritan"""
+    TRIGGER = "trigger"          # Start of workflow
+    ACTION = "action"            # Generic action
+    CONDITION = "condition"      # If/else branching
+    LOOP = "loop"               # Iterate over data
+    LLM = "llm"                 # AI/LLM node
+    HTTP = "http"               # HTTP request
+    CODE = "code"               # Custom code execution
+    TRANSFORM = "transform"     # Data transformation
+
+
+class Position(BaseModel):
+    """Node position on canvas"""
+    x: float
+    y: float
+
+
+class NodeData(BaseModel):
+    """Node configuration data"""
+    label: str
+    description: Optional[str] = None
+    config: dict[str, Any] = Field(default_factory=dict)
+    # For LLM nodes
+    provider: Optional[str] = None  # groq, openrouter, cerebras, gemini
+    model: Optional[str] = None
+    prompt: Optional[str] = None
+    temperature: Optional[float] = 0.7
+
+
+class Node(BaseModel):
+    """A node in the workflow"""
+    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
+    type: NodeType
+    position: Position
+    data: NodeData
+
+
+class Edge(BaseModel):
+    """Connection between nodes"""
+    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
+    source: str          # Source node ID
+    target: str          # Target node ID
+    sourceHandle: Optional[str] = None  # For condition nodes (true/false)
+    targetHandle: Optional[str] = None
+
+
+class Workflow(BaseModel):
+    """A complete workflow definition"""
+    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
+    name: str
+    description: Optional[str] = None
+    nodes: list[Node] = Field(default_factory=list)
+    edges: list[Edge] = Field(default_factory=list)
+    created_at: Optional[str] = None
+    updated_at: Optional[str] = None

requirements.txt

@@ -0,0 +1,6 @@
+fastapi>=0.109.0
+uvicorn[standard]>=0.27.0
+pydantic>=2.5.0
+httpx>=0.26.0
+python-dotenv>=1.0.0
+litellm>=1.16.0