Upload 46 files

app/__init__.py

@@ -0,0 +1 @@
+"""FlowForge API package."""

app/api/routes.py

@@ -0,0 +1,423 @@
+import json
+from typing import Annotated
+
+import httpx
+from fastapi import APIRouter, Depends, HTTPException, Query
+
+from app.core.config import Settings, get_settings
+from app.core.security import CurrentUser, get_current_user
+from app.models.catalog import NodeDefinition, ProjectSummary, TemplateSummary
+from app.models.workflow import (
+    ChatRequest,
+    ChatResponse,
+    CommentRequest,
+    ContractRequest,
+    ContractResponse,
+    CostEstimate,
+    CostEstimateRequest,
+    DeploymentRequest,
+    DeploymentResponse,
+    DependencyImpactRequest,
+    DependencyImpactResponse,
+    DocumentationResponse,
+    EnvironmentPromotionRequest,
+    EnvironmentPromotionResponse,
+    ExportRequest,
+    ExpressionRequest,
+    ExpressionResponse,
+    GenerateWorkflowRequest,
+    GenerateWorkflowResponse,
+    ImportRequest,
+    IntentDriftRequest,
+    IntentDriftResponse,
+    LineageResponse,
+    OptimizationResponse,
+    QualityResponse,
+    ReleasePlanRequest,
+    ReleasePlanResponse,
+    ReplayRequest,
+    RoiRequest,
+    RoiResponse,
+    SaveRequest,
+    SaveResponse,
+    ShareRequest,
+    ShareResponse,
+    SharedWorkflowResponse,
+    SimulationRequest,
+    SimulationResponse,
+    TestWorkflowRequest,
+    TestWorkflowResponse,
+    SelfHealRequest,
+    SelfHealResponse,
+    ValidationResult,
+    VersionSummary,
+    WorkflowComment,
+    WorkflowDiff,
+    WorkflowDiffRequest,
+    WorkflowDocument,
+    WorkflowPackageRequest,
+    WorkflowPackageResponse,
+    WorkflowRequest,
+    WebhookInspectRequest,
+    WebhookInspectResponse,
+)
+from app.services.adapters import adapters
+from app.services.catalog import NODE_CATALOG, TEMPLATES
+from app.services.chat import chat_service
+from app.services.collaboration import CollaborationRepository
+from app.services.deployment import deploy_to_n8n
+from app.services.generator import providers
+from app.services.intelligence import (
+    contract_check,
+    dependency_impact,
+    documentation,
+    inspect_webhook,
+    intent_drift,
+    lineage,
+    package_workflow,
+    promote,
+    quality,
+    release_plan,
+    roi,
+    self_heal,
+)
+from app.services.optimizer import optimizer
+from app.services.operations import (
+    diff_workflows,
+    estimate_cost,
+    replay_from_node,
+    run_test,
+    simulate,
+)
+from app.services.repository import WorkflowRepository
+from app.services.remote import fetch_remote_json
+from app.services.validation import validator
+
+router = APIRouter()
+User = Annotated[CurrentUser, Depends(get_current_user)]
+AppSettings = Annotated[Settings, Depends(get_settings)]
+
+
+@router.post("/generate-workflow", response_model=GenerateWorkflowResponse)
+async def generate_workflow(
+    request: GenerateWorkflowRequest,
+    user: User,
+    settings: AppSettings,
+) -> GenerateWorkflowResponse:
+    provider = providers.get(request.provider or settings.ai_provider)
+    return await provider.generate(request.prompt)
+
+
+@router.post("/validate", response_model=ValidationResult)
+async def validate_workflow(request: WorkflowRequest, user: User) -> ValidationResult:
+    return validator.validate(request.workflow)
+
+
+@router.post("/optimize", response_model=OptimizationResponse)
+async def optimize_workflow(request: WorkflowRequest, user: User) -> OptimizationResponse:
+    return optimizer.optimize(request.workflow)
+
+
+@router.post("/simulate", response_model=SimulationResponse)
+async def simulate_workflow(request: SimulationRequest, user: User) -> SimulationResponse:
+    return simulate(request.workflow, request.input_data)
+
+
+@router.post("/test-workflow", response_model=TestWorkflowResponse)
+async def test_workflow(request: TestWorkflowRequest, user: User) -> TestWorkflowResponse:
+    results = [run_test(request.workflow, case) for case in request.cases]
+    return TestWorkflowResponse(
+        passed=sum(result.passed for result in results),
+        failed=sum(not result.passed for result in results),
+        results=results,
+    )
+
+
+@router.post("/estimate-cost", response_model=CostEstimate)
+async def workflow_cost(request: CostEstimateRequest, user: User) -> CostEstimate:
+    return estimate_cost(request.workflow, request.executions_per_month)
+
+
+@router.post("/diff", response_model=WorkflowDiff)
+async def workflow_diff(request: WorkflowDiffRequest, user: User) -> WorkflowDiff:
+    return diff_workflows(request.before, request.after)
+
+
+@router.post("/lineage", response_model=LineageResponse)
+async def workflow_lineage(request: WorkflowRequest, user: User) -> LineageResponse:
+    return lineage(request.workflow)
+
+
+@router.post("/contract-test", response_model=ContractResponse)
+async def contract_test(request: ContractRequest, user: User) -> ContractResponse:
+    return contract_check(request.sample_data, request.expected_schema)
+
+
+@router.post("/quality", response_model=QualityResponse)
+async def workflow_quality(request: WorkflowRequest, user: User) -> QualityResponse:
+    return quality(request.workflow)
+
+
+@router.post("/intent-drift", response_model=IntentDriftResponse)
+async def workflow_intent_drift(
+    request: IntentDriftRequest, user: User
+) -> IntentDriftResponse:
+    return intent_drift(request.workflow, request.requirement)
+
+
+@router.post("/replay", response_model=SimulationResponse)
+async def replay(request: ReplayRequest, user: User) -> SimulationResponse:
+    return replay_from_node(request.workflow, request.node_id, request.input_data)
+
+
+@router.post("/promote", response_model=EnvironmentPromotionResponse)
+async def promote_environment(
+    request: EnvironmentPromotionRequest, user: User
+) -> EnvironmentPromotionResponse:
+    return promote(request.workflow, request.environment, request.values)
+
+
+@router.post("/release-plan", response_model=ReleasePlanResponse)
+async def create_release_plan(
+    request: ReleasePlanRequest, user: User
+) -> ReleasePlanResponse:
+    return release_plan(request)
+
+
+@router.post("/package", response_model=WorkflowPackageResponse)
+async def create_workflow_package(
+    request: WorkflowPackageRequest, user: User
+) -> WorkflowPackageResponse:
+    return package_workflow(request)
+
+
+@router.post("/documentation", response_model=DocumentationResponse)
+async def generate_documentation(
+    request: WorkflowRequest, user: User
+) -> DocumentationResponse:
+    return documentation(request.workflow)
+
+
+@router.post("/roi", response_model=RoiResponse)
+async def calculate_roi(request: RoiRequest, user: User) -> RoiResponse:
+    return roi(
+        request.workflow,
+        request.executions_per_month,
+        request.minutes_saved_per_execution,
+        request.hourly_rate_usd,
+        request.sla_minutes,
+    )
+
+
+@router.post("/inspect-webhook", response_model=WebhookInspectResponse)
+async def inspect_webhook_payload(
+    request: WebhookInspectRequest, user: User
+) -> WebhookInspectResponse:
+    return inspect_webhook(request.payload, request.redact)
+
+
+@router.post("/dependency-impact", response_model=DependencyImpactResponse)
+async def analyze_dependency_impact(
+    request: DependencyImpactRequest, user: User
+) -> DependencyImpactResponse:
+    return dependency_impact(request.workflow, request.dependency)
+
+
+@router.post("/self-heal", response_model=SelfHealResponse)
+async def propose_self_heal(request: SelfHealRequest, user: User) -> SelfHealResponse:
+    return self_heal(request.workflow, request.errors)
+
+
+@router.post("/chat", response_model=ChatResponse)
+async def chat(request: ChatRequest, user: User) -> ChatResponse:
+    return await chat_service.respond(request.message, request.workflow)
+
+
+@router.post("/generate-expression", response_model=ExpressionResponse)
+async def generate_expression(request: ExpressionRequest, user: User) -> ExpressionResponse:
+    description = request.description.lower()
+    if "email" in description:
+        expression = "={{ $json.email }}"
+    elif "name" in description:
+        expression = "={{ $json.name }}"
+    elif "current" in description and ("date" in description or "time" in description):
+        expression = "={{ $now }}"
+    elif "index" in description:
+        expression = "={{ $itemIndex }}"
+    elif request.context.get("node"):
+        node_name = str(request.context["node"])
+        expression = f'={{{{ $node["{node_name}"].json.data }}}}'
+    else:
+        expression = "={{ $json }}"
+    return ExpressionResponse(
+        expression=expression,
+        explanation="The expression reads data from the current n8n item at execution time.",
+        alternatives=["={{ $json }}", "={{ $itemIndex }}", "={{ $now }}"],
+    )
+
+
+@router.post("/import", response_model=WorkflowDocument)
+async def import_workflow(request: ImportRequest, user: User) -> WorkflowDocument:
+    content = request.content
+    if request.source in {"url", "github"}:
+        try:
+            content = await fetch_remote_json(
+                request.content,
+                github_only=request.source == "github",
+            )
+        except (ValueError, httpx.HTTPError, UnicodeDecodeError) as exc:
+            raise HTTPException(status_code=422, detail=f"Remote import failed: {exc}") from exc
+    adapter = adapters.importers.get(
+        "json" if request.source in {"url", "github"} else request.source
+    )
+    if not adapter:
+        raise HTTPException(status_code=400, detail="Unsupported import source")
+    try:
+        return adapter.load(content)
+    except (json.JSONDecodeError, ValueError, TypeError) as exc:
+        raise HTTPException(status_code=422, detail=f"Invalid workflow: {exc}") from exc
+
+
+@router.post("/export")
+async def export_workflow(request: ExportRequest, user: User):
+    if request.format == "internal":
+        return request.workflow.model_dump(mode="json")
+    adapter = adapters.exporters.get(request.format)
+    if not adapter:
+        raise HTTPException(status_code=400, detail="Unsupported export format")
+    return adapter.dump(request.workflow)
+
+
+@router.get("/templates")
+async def list_templates(
+    user: User,
+    q: str = Query(default="", max_length=200),
+    category: str | None = Query(default=None, max_length=80),
+    limit: int = Query(default=24, ge=1, le=100),
+    offset: int = Query(default=0, ge=0),
+) -> dict[str, list[TemplateSummary] | int]:
+    query = q.lower().strip()
+    items = [
+        template
+        for template in TEMPLATES
+        if (not query or query in f"{template.name} {template.description}".lower())
+        and (not category or template.category.lower() == category.lower())
+    ]
+    return {"items": items[offset : offset + limit], "total": len(items)}
+
+
+@router.get("/nodes")
+async def list_nodes(
+    user: User,
+    q: str = Query(default="", max_length=200),
+    category: str | None = Query(default=None, max_length=80),
+    limit: int = Query(default=100, ge=1, le=500),
+    offset: int = Query(default=0, ge=0),
+) -> dict[str, list[NodeDefinition] | int]:
+    query = q.lower().strip()
+    items = [
+        node
+        for node in NODE_CATALOG
+        if (
+            not query
+            or query in f"{node.displayName} {node.description} {node.type}".lower()
+        )
+        and (not category or node.category.lower() == category.lower())
+    ]
+    return {"items": items[offset : offset + limit], "total": len(items)}
+
+
+@router.get("/projects", response_model=list[ProjectSummary])
+async def list_projects(user: User, settings: AppSettings) -> list[ProjectSummary]:
+    return await WorkflowRepository(settings).projects(user.id)
+
+
+@router.post("/save", response_model=SaveResponse)
+async def save_workflow(
+    request: SaveRequest,
+    user: User,
+    settings: AppSettings,
+) -> SaveResponse:
+    try:
+        return await WorkflowRepository(settings).save(request, user.id)
+    except ValueError as exc:
+        raise HTTPException(status_code=403, detail=str(exc)) from exc
+
+
+@router.post("/shares", response_model=ShareResponse)
+async def create_share(
+    request: ShareRequest, user: User, settings: AppSettings
+) -> ShareResponse:
+    try:
+        return await CollaborationRepository(settings).create_share(request, user.id)
+    except ValueError as exc:
+        raise HTTPException(status_code=403, detail=str(exc)) from exc
+
+
+@router.get("/shares/{token}", response_model=SharedWorkflowResponse)
+async def get_shared_workflow(token: str, settings: AppSettings) -> SharedWorkflowResponse:
+    if len(token) < 32 or len(token) > 128:
+        raise HTTPException(status_code=404, detail="Share link was not found")
+    try:
+        return await CollaborationRepository(settings).shared_workflow(token)
+    except ValueError as exc:
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+
+
+@router.get("/workflows/{workflow_id}/versions", response_model=list[VersionSummary])
+async def list_versions(
+    workflow_id: str, user: User, settings: AppSettings
+) -> list[VersionSummary]:
+    try:
+        return await CollaborationRepository(settings).versions(workflow_id, user.id)
+    except ValueError as exc:
+        raise HTTPException(status_code=403, detail=str(exc)) from exc
+
+
+@router.post("/workflows/{workflow_id}/versions/{version}/restore", response_model=WorkflowDocument)
+async def restore_version(
+    workflow_id: str, version: int, user: User, settings: AppSettings
+) -> WorkflowDocument:
+    try:
+        return await CollaborationRepository(settings).restore(workflow_id, version, user.id)
+    except ValueError as exc:
+        raise HTTPException(status_code=403, detail=str(exc)) from exc
+
+
+@router.get("/workflows/{workflow_id}/comments", response_model=list[WorkflowComment])
+async def list_comments(
+    workflow_id: str, user: User, settings: AppSettings
+) -> list[WorkflowComment]:
+    try:
+        return await CollaborationRepository(settings).comments(workflow_id, user.id)
+    except ValueError as exc:
+        raise HTTPException(status_code=403, detail=str(exc)) from exc
+
+
+@router.post("/comments", response_model=WorkflowComment)
+async def add_comment(
+    request: CommentRequest, user: User, settings: AppSettings
+) -> WorkflowComment:
+    try:
+        return await CollaborationRepository(settings).add_comment(
+            request.workflow_id, request.body, request.node_id, user.id
+        )
+    except ValueError as exc:
+        raise HTTPException(status_code=403, detail=str(exc)) from exc
+
+
+@router.post("/deploy", response_model=DeploymentResponse)
+async def deploy(
+    request: DeploymentRequest, user: User, settings: AppSettings
+) -> DeploymentResponse:
+    try:
+        authorized = False
+        if request.workflow.id:
+            authorized = await CollaborationRepository(settings).authorize_deploy(
+                request.workflow.id, user.id
+            )
+        return await deploy_to_n8n(
+            request.workflow, request.activate, settings, authorized=authorized
+        )
+    except (ValueError, httpx.HTTPError) as exc:
+        raise HTTPException(status_code=422, detail=f"Deployment failed: {exc}") from exc

app/core/config.py

@@ -0,0 +1,36 @@
+from functools import lru_cache
+from typing import Literal
+
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    app_name: str = "FlowForge API"
+    environment: Literal["development", "test", "production"] = "development"
+    frontend_url: str = "http://localhost:3000"
+    supabase_url: str = ""
+    supabase_anon_key: str = ""
+    supabase_service_role_key: str = ""
+    supabase_jwt_audience: str = "authenticated"
+    ai_provider: str = "mock"
+    openai_api_key: str = ""
+    n8n_base_url: str = ""
+    n8n_api_key: str = ""
+    rate_limit_per_minute: int = Field(default=120, ge=10, le=10_000)
+    auth_required: bool = False
+
+    model_config = SettingsConfigDict(
+        env_file=(".env", "../../.env"),
+        env_file_encoding="utf-8",
+        extra="ignore",
+    )
+
+    @property
+    def allowed_origins(self) -> list[str]:
+        return [origin.strip() for origin in self.frontend_url.split(",") if origin.strip()]
+
+
+@lru_cache
+def get_settings() -> Settings:
+    return Settings()

app/core/rate_limit.py

@@ -0,0 +1,45 @@
+import asyncio
+import time
+from collections import defaultdict, deque
+
+from fastapi import Request
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.responses import JSONResponse
+
+
+class RateLimitMiddleware(BaseHTTPMiddleware):
+    def __init__(self, app, requests_per_minute: int = 120):
+        super().__init__(app)
+        self.requests_per_minute = requests_per_minute
+        self.requests: dict[str, deque[float]] = defaultdict(deque)
+        self.lock = asyncio.Lock()
+
+    async def dispatch(self, request: Request, call_next):
+        if request.url.path in {"/health", "/docs", "/openapi.json"}:
+            return await call_next(request)
+
+        forwarded = request.headers.get("x-forwarded-for", "")
+        client_id = forwarded.split(",")[0].strip() or (
+            request.client.host if request.client else "unknown"
+        )
+        now = time.monotonic()
+
+        async with self.lock:
+            window = self.requests[client_id]
+            while window and now - window[0] > 60:
+                window.popleft()
+            if len(window) >= self.requests_per_minute:
+                retry_after = max(1, int(60 - (now - window[0])))
+                return JSONResponse(
+                    status_code=429,
+                    content={"detail": "Rate limit exceeded"},
+                    headers={"Retry-After": str(retry_after)},
+                )
+            window.append(now)
+
+        response = await call_next(request)
+        response.headers["X-RateLimit-Limit"] = str(self.requests_per_minute)
+        response.headers["X-RateLimit-Remaining"] = str(
+            max(0, self.requests_per_minute - len(self.requests[client_id]))
+        )
+        return response

app/core/security.py

@@ -0,0 +1,65 @@
+from dataclasses import dataclass
+from typing import Annotated
+
+import jwt
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+
+from app.core.config import Settings, get_settings
+
+bearer = HTTPBearer(auto_error=False)
+
+
+@dataclass(slots=True)
+class CurrentUser:
+    id: str
+    email: str | None = None
+    role: str = "authenticated"
+
+
+async def get_current_user(
+    credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer)],
+    settings: Annotated[Settings, Depends(get_settings)],
+) -> CurrentUser:
+    if credentials is None:
+        if settings.auth_required:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Authentication required",
+            )
+        return CurrentUser(id="00000000-0000-0000-0000-000000000000", role="anonymous")
+
+    try:
+        if not settings.supabase_url:
+            if settings.environment == "production":
+                raise ValueError("SUPABASE_URL is not configured")
+            claims = jwt.decode(
+                credentials.credentials,
+                options={"verify_signature": False},
+                algorithms=["HS256", "RS256"],
+            )
+        else:
+            jwks_client = jwt.PyJWKClient(
+                f"{settings.supabase_url.rstrip('/')}/auth/v1/.well-known/jwks.json",
+                cache_jwk_set=True,
+                lifespan=3600,
+            )
+            signing_key = jwks_client.get_signing_key_from_jwt(credentials.credentials)
+            claims = jwt.decode(
+                credentials.credentials,
+                signing_key.key,
+                algorithms=["ES256", "RS256"],
+                audience=settings.supabase_jwt_audience,
+                options={"require": ["exp", "sub"]},
+            )
+    except (jwt.PyJWTError, ValueError) as exc:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid or expired access token",
+        ) from exc
+
+    return CurrentUser(
+        id=str(claims["sub"]),
+        email=claims.get("email"),
+        role=claims.get("role", "authenticated"),
+    )

app/main.py

@@ -0,0 +1,36 @@
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+from app.api.routes import router
+from app.core.config import get_settings
+from app.core.rate_limit import RateLimitMiddleware
+
+settings = get_settings()
+
+app = FastAPI(
+    title=settings.app_name,
+    description=(
+        "Generate, validate, optimize, import, export, and persist n8n-compatible workflows."
+    ),
+    version="0.1.0",
+    docs_url="/docs",
+    redoc_url="/redoc",
+    openapi_url="/openapi.json",
+)
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=settings.allowed_origins,
+    allow_credentials=True,
+    allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
+    allow_headers=["Authorization", "Content-Type", "X-Request-ID"],
+)
+app.add_middleware(
+    RateLimitMiddleware,
+    requests_per_minute=settings.rate_limit_per_minute,
+)
+app.include_router(router)
+
+
+@app.get("/health", tags=["System"])
+async def health() -> dict[str, str]:
+    return {"status": "ok", "service": settings.app_name, "version": "0.1.0"}

app/models/catalog.py

@@ -0,0 +1,34 @@
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+
+class NodeDefinition(BaseModel):
+    type: str
+    displayName: str
+    description: str
+    category: str
+    icon: str
+    color: str
+    typeVersion: float = 1
+    defaults: dict[str, Any] = Field(default_factory=dict)
+    credentials: list[str] = Field(default_factory=list)
+    community: bool = False
+
+
+class TemplateSummary(BaseModel):
+    id: str
+    name: str
+    description: str
+    category: str
+    node_count: int
+    use_count: int = 0
+    tags: list[str] = Field(default_factory=list)
+
+
+class ProjectSummary(BaseModel):
+    id: str
+    name: str
+    description: str | None = None
+    workflow_count: int = 0
+    updated_at: str

app/models/workflow.py

@@ -0,0 +1,461 @@
+from datetime import datetime
+from typing import Any, Literal
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+
+class Position(BaseModel):
+    x: float
+    y: float
+
+
+class CredentialPlaceholder(BaseModel):
+    id: str = "__CREDENTIAL_ID__"
+    name: str
+    type: str
+
+
+class WorkflowNodeData(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    label: str = Field(min_length=1, max_length=128)
+    type: str = Field(min_length=1, max_length=256)
+    typeVersion: float = Field(default=1, ge=0)
+    category: Literal[
+        "trigger", "core", "ai", "database", "communication", "cloud", "developer"
+    ]
+    subtitle: str | None = Field(default=None, max_length=256)
+    parameters: dict[str, Any] = Field(default_factory=dict)
+    credentials: dict[str, CredentialPlaceholder] | None = None
+    disabled: bool = False
+    issues: int = 0
+
+
+class WorkflowNode(BaseModel):
+    id: str = Field(min_length=1, max_length=128)
+    type: str = "workflow"
+    position: Position
+    data: WorkflowNodeData
+    selected: bool | None = None
+
+
+class WorkflowEdge(BaseModel):
+    id: str = Field(min_length=1, max_length=256)
+    source: str
+    target: str
+    type: str = "smoothstep"
+    animated: bool = False
+    sourceHandle: str | None = None
+    targetHandle: str | None = None
+
+
+class WorkflowMeta(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    description: str | None = None
+    generatedBy: str | None = None
+    version: int | None = None
+    tags: list[str] = Field(default_factory=list)
+
+
+class WorkflowDocument(BaseModel):
+    id: str | None = None
+    name: str = Field(min_length=1, max_length=160)
+    active: bool = False
+    nodes: list[WorkflowNode] = Field(default_factory=list, max_length=1000)
+    edges: list[WorkflowEdge] = Field(default_factory=list, max_length=5000)
+    settings: dict[str, Any] = Field(default_factory=dict)
+    meta: WorkflowMeta = Field(default_factory=WorkflowMeta)
+    pinData: dict[str, Any] = Field(default_factory=dict)
+
+    @field_validator("nodes")
+    @classmethod
+    def unique_node_ids(cls, nodes: list[WorkflowNode]) -> list[WorkflowNode]:
+        ids = [node.id for node in nodes]
+        if len(ids) != len(set(ids)):
+            raise ValueError("Node IDs must be unique")
+        return nodes
+
+    @field_validator("edges")
+    @classmethod
+    def unique_edge_ids(cls, edges: list[WorkflowEdge]) -> list[WorkflowEdge]:
+        ids = [edge.id for edge in edges]
+        if len(ids) != len(set(ids)):
+            raise ValueError("Edge IDs must be unique")
+        return edges
+
+
+class GenerateWorkflowRequest(BaseModel):
+    prompt: str = Field(min_length=10, max_length=20_000)
+    provider: str | None = None
+    model: str | None = None
+
+
+class GenerateWorkflowResponse(BaseModel):
+    workflow: WorkflowDocument
+    explanation: str
+    warnings: list[str] = Field(default_factory=list)
+
+
+class WorkflowRequest(BaseModel):
+    workflow: WorkflowDocument
+
+
+class ValidationIssue(BaseModel):
+    code: str
+    severity: Literal["error", "warning", "info"]
+    message: str
+    nodeId: str | None = None
+    suggestion: str | None = None
+
+
+class ValidationResult(BaseModel):
+    valid: bool
+    score: int = Field(ge=0, le=100)
+    issues: list[ValidationIssue]
+
+
+class OptimizationSuggestion(BaseModel):
+    title: str
+    description: str
+    impact: Literal["low", "medium", "high"]
+    nodeIds: list[str] = Field(default_factory=list)
+
+
+class OptimizationResponse(BaseModel):
+    workflow: WorkflowDocument
+    suggestions: list[OptimizationSuggestion]
+
+
+class ChatRequest(BaseModel):
+    message: str = Field(min_length=1, max_length=20_000)
+    workflow: WorkflowDocument
+    conversation_id: str | None = None
+
+
+class ChatResponse(BaseModel):
+    message: str
+    workflow: WorkflowDocument | None = None
+    actions: list[str] = Field(default_factory=list)
+
+
+class ExpressionRequest(BaseModel):
+    description: str = Field(min_length=2, max_length=4000)
+    context: dict[str, Any] = Field(default_factory=dict)
+
+
+class ExpressionResponse(BaseModel):
+    expression: str
+    explanation: str
+    alternatives: list[str] = Field(default_factory=list)
+
+
+class ImportRequest(BaseModel):
+    content: str = Field(min_length=2, max_length=5_000_000)
+    source: Literal["json", "clipboard", "url", "github"] = "json"
+
+
+class ExportRequest(BaseModel):
+    workflow: WorkflowDocument
+    format: Literal["n8n", "internal"] = "n8n"
+
+
+class SaveRequest(BaseModel):
+    workflow: WorkflowDocument
+    project_id: str | None = None
+    change_summary: str = Field(default="Manual save", max_length=500)
+
+
+class SaveResponse(BaseModel):
+    id: str
+    version: int
+    saved_at: str
+
+
+class SimulationRequest(BaseModel):
+    workflow: WorkflowDocument
+    input_data: dict[str, Any] = Field(default_factory=dict)
+
+
+class NodeRunResult(BaseModel):
+    node_id: str
+    node_name: str
+    status: Literal["success", "skipped", "error"]
+    duration_ms: int = Field(ge=0)
+    input_data: dict[str, Any] = Field(default_factory=dict)
+    output_data: dict[str, Any] = Field(default_factory=dict)
+    error: str | None = None
+
+
+class SimulationResponse(BaseModel):
+    status: Literal["success", "error"]
+    duration_ms: int = Field(ge=0)
+    trace: list[NodeRunResult]
+    output_data: dict[str, Any] = Field(default_factory=dict)
+    warnings: list[str] = Field(default_factory=list)
+
+
+class TestAssertion(BaseModel):
+    path: str = Field(min_length=1, max_length=500)
+    operator: Literal["equals", "not_equals", "exists", "contains"] = "equals"
+    expected: Any = None
+
+
+class WorkflowTestCase(BaseModel):
+    name: str = Field(min_length=1, max_length=160)
+    input_data: dict[str, Any] = Field(default_factory=dict)
+    assertions: list[TestAssertion] = Field(default_factory=list, max_length=100)
+
+
+class TestWorkflowRequest(BaseModel):
+    workflow: WorkflowDocument
+    cases: list[WorkflowTestCase] = Field(min_length=1, max_length=100)
+
+
+class TestCaseResult(BaseModel):
+    name: str
+    passed: bool
+    failures: list[str] = Field(default_factory=list)
+    duration_ms: int = Field(ge=0)
+
+
+class TestWorkflowResponse(BaseModel):
+    passed: int
+    failed: int
+    results: list[TestCaseResult]
+
+
+class CostEstimate(BaseModel):
+    executions_per_month: int
+    estimated_api_calls: int
+    estimated_ai_tokens: int
+    estimated_monthly_usd: float
+    assumptions: list[str]
+    rate_limit_warnings: list[str] = Field(default_factory=list)
+
+
+class CostEstimateRequest(BaseModel):
+    workflow: WorkflowDocument
+    executions_per_month: int = Field(default=1000, ge=1, le=100_000_000)
+
+
+class WorkflowDiffRequest(BaseModel):
+    before: WorkflowDocument
+    after: WorkflowDocument
+
+
+class WorkflowDiff(BaseModel):
+    added_nodes: list[str] = Field(default_factory=list)
+    removed_nodes: list[str] = Field(default_factory=list)
+    modified_nodes: list[str] = Field(default_factory=list)
+    moved_nodes: list[str] = Field(default_factory=list)
+    added_edges: int = 0
+    removed_edges: int = 0
+
+
+class ShareRequest(BaseModel):
+    workflow_id: str
+    workflow: WorkflowDocument | None = None
+    permission: Literal["view", "copy"] = "view"
+    expires_in_days: int | None = Field(default=30, ge=1, le=365)
+
+
+class ShareResponse(BaseModel):
+    id: str
+    url: str
+    permission: Literal["view", "copy"]
+    expires_at: datetime | None = None
+
+
+class SharedWorkflowResponse(BaseModel):
+    workflow: WorkflowDocument
+    permission: Literal["view", "copy"]
+    expires_at: datetime | None = None
+
+
+class VersionSummary(BaseModel):
+    id: str
+    version: int
+    change_summary: str | None = None
+    created_at: str
+    created_by: str
+
+
+class RestoreVersionRequest(BaseModel):
+    version: int = Field(ge=1)
+
+
+class CommentRequest(BaseModel):
+    workflow_id: str
+    body: str = Field(min_length=1, max_length=10_000)
+    node_id: str | None = Field(default=None, max_length=128)
+
+
+class WorkflowComment(BaseModel):
+    id: str
+    workflow_id: str
+    user_id: str
+    node_id: str | None = None
+    body: str
+    resolved_at: str | None = None
+    created_at: str
+
+
+class DeploymentRequest(BaseModel):
+    workflow: WorkflowDocument
+    activate: bool = False
+
+
+class DeploymentResponse(BaseModel):
+    status: Literal["deployed", "preview"]
+    remote_workflow_id: str | None = None
+    message: str
+
+
+class LineageField(BaseModel):
+    field: str
+    source_nodes: list[str] = Field(default_factory=list)
+    consumer_nodes: list[str] = Field(default_factory=list)
+    classification: Literal["public", "internal", "personal", "financial", "secret"]
+
+
+class LineageResponse(BaseModel):
+    fields: list[LineageField]
+    node_dependencies: dict[str, list[str]]
+    sensitive_paths: list[str] = Field(default_factory=list)
+
+
+class ContractRequest(BaseModel):
+    workflow: WorkflowDocument
+    sample_data: dict[str, Any] = Field(default_factory=dict)
+    expected_schema: dict[str, Literal["string", "number", "boolean", "object", "array", "null"]]
+
+
+class ContractResponse(BaseModel):
+    valid: bool
+    inferred_schema: dict[str, str]
+    violations: list[str] = Field(default_factory=list)
+
+
+class QualityResponse(BaseModel):
+    overall: int = Field(ge=0, le=100)
+    scores: dict[str, int]
+    findings: list[ValidationIssue]
+
+
+class IntentDriftRequest(BaseModel):
+    workflow: WorkflowDocument
+    requirement: str = Field(min_length=10, max_length=20_000)
+
+
+class IntentDriftResponse(BaseModel):
+    alignment_score: int = Field(ge=0, le=100)
+    covered_terms: list[str]
+    missing_terms: list[str]
+
+
+class ReplayRequest(BaseModel):
+    workflow: WorkflowDocument
+    node_id: str = Field(min_length=1, max_length=128)
+    input_data: dict[str, Any] = Field(default_factory=dict)
+
+
+class EnvironmentPromotionRequest(BaseModel):
+    workflow: WorkflowDocument
+    environment: Literal["development", "staging", "production"]
+    values: dict[str, str | int | float | bool] = Field(default_factory=dict)
+
+
+class EnvironmentPromotionResponse(BaseModel):
+    workflow: WorkflowDocument
+    environment: str
+    replacements: int
+    unresolved: list[str] = Field(default_factory=list)
+
+
+class ReleasePlanRequest(BaseModel):
+    workflow: WorkflowDocument
+    strategy: Literal["shadow", "canary", "synthetic"]
+    traffic_percentage: int = Field(default=10, ge=0, le=100)
+    success_threshold: float = Field(default=0.99, ge=0, le=1)
+    max_error_rate: float = Field(default=0.02, ge=0, le=1)
+
+
+class ReleasePlanResponse(BaseModel):
+    strategy: str
+    status: Literal["draft", "blocked"]
+    requires_approval: bool = True
+    steps: list[str]
+    rollback_conditions: list[str]
+    warnings: list[str] = Field(default_factory=list)
+
+
+class WorkflowPackageRequest(BaseModel):
+    workflow: WorkflowDocument
+    tests: list[WorkflowTestCase] = Field(default_factory=list)
+    contracts: dict[str, Any] = Field(default_factory=dict)
+    environments: dict[str, dict[str, Any]] = Field(default_factory=dict)
+
+
+class WorkflowPackageResponse(BaseModel):
+    manifest: dict[str, Any]
+    workflow: WorkflowDocument
+    tests: list[WorkflowTestCase]
+    contracts: dict[str, Any]
+    environments: dict[str, dict[str, Any]]
+
+
+class DocumentationResponse(BaseModel):
+    markdown: str
+
+
+class RoiRequest(BaseModel):
+    workflow: WorkflowDocument
+    executions_per_month: int = Field(default=1000, ge=1, le=100_000_000)
+    minutes_saved_per_execution: float = Field(default=5, ge=0, le=100_000)
+    hourly_rate_usd: float = Field(default=30, ge=0, le=100_000)
+    sla_minutes: float = Field(default=60, gt=0, le=1_000_000)
+
+
+class RoiResponse(BaseModel):
+    hours_saved: float
+    labor_value_usd: float
+    estimated_operating_cost_usd: float
+    net_value_usd: float
+    estimated_duration_ms: int
+    sla_headroom_percent: float
+
+
+class WebhookInspectRequest(BaseModel):
+    payload: dict[str, Any]
+    redact: bool = True
+
+
+class WebhookInspectResponse(BaseModel):
+    payload: dict[str, Any]
+    schema_map: dict[str, str]
+    redacted_fields: list[str]
+
+
+class DependencyImpactRequest(BaseModel):
+    workflow: WorkflowDocument
+    dependency: str = Field(min_length=1, max_length=500)
+
+
+class DependencyImpactResponse(BaseModel):
+    affected_nodes: list[str]
+    downstream_nodes: list[str]
+    severity: Literal["none", "low", "medium", "high"]
+
+
+class SelfHealRequest(BaseModel):
+    workflow: WorkflowDocument
+    errors: list[str] = Field(default_factory=list, max_length=100)
+
+
+class SelfHealResponse(BaseModel):
+    proposed_workflow: WorkflowDocument
+    changes: list[str]
+    quality_before: int
+    quality_after: int
+    requires_approval: bool = True

app/services/adapters.py

@@ -0,0 +1,175 @@
+import json
+from typing import Any, Protocol
+
+from app.models.workflow import (
+    CredentialPlaceholder,
+    Position,
+    WorkflowDocument,
+    WorkflowEdge,
+    WorkflowMeta,
+    WorkflowNode,
+    WorkflowNodeData,
+)
+
+
+def infer_category(node_type: str) -> str:
+    lowered = node_type.lower()
+    if "trigger" in lowered or lowered.endswith(".webhook"):
+        return "trigger"
+    if "langchain" in lowered or "openai" in lowered or "agent" in lowered:
+        return "ai"
+    if any(term in lowered for term in ("postgres", "mysql", "supabase", "mongo", "redis")):
+        return "database"
+    if any(term in lowered for term in ("slack", "gmail", "email", "discord", "teams")):
+        return "communication"
+    if any(term in lowered for term in ("code", "function", "graphql")):
+        return "developer"
+    return "core"
+
+
+class ImportAdapter(Protocol):
+    def load(self, content: str) -> WorkflowDocument: ...
+
+
+class ExportAdapter(Protocol):
+    def dump(self, workflow: WorkflowDocument) -> dict[str, Any]: ...
+
+
+class N8nJsonAdapter:
+    def load(self, content: str) -> WorkflowDocument:
+        payload = json.loads(content)
+        if not isinstance(payload, dict):
+            raise ValueError("n8n workflow must be a JSON object")
+        raw_nodes = payload.get("nodes", [])
+        labels_to_ids = {
+            node.get("name", node.get("id", f"node-{index}")): node.get(
+                "id", f"node-{index}"
+            )
+            for index, node in enumerate(raw_nodes)
+        }
+        nodes = []
+        for index, node in enumerate(raw_nodes):
+            node_type = node.get("type", "n8n-nodes-base.noOp")
+            position = node.get("position", [80 + index * 300, 200])
+            credentials = {
+                key: CredentialPlaceholder(
+                    id=str(value.get("id", "__CREDENTIAL_ID__")),
+                    name=str(value.get("name", f"Connect {key}")),
+                    type=key,
+                )
+                for key, value in (node.get("credentials") or {}).items()
+            }
+            nodes.append(
+                WorkflowNode(
+                    id=str(node.get("id", f"node-{index}")),
+                    position=Position(x=position[0], y=position[1]),
+                    data=WorkflowNodeData(
+                        label=node.get("name", f"Node {index + 1}"),
+                        type=node_type,
+                        typeVersion=node.get("typeVersion", 1),
+                        category=infer_category(node_type),  # type: ignore[arg-type]
+                        parameters=node.get("parameters", {}),
+                        credentials=credentials or None,
+                        disabled=node.get("disabled", False),
+                    ),
+                )
+            )
+
+        edges: list[WorkflowEdge] = []
+        for source_name, output_groups in payload.get("connections", {}).items():
+            source_id = labels_to_ids.get(source_name)
+            if not source_id:
+                continue
+            for output_type, indexes in output_groups.items():
+                for output_index, connections in enumerate(indexes):
+                    for connection_index, connection in enumerate(connections):
+                        target_id = labels_to_ids.get(connection.get("node"))
+                        if not target_id:
+                            continue
+                        edges.append(
+                            WorkflowEdge(
+                                id=(
+                                    f"{source_id}-{target_id}-{output_type}-"
+                                    f"{output_index}-{connection_index}"
+                                ),
+                                source=source_id,
+                                target=target_id,
+                            )
+                        )
+
+        return WorkflowDocument(
+            id=str(payload["id"]) if payload.get("id") else None,
+            name=payload.get("name", "Imported workflow"),
+            active=payload.get("active", False),
+            nodes=nodes,
+            edges=edges,
+            settings=payload.get("settings", {}),
+            meta=WorkflowMeta(
+                description=(payload.get("meta") or {}).get("description"),
+                generatedBy="Imported from n8n",
+                tags=[
+                    tag.get("name", str(tag)) if isinstance(tag, dict) else str(tag)
+                    for tag in payload.get("tags", [])
+                ],
+            ),
+            pinData=payload.get("pinData", {}),
+        )
+
+    def dump(self, workflow: WorkflowDocument) -> dict[str, Any]:
+        connections: dict[str, dict[str, list[list[dict[str, Any]]]]] = {}
+        nodes_by_id = {node.id: node for node in workflow.nodes}
+        for edge in workflow.edges:
+            source = nodes_by_id.get(edge.source)
+            target = nodes_by_id.get(edge.target)
+            if not source or not target:
+                continue
+            outputs = connections.setdefault(source.data.label, {"main": [[]]})
+            outputs["main"][0].append(
+                {"node": target.data.label, "type": "main", "index": 0}
+            )
+
+        return {
+            "name": workflow.name,
+            "active": workflow.active,
+            "nodes": [
+                {
+                    "id": node.id,
+                    "name": node.data.label,
+                    "type": node.data.type,
+                    "typeVersion": node.data.typeVersion,
+                    "position": [round(node.position.x), round(node.position.y)],
+                    "parameters": node.data.parameters,
+                    "credentials": {
+                        key: {"id": value.id, "name": value.name}
+                        for key, value in (node.data.credentials or {}).items()
+                    },
+                    **({"disabled": True} if node.data.disabled else {}),
+                }
+                for node in workflow.nodes
+            ],
+            "connections": connections,
+            "settings": workflow.settings,
+            "staticData": None,
+            "meta": workflow.meta.model_dump(exclude_none=True),
+            "pinData": workflow.pinData,
+            "tags": [{"name": tag} for tag in workflow.meta.tags],
+        }
+
+
+class AdapterRegistry:
+    def __init__(self) -> None:
+        self.importers: dict[str, ImportAdapter] = {}
+        self.exporters: dict[str, ExportAdapter] = {}
+
+    def register_importer(self, name: str, adapter: ImportAdapter) -> None:
+        self.importers[name] = adapter
+
+    def register_exporter(self, name: str, adapter: ExportAdapter) -> None:
+        self.exporters[name] = adapter
+
+
+adapters = AdapterRegistry()
+n8n_adapter = N8nJsonAdapter()
+adapters.register_importer("json", n8n_adapter)
+adapters.register_importer("clipboard", n8n_adapter)
+adapters.register_exporter("n8n", n8n_adapter)

app/services/catalog.py

@@ -0,0 +1,170 @@
+from app.models.catalog import NodeDefinition, TemplateSummary
+
+
+NODE_CATALOG = [
+    NodeDefinition(
+        type="n8n-nodes-base.webhook",
+        displayName="Webhook",
+        description="Start a workflow from an HTTP request",
+        category="trigger",
+        icon="Webhook",
+        color="#e6573d",
+        typeVersion=2,
+        defaults={"httpMethod": "POST", "path": ""},
+    ),
+    NodeDefinition(
+        type="n8n-nodes-base.gmailTrigger",
+        displayName="Gmail Trigger",
+        description="Run when a Gmail message arrives",
+        category="trigger",
+        icon="Mail",
+        color="#ea4335",
+        typeVersion=1.2,
+        defaults={"filters": {"hasAttachment": True}},
+        credentials=["gmailOAuth2"],
+    ),
+    NodeDefinition(
+        type="n8n-nodes-base.scheduleTrigger",
+        displayName="Schedule Trigger",
+        description="Run on a schedule",
+        category="trigger",
+        icon="Clock3",
+        color="#5a67d8",
+        typeVersion=1.2,
+        defaults={"rule": {"interval": [{"field": "hours", "hoursInterval": 1}]}},
+    ),
+    NodeDefinition(
+        type="n8n-nodes-base.httpRequest",
+        displayName="HTTP Request",
+        description="Call any REST API",
+        category="core",
+        icon="Globe2",
+        color="#3a7afe",
+        typeVersion=4.2,
+        defaults={"method": "GET", "url": ""},
+    ),
+    NodeDefinition(
+        type="n8n-nodes-base.code",
+        displayName="Code",
+        description="Transform data with JavaScript or Python",
+        category="developer",
+        icon="Code2",
+        color="#313a4d",
+        typeVersion=2,
+        defaults={"mode": "runOnceForAllItems", "jsCode": "return $input.all();"},
+    ),
+    NodeDefinition(
+        type="n8n-nodes-base.if",
+        displayName="If",
+        description="Branch based on conditions",
+        category="core",
+        icon="GitBranch",
+        color="#8b5cf6",
+        typeVersion=2.2,
+        defaults={"conditions": {"options": {}, "conditions": []}},
+    ),
+    NodeDefinition(
+        type="n8n-nodes-base.set",
+        displayName="Edit Fields",
+        description="Set or transform item fields",
+        category="core",
+        icon="Braces",
+        color="#65809b",
+        typeVersion=3.4,
+        defaults={"assignments": {"assignments": []}},
+    ),
+    NodeDefinition(
+        type="n8n-nodes-base.slack",
+        displayName="Slack",
+        description="Send messages and manage Slack resources",
+        category="communication",
+        icon="MessageSquare",
+        color="#4a154b",
+        typeVersion=2.2,
+        defaults={"resource": "message", "operation": "send"},
+        credentials=["slackOAuth2Api"],
+    ),
+    NodeDefinition(
+        type="n8n-nodes-base.supabase",
+        displayName="Supabase",
+        description="Read and write Supabase records",
+        category="database",
+        icon="Database",
+        color="#3ecf8e",
+        defaults={"operation": "getAll", "tableId": ""},
+        credentials=["supabaseApi"],
+    ),
+    NodeDefinition(
+        type="@n8n/n8n-nodes-langchain.openAi",
+        displayName="OpenAI",
+        description="Use OpenAI models in AI workflows",
+        category="ai",
+        icon="Sparkles",
+        color="#10a37f",
+        typeVersion=1.8,
+        defaults={"modelId": "gpt-4.1-mini"},
+        credentials=["openAiApi"],
+    ),
+    NodeDefinition(
+        type="@n8n/n8n-nodes-langchain.agent",
+        displayName="AI Agent",
+        description="Build a tool-using AI agent",
+        category="ai",
+        icon="Bot",
+        color="#f97316",
+        typeVersion=1.7,
+        defaults={"promptType": "define", "text": "={{ $json.chatInput }}"},
+    ),
+    NodeDefinition(
+        type="n8n-nodes-base.postgres",
+        displayName="Postgres",
+        description="Execute queries against PostgreSQL",
+        category="database",
+        icon="Cylinder",
+        color="#336791",
+        typeVersion=2.6,
+        defaults={"operation": "executeQuery", "query": ""},
+        credentials=["postgres"],
+    ),
+]
+
+SUPPORTED_NODE_TYPES = {node.type for node in NODE_CATALOG}
+
+TEMPLATES = [
+    TemplateSummary(
+        id="invoice-intake",
+        name="AI invoice intake",
+        description="Extract invoice data from Gmail attachments and store it in Supabase.",
+        category="Finance",
+        node_count=6,
+        use_count=1842,
+        tags=["AI", "Gmail", "Supabase", "Slack"],
+    ),
+    TemplateSummary(
+        id="lead-enrichment",
+        name="Lead enrichment pipeline",
+        description="Enrich new CRM leads and route qualified accounts to sales.",
+        category="CRM",
+        node_count=9,
+        use_count=1270,
+        tags=["CRM", "AI", "Sales"],
+    ),
+    TemplateSummary(
+        id="support-triage",
+        name="Support ticket triage",
+        description="Classify, prioritize, and assign incoming support tickets.",
+        category="AI",
+        node_count=8,
+        use_count=966,
+        tags=["AI", "Support", "Slack"],
+    ),
+    TemplateSummary(
+        id="content-repurposing",
+        name="Content repurposing",
+        description="Turn long-form content into scheduled social posts.",
+        category="Social Media",
+        node_count=11,
+        use_count=2213,
+        tags=["Marketing", "AI", "Social"],
+    ),
+]

app/services/chat.py

@@ -0,0 +1,84 @@
+from copy import deepcopy
+
+from app.models.workflow import ChatResponse, WorkflowDocument
+from app.services.optimizer import optimizer
+from app.services.validation import validator
+
+
+class WorkflowChatService:
+    async def respond(self, message: str, workflow: WorkflowDocument) -> ChatResponse:
+        lowered = message.lower()
+
+        if "optimize" in lowered or "performance" in lowered:
+            result = optimizer.optimize(workflow)
+            return ChatResponse(
+                message=(
+                    f"I reviewed the workflow and applied {len(result.suggestions)} "
+                    "deterministic improvements."
+                ),
+                workflow=result.workflow,
+                actions=[suggestion.title for suggestion in result.suggestions],
+            )
+
+        if "valid" in lowered or "fix" in lowered:
+            result = validator.validate(workflow)
+            actions = [
+                issue.suggestion or issue.message
+                for issue in result.issues[:5]
+            ]
+            return ChatResponse(
+                message=(
+                    f"The workflow scores {result.score}/100 with "
+                    f"{len(result.issues)} findings. I listed the highest-priority fixes."
+                ),
+                actions=actions or ["No validation issues found"],
+            )
+
+        if "retry" in lowered:
+            updated = deepcopy(workflow)
+            changed = []
+            for node in updated.nodes:
+                if node.data.type == "n8n-nodes-base.httpRequest":
+                    options = node.data.parameters.setdefault("options", {})
+                    if isinstance(options, dict):
+                        options["retry"] = {"maxTries": 3, "waitBetweenTries": 1000}
+                        changed.append(node.data.label)
+            return ChatResponse(
+                message=(
+                    "I added bounded exponential retry settings to HTTP nodes."
+                    if changed
+                    else "There are no HTTP Request nodes to add retry behavior to."
+                ),
+                workflow=updated if changed else None,
+                actions=[f"Added retries to {name}" for name in changed],
+            )
+
+        if "explain" in lowered:
+            triggers = [
+                node.data.label for node in workflow.nodes if node.data.category == "trigger"
+            ]
+            return ChatResponse(
+                message=(
+                    f"This workflow starts with {', '.join(triggers) or 'no trigger'}, "
+                    f"then routes through {len(workflow.nodes) - len(triggers)} action nodes "
+                    f"using {len(workflow.edges)} connections."
+                ),
+                actions=[
+                    f"{node.data.label}: {node.data.subtitle or node.data.type}"
+                    for node in workflow.nodes[:6]
+                ],
+            )
+
+        return ChatResponse(
+            message=(
+                "I inspected the current workflow. Ask me to validate it, explain a node, "
+                "add retries, or optimize the execution path."
+            ),
+            actions=[
+                "Validation and credential checks are available",
+                "Workflow changes are returned as a reviewable document",
+            ],
+        )
+
+
+chat_service = WorkflowChatService()

app/services/collaboration.py

@@ -0,0 +1,263 @@
+import hashlib
+import secrets
+from datetime import UTC, datetime, timedelta
+from uuid import uuid4
+
+import httpx
+
+from app.core.config import Settings
+from app.models.workflow import (
+    ShareRequest,
+    ShareResponse,
+    SharedWorkflowResponse,
+    VersionSummary,
+    WorkflowComment,
+    WorkflowDocument,
+)
+
+
+_demo_shares: dict[str, SharedWorkflowResponse] = {}
+
+
+class CollaborationRepository:
+    def __init__(self, settings: Settings):
+        self.settings = settings
+
+    @property
+    def configured(self) -> bool:
+        return bool(self.settings.supabase_url and self.settings.supabase_service_role_key)
+
+    @property
+    def base_url(self) -> str:
+        return f"{self.settings.supabase_url.rstrip('/')}/rest/v1"
+
+    def _headers(self, prefer: str = "return=representation") -> dict[str, str]:
+        return {
+            "apikey": self.settings.supabase_service_role_key,
+            "Authorization": f"Bearer {self.settings.supabase_service_role_key}",
+            "Content-Type": "application/json",
+            "Prefer": prefer,
+        }
+
+    async def _workspace_access(
+        self, client: httpx.AsyncClient, user_id: str, *, edit: bool
+    ) -> set[str]:
+        response = await client.get(
+            f"{self.base_url}/workspace_members?select=workspace_id,role&user_id=eq.{user_id}",
+            headers=self._headers(),
+        )
+        response.raise_for_status()
+        allowed = {"owner", "admin", "editor"} if edit else {"owner", "admin", "editor", "viewer"}
+        return {row["workspace_id"] for row in response.json() if row["role"] in allowed}
+
+    async def _authorized_workflow(
+        self,
+        client: httpx.AsyncClient,
+        workflow_id: str,
+        user_id: str,
+        *,
+        edit: bool,
+    ) -> dict:
+        workspace_ids = await self._workspace_access(client, user_id, edit=edit)
+        response = await client.get(
+            f"{self.base_url}/workflows?select=id,workspace_id,definition&id=eq.{workflow_id}&limit=1",
+            headers=self._headers(),
+        )
+        response.raise_for_status()
+        rows = response.json()
+        if not rows or rows[0]["workspace_id"] not in workspace_ids:
+            raise ValueError("Workflow is not accessible by this user")
+        return rows[0]
+
+    async def authorize_deploy(self, workflow_id: str, user_id: str) -> bool:
+        if not self.configured:
+            return False
+        async with httpx.AsyncClient(timeout=10) as client:
+            response = await client.get(
+                (
+                    f"{self.base_url}/workflows?select=workspace_id"
+                    f"&id=eq.{workflow_id}&limit=1"
+                ),
+                headers=self._headers(),
+            )
+            response.raise_for_status()
+            workflows = response.json()
+            if not workflows:
+                raise ValueError("Workflow must be saved before deployment")
+            membership = await client.get(
+                (
+                    f"{self.base_url}/workspace_members?select=role"
+                    f"&workspace_id=eq.{workflows[0]['workspace_id']}"
+                    f"&user_id=eq.{user_id}&limit=1"
+                ),
+                headers=self._headers(),
+            )
+            membership.raise_for_status()
+            rows = membership.json()
+            if not rows or rows[0]["role"] not in {"owner", "admin"}:
+                raise ValueError("Only workspace owners and admins can deploy workflows")
+        return True
+
+    async def create_share(self, request: ShareRequest, user_id: str) -> ShareResponse:
+        token = secrets.token_urlsafe(32)
+        token_hash = hashlib.sha256(token.encode()).hexdigest()
+        expires_at = (
+            datetime.now(UTC) + timedelta(days=request.expires_in_days)
+            if request.expires_in_days
+            else None
+        )
+        share_id = str(uuid4())
+        if not self.configured:
+            if request.workflow is None:
+                raise ValueError("A workflow snapshot is required in local mode")
+            _demo_shares[token_hash] = SharedWorkflowResponse(
+                workflow=request.workflow,
+                permission=request.permission,
+                expires_at=expires_at,
+            )
+        else:
+            async with httpx.AsyncClient(timeout=10) as client:
+                await self._authorized_workflow(
+                    client, request.workflow_id, user_id, edit=True
+                )
+                response = await client.post(
+                    f"{self.base_url}/workflow_shares",
+                    headers=self._headers(),
+                    json={
+                        "id": share_id,
+                        "workflow_id": request.workflow_id,
+                        "created_by": user_id,
+                        "token_hash": token_hash,
+                        "permission": request.permission,
+                        "expires_at": expires_at.isoformat() if expires_at else None,
+                    },
+                )
+                response.raise_for_status()
+        return ShareResponse(
+            id=share_id,
+            url=f"{self.settings.allowed_origins[0].rstrip('/')}/share/{token}",
+            permission=request.permission,
+            expires_at=expires_at,
+        )
+
+    async def shared_workflow(self, token: str) -> SharedWorkflowResponse:
+        token_hash = hashlib.sha256(token.encode()).hexdigest()
+        if not self.configured:
+            shared = _demo_shares.get(token_hash)
+            if not shared:
+                raise ValueError("Share link was not found")
+        else:
+            async with httpx.AsyncClient(timeout=10) as client:
+                response = await client.get(
+                    (
+                        f"{self.base_url}/workflow_shares"
+                        "?select=permission,expires_at,revoked_at,workflow:workflows(definition)"
+                        f"&token_hash=eq.{token_hash}&limit=1"
+                    ),
+                    headers=self._headers(),
+                )
+                response.raise_for_status()
+                rows = response.json()
+                if not rows or rows[0]["revoked_at"]:
+                    raise ValueError("Share link was not found")
+                row = rows[0]
+                shared = SharedWorkflowResponse(
+                    workflow=WorkflowDocument.model_validate(row["workflow"]["definition"]),
+                    permission=row["permission"],
+                    expires_at=row["expires_at"],
+                )
+        if shared.expires_at and shared.expires_at < datetime.now(UTC):
+            raise ValueError("Share link has expired")
+        return shared
+
+    async def versions(self, workflow_id: str, user_id: str) -> list[VersionSummary]:
+        if not self.configured:
+            return []
+        async with httpx.AsyncClient(timeout=10) as client:
+            await self._authorized_workflow(client, workflow_id, user_id, edit=False)
+            response = await client.get(
+                (
+                    f"{self.base_url}/workflow_versions"
+                    "?select=id,version_number,change_summary,created_at,created_by"
+                    f"&workflow_id=eq.{workflow_id}&order=version_number.desc&limit=100"
+                ),
+                headers=self._headers(),
+            )
+            response.raise_for_status()
+        return [
+            VersionSummary(
+                id=row["id"],
+                version=row["version_number"],
+                change_summary=row["change_summary"],
+                created_at=row["created_at"],
+                created_by=row["created_by"],
+            )
+            for row in response.json()
+        ]
+
+    async def restore(self, workflow_id: str, version: int, user_id: str) -> WorkflowDocument:
+        if not self.configured:
+            raise ValueError("Version restore requires Supabase")
+        async with httpx.AsyncClient(timeout=10) as client:
+            await self._authorized_workflow(client, workflow_id, user_id, edit=True)
+            response = await client.get(
+                (
+                    f"{self.base_url}/workflow_versions?select=definition"
+                    f"&workflow_id=eq.{workflow_id}&version_number=eq.{version}&limit=1"
+                ),
+                headers=self._headers(),
+            )
+            response.raise_for_status()
+            rows = response.json()
+            if not rows:
+                raise ValueError("Workflow version was not found")
+            definition = rows[0]["definition"]
+            definition["id"] = workflow_id
+            update = await client.patch(
+                f"{self.base_url}/workflows?id=eq.{workflow_id}",
+                headers=self._headers(),
+                json={"definition": definition, "name": definition["name"]},
+            )
+            update.raise_for_status()
+            history = await client.post(
+                f"{self.base_url}/workflow_versions",
+                headers=self._headers(),
+                json={
+                    "workflow_id": workflow_id,
+                    "created_by": user_id,
+                    "definition": definition,
+                    "change_summary": f"Restored version {version}",
+                },
+            )
+            history.raise_for_status()
+        return WorkflowDocument.model_validate(definition)
+
+    async def comments(self, workflow_id: str, user_id: str) -> list[WorkflowComment]:
+        if not self.configured:
+            return []
+        async with httpx.AsyncClient(timeout=10) as client:
+            await self._authorized_workflow(client, workflow_id, user_id, edit=False)
+            response = await client.get(
+                f"{self.base_url}/workflow_comments?select=*&workflow_id=eq.{workflow_id}&order=created_at.desc",
+                headers=self._headers(),
+            )
+            response.raise_for_status()
+        return [WorkflowComment(**row) for row in response.json()]
+
+    async def add_comment(
+        self, workflow_id: str, body: str, node_id: str | None, user_id: str
+    ) -> WorkflowComment:
+        if not self.configured:
+            return WorkflowComment(
+                id=str(uuid4()), workflow_id=workflow_id, user_id=user_id,
+                node_id=node_id, body=body, created_at=datetime.now(UTC).isoformat()
+            )
+        async with httpx.AsyncClient(timeout=10) as client:
+            await self._authorized_workflow(client, workflow_id, user_id, edit=False)
+            response = await client.post(
+                f"{self.base_url}/workflow_comments",
+                headers=self._headers(),
+                json={"workflow_id": workflow_id, "user_id": user_id, "node_id": node_id, "body": body},
+            )
+            response.raise_for_status()
+        return WorkflowComment(**response.json()[0])

app/services/deployment.py

@@ -0,0 +1,43 @@
+from urllib.parse import urlparse
+
+import httpx
+
+from app.core.config import Settings
+from app.models.workflow import DeploymentResponse, WorkflowDocument
+from app.services.adapters import adapters
+
+
+async def deploy_to_n8n(
+    workflow: WorkflowDocument, activate: bool, settings: Settings, *, authorized: bool
+) -> DeploymentResponse:
+    if not authorized or not settings.n8n_base_url or not settings.n8n_api_key:
+        return DeploymentResponse(
+            status="preview",
+            message="Deployment preview is ready. Configure N8N_BASE_URL and N8N_API_KEY to deploy.",
+        )
+    parsed = urlparse(settings.n8n_base_url)
+    if parsed.scheme not in ({"https"} if settings.environment == "production" else {"http", "https"}):
+        raise ValueError("The configured n8n URL must use HTTPS in production")
+    payload = adapters.exporters["n8n"].dump(workflow)
+    payload.pop("active", None)
+    headers = {"X-N8N-API-KEY": settings.n8n_api_key, "Content-Type": "application/json"}
+    async with httpx.AsyncClient(timeout=20, follow_redirects=False) as client:
+        response = await client.post(
+            f"{settings.n8n_base_url.rstrip('/')}/api/v1/workflows",
+            headers=headers,
+            json=payload,
+        )
+        response.raise_for_status()
+        remote = response.json()
+        remote_id = str(remote.get("id"))
+        if activate:
+            activation = await client.post(
+                f"{settings.n8n_base_url.rstrip('/')}/api/v1/workflows/{remote_id}/activate",
+                headers=headers,
+            )
+            activation.raise_for_status()
+    return DeploymentResponse(
+        status="deployed",
+        remote_workflow_id=remote_id,
+        message="Workflow deployed to the configured n8n instance.",
+    )

app/services/generator.py

@@ -0,0 +1,271 @@
+import re
+from dataclasses import dataclass
+from typing import Protocol
+
+from app.models.workflow import (
+    CredentialPlaceholder,
+    GenerateWorkflowResponse,
+    Position,
+    WorkflowDocument,
+    WorkflowEdge,
+    WorkflowMeta,
+    WorkflowNode,
+    WorkflowNodeData,
+)
+
+
+class WorkflowProvider(Protocol):
+    async def generate(self, prompt: str) -> GenerateWorkflowResponse: ...
+
+
+@dataclass(slots=True)
+class PlannedNode:
+    key: str
+    label: str
+    node_type: str
+    category: str
+    subtitle: str
+    parameters: dict
+    credentials: list[str]
+
+
+def _credential_map(types: list[str], label: str) -> dict[str, CredentialPlaceholder] | None:
+    if not types:
+        return None
+    return {
+        credential_type: CredentialPlaceholder(
+            name=f"Connect {label}", type=credential_type
+        )
+        for credential_type in types
+    }
+
+
+class RuleBasedWorkflowProvider:
+    """CPU-friendly baseline provider used as a fallback and in tests."""
+
+    async def generate(self, prompt: str) -> GenerateWorkflowResponse:
+        text = prompt.lower()
+        plan: list[PlannedNode] = []
+
+        if "gmail" in text or "email" in text:
+            plan.append(
+                PlannedNode(
+                    "gmail-trigger",
+                    "Gmail Trigger",
+                    "n8n-nodes-base.gmailTrigger",
+                    "trigger",
+                    "New matching email",
+                    {"filters": {"hasAttachment": "attachment" in text}},
+                    ["gmailOAuth2"],
+                )
+            )
+        elif any(term in text for term in ("schedule", "daily", "hourly", "weekly")):
+            plan.append(
+                PlannedNode(
+                    "schedule-trigger",
+                    "Schedule Trigger",
+                    "n8n-nodes-base.scheduleTrigger",
+                    "trigger",
+                    "Every hour",
+                    {"rule": {"interval": [{"field": "hours", "hoursInterval": 1}]}},
+                    [],
+                )
+            )
+        else:
+            plan.append(
+                PlannedNode(
+                    "webhook",
+                    "Webhook",
+                    "n8n-nodes-base.webhook",
+                    "trigger",
+                    "POST /flowforge-webhook",
+                    {"httpMethod": "POST", "path": "flowforge-webhook"},
+                    [],
+                )
+            )
+
+        if any(term in text for term in ("extract", "classify", "summar", "invoice", " ai ")):
+            plan.append(
+                PlannedNode(
+                    "ai-extract",
+                    "Extract Structured Data",
+                    "@n8n/n8n-nodes-langchain.informationExtractor",
+                    "ai",
+                    "AI structured output",
+                    {
+                        "text": "={{ $binary.data || $json.text || $json.body }}",
+                        "schemaType": "manual",
+                        "inputSchema": (
+                            '{"invoice_number":"string","vendor":"string",'
+                            '"amount":"number","due_date":"string"}'
+                            if "invoice" in text
+                            else '{"result":"string","confidence":"number"}'
+                        ),
+                    },
+                    ["openAiApi"],
+                )
+            )
+
+        if "http" in text or " api" in text:
+            plan.append(
+                PlannedNode(
+                    "http-request",
+                    "HTTP Request",
+                    "n8n-nodes-base.httpRequest",
+                    "core",
+                    "Call external API",
+                    {
+                        "method": "GET",
+                        "url": "https://api.example.com/resource",
+                        "options": {"timeout": 30000},
+                    },
+                    [],
+                )
+            )
+
+        if "supabase" in text:
+            plan.append(
+                PlannedNode(
+                    "supabase",
+                    "Store in Supabase",
+                    "n8n-nodes-base.supabase",
+                    "database",
+                    "Insert record",
+                    {
+                        "operation": "create",
+                        "tableId": "invoices" if "invoice" in text else "records",
+                        "fieldsUi": {"fieldValues": []},
+                    },
+                    ["supabaseApi"],
+                )
+            )
+        elif "postgres" in text or "database" in text:
+            plan.append(
+                PlannedNode(
+                    "postgres",
+                    "Save to Postgres",
+                    "n8n-nodes-base.postgres",
+                    "database",
+                    "Insert record",
+                    {
+                        "operation": "executeQuery",
+                        "query": "INSERT INTO records (payload) VALUES ($1)",
+                        "options": {"queryReplacement": "={{ [$json] }}"},
+                    },
+                    ["postgres"],
+                )
+            )
+
+        if "slack" in text or "notification" in text or "notify" in text:
+            plan.append(
+                PlannedNode(
+                    "slack",
+                    "Send Slack Notification",
+                    "n8n-nodes-base.slack",
+                    "communication",
+                    "#automation",
+                    {
+                        "resource": "message",
+                        "operation": "send",
+                        "channel": "#automation",
+                        "text": "=Workflow completed for {{ $json.invoice_number || $json.id }}",
+                    },
+                    ["slackOAuth2Api"],
+                )
+            )
+
+        if len(plan) == 1:
+            plan.append(
+                PlannedNode(
+                    "edit-fields",
+                    "Prepare Output",
+                    "n8n-nodes-base.set",
+                    "core",
+                    "Normalize response",
+                    {
+                        "assignments": {
+                            "assignments": [
+                                {
+                                    "name": "status",
+                                    "value": "completed",
+                                    "type": "string",
+                                }
+                            ]
+                        }
+                    },
+                    [],
+                )
+            )
+
+        nodes = [
+            WorkflowNode(
+                id=item.key,
+                position=Position(x=80 + index * 315, y=190 + (index % 2) * 45),
+                data=WorkflowNodeData(
+                    label=item.label,
+                    type=item.node_type,
+                    typeVersion=1,
+                    category=item.category,  # type: ignore[arg-type]
+                    subtitle=item.subtitle,
+                    parameters=item.parameters,
+                    credentials=_credential_map(item.credentials, item.label),
+                ),
+            )
+            for index, item in enumerate(plan)
+        ]
+        edges = [
+            WorkflowEdge(
+                id=f"{source.key}-{target.key}",
+                source=source.key,
+                target=target.key,
+                animated=index == 0,
+            )
+            for index, (source, target) in enumerate(zip(plan, plan[1:]))
+        ]
+
+        title_words = re.findall(r"[a-zA-Z0-9]+", prompt)[:7]
+        title = " ".join(title_words).strip().capitalize() or "Generated workflow"
+        if len(title) > 62:
+            title = f"{title[:59]}..."
+
+        workflow = WorkflowDocument(
+            name=title,
+            nodes=nodes,
+            edges=edges,
+            settings={
+                "executionOrder": "v1",
+                "saveManualExecutions": True,
+                "saveExecutionProgress": True,
+                "errorWorkflow": "",
+                "timezone": "UTC",
+            },
+            meta=WorkflowMeta(
+                description=prompt[:500],
+                generatedBy="FlowForge rule-based provider",
+                version=1,
+                tags=["AI generated"],
+            ),
+        )
+        return GenerateWorkflowResponse(
+            workflow=workflow,
+            explanation=f"Created a {len(nodes)}-node workflow with a trigger and connected actions.",
+            warnings=[
+                "Replace credential placeholders before activating the workflow.",
+                "Review generated expressions with representative execution data.",
+            ],
+        )
+
+
+class ProviderRegistry:
+    def __init__(self) -> None:
+        self._providers: dict[str, WorkflowProvider] = {}
+
+    def register(self, name: str, provider: WorkflowProvider) -> None:
+        self._providers[name] = provider
+
+    def get(self, name: str) -> WorkflowProvider:
+        return self._providers.get(name, self._providers["mock"])
+
+
+providers = ProviderRegistry()
+providers.register("mock", RuleBasedWorkflowProvider())

app/services/intelligence.py

@@ -0,0 +1,300 @@
+import re
+from copy import deepcopy
+from typing import Any
+
+from app.models.workflow import (
+    ContractResponse,
+    DependencyImpactResponse,
+    DocumentationResponse,
+    EnvironmentPromotionResponse,
+    IntentDriftResponse,
+    LineageField,
+    LineageResponse,
+    QualityResponse,
+    ReleasePlanRequest,
+    ReleasePlanResponse,
+    RoiResponse,
+    SelfHealResponse,
+    ValidationIssue,
+    WebhookInspectResponse,
+    WorkflowDocument,
+    WorkflowPackageRequest,
+    WorkflowPackageResponse,
+)
+from app.services.operations import estimate_cost
+from app.services.optimizer import optimizer
+from app.services.validation import validator
+
+
+FIELD_REF = re.compile(r"\$json\.([A-Za-z_][\w.]*)")
+ENV_REF = re.compile(r"\$\{ENV\.([A-Z][A-Z0-9_]*)\}")
+WORD = re.compile(r"[a-z][a-z0-9_-]{2,}")
+SECRET_NAMES = {"password", "secret", "token", "api_key", "apikey", "authorization"}
+PERSONAL_NAMES = {"email", "phone", "address", "name", "ssn", "dob"}
+FINANCIAL_NAMES = {"amount", "invoice", "card", "iban", "account", "price"}
+STOP_WORDS = {"that", "with", "from", "this", "then", "into", "workflow", "create", "build"}
+
+
+def _walk(value: Any, prefix: str = ""):
+    if isinstance(value, dict):
+        for key, nested in value.items():
+            path = f"{prefix}.{key}" if prefix else str(key)
+            yield path, nested
+            yield from _walk(nested, path)
+    elif isinstance(value, list):
+        for index, nested in enumerate(value):
+            yield from _walk(nested, f"{prefix}[{index}]")
+
+
+def _classification(field: str) -> str:
+    parts = {part.lower() for part in re.split(r"[.\[\]_-]+", field) if part}
+    if parts & SECRET_NAMES:
+        return "secret"
+    if parts & FINANCIAL_NAMES:
+        return "financial"
+    if parts & PERSONAL_NAMES:
+        return "personal"
+    return "internal"
+
+
+def lineage(workflow: WorkflowDocument) -> LineageResponse:
+    dependencies: dict[str, list[str]] = {node.id: [] for node in workflow.nodes}
+    fields: dict[str, dict[str, set[str] | str]] = {}
+    labels = {node.id: node.data.label for node in workflow.nodes}
+    for edge in workflow.edges:
+        if edge.target in dependencies and edge.source in labels:
+            dependencies[edge.target].append(labels[edge.source])
+    for node in workflow.nodes:
+        for path, value in _walk(node.data.parameters):
+            candidates = {path}
+            if isinstance(value, str):
+                candidates.update(FIELD_REF.findall(value))
+            for field in candidates:
+                record = fields.setdefault(
+                    field,
+                    {"sources": set(), "consumers": set(), "classification": _classification(field)},
+                )
+                if field == path:
+                    record["sources"].add(node.data.label)  # type: ignore[union-attr]
+                else:
+                    record["consumers"].add(node.data.label)  # type: ignore[union-attr]
+    result = [
+        LineageField(
+            field=field,
+            source_nodes=sorted(record["sources"]),  # type: ignore[arg-type]
+            consumer_nodes=sorted(record["consumers"]),  # type: ignore[arg-type]
+            classification=record["classification"],  # type: ignore[arg-type]
+        )
+        for field, record in sorted(fields.items())
+    ]
+    return LineageResponse(
+        fields=result,
+        node_dependencies=dependencies,
+        sensitive_paths=[field.field for field in result if field.classification in {"secret", "personal", "financial"}],
+    )
+
+
+def _type_name(value: Any) -> str:
+    if value is None:
+        return "null"
+    if isinstance(value, bool):
+        return "boolean"
+    if isinstance(value, (int, float)):
+        return "number"
+    if isinstance(value, dict):
+        return "object"
+    if isinstance(value, list):
+        return "array"
+    return "string"
+
+
+def contract_check(sample: dict[str, Any], expected: dict[str, str]) -> ContractResponse:
+    inferred = {path: _type_name(value) for path, value in _walk(sample) if not isinstance(value, (dict, list))}
+    violations = []
+    for path, expected_type in expected.items():
+        actual = inferred.get(path)
+        if actual is None:
+            violations.append(f"Missing required field: {path}")
+        elif actual != expected_type:
+            violations.append(f"{path} expected {expected_type}, received {actual}")
+    return ContractResponse(valid=not violations, inferred_schema=inferred, violations=violations)
+
+
+def quality(workflow: WorkflowDocument) -> QualityResponse:
+    validation = validator.validate(workflow)
+    findings = list(validation.issues)
+    lineage_result = lineage(workflow)
+    code_nodes = [node for node in workflow.nodes if node.data.type == "n8n-nodes-base.code"]
+    if code_nodes:
+        findings.append(ValidationIssue(code="review_code", severity="warning", message="Code nodes require review and sandboxed testing."))
+    if lineage_result.sensitive_paths:
+        findings.append(ValidationIssue(code="sensitive_data", severity="warning", message="Sensitive fields require destination and retention review."))
+    retry_nodes = sum("retry" in str(node.data.parameters).lower() for node in workflow.nodes)
+    scores = {
+        "reliability": max(0, validation.score - (10 if retry_nodes == 0 else 0)),
+        "maintainability": max(0, 100 - len(code_nodes) * 12 - max(0, len(workflow.nodes) - 30)),
+        "security": max(0, 100 - len(lineage_result.sensitive_paths) * 4 - len(code_nodes) * 8),
+        "cost": max(0, 100 - sum("langchain" in node.data.type for node in workflow.nodes) * 7),
+        "observability": 90 if workflow.settings.get("saveExecutionProgress") else 55,
+    }
+    return QualityResponse(overall=round(sum(scores.values()) / len(scores)), scores=scores, findings=findings)
+
+
+def intent_drift(workflow: WorkflowDocument, requirement: str) -> IntentDriftResponse:
+    terms = {term for term in WORD.findall(requirement.lower()) if term not in STOP_WORDS}
+    workflow_text = " ".join(
+        f"{node.data.label} {node.data.type} {node.data.subtitle or ''}" for node in workflow.nodes
+    ).lower()
+    covered = sorted(term for term in terms if term in workflow_text)
+    missing = sorted(terms - set(covered))
+    return IntentDriftResponse(
+        alignment_score=round(len(covered) / max(1, len(terms)) * 100),
+        covered_terms=covered,
+        missing_terms=missing,
+    )
+
+
+def promote(workflow: WorkflowDocument, environment: str, values: dict[str, Any]) -> EnvironmentPromotionResponse:
+    promoted = deepcopy(workflow)
+    replacements = 0
+    unresolved: set[str] = set()
+
+    def replace(value: Any) -> Any:
+        nonlocal replacements
+        if isinstance(value, str):
+            def substitute(match: re.Match[str]) -> str:
+                nonlocal replacements
+                key = match.group(1)
+                if key not in values:
+                    unresolved.add(key)
+                    return match.group(0)
+                replacements += 1
+                return str(values[key])
+            return ENV_REF.sub(substitute, value)
+        if isinstance(value, dict):
+            return {key: replace(nested) for key, nested in value.items()}
+        if isinstance(value, list):
+            return [replace(nested) for nested in value]
+        return value
+
+    for node in promoted.nodes:
+        node.data.parameters = replace(node.data.parameters)
+    promoted.meta.model_extra["environment"] = environment
+    return EnvironmentPromotionResponse(
+        workflow=promoted, environment=environment, replacements=replacements, unresolved=sorted(unresolved)
+    )
+
+
+def release_plan(request: ReleasePlanRequest) -> ReleasePlanResponse:
+    validation = validator.validate(request.workflow)
+    blocked = not validation.valid
+    strategy_steps = {
+        "shadow": ["Mirror sanitized input to the candidate version", "Suppress side-effect nodes", "Compare schemas, outputs, and latency"],
+        "canary": [f"Route {request.traffic_percentage}% of eligible traffic to the candidate", "Monitor success rate and latency", "Increase traffic only after approval"],
+        "synthetic": ["Schedule representative test inputs", "Verify credentials and contracts", "Alert on consecutive failures"],
+    }
+    return ReleasePlanResponse(
+        strategy=request.strategy,
+        status="blocked" if blocked else "draft",
+        steps=strategy_steps[request.strategy],
+        rollback_conditions=[
+            f"Error rate exceeds {request.max_error_rate:.1%}",
+            f"Success rate drops below {request.success_threshold:.1%}",
+            "Output contract or sensitive-data policy fails",
+        ],
+        warnings=[issue.message for issue in validation.issues if issue.severity == "error"],
+    )
+
+
+def package_workflow(request: WorkflowPackageRequest) -> WorkflowPackageResponse:
+    return WorkflowPackageResponse(
+        manifest={
+            "format": "flowforge.workflow-package/v1",
+            "name": request.workflow.name,
+            "workflowVersion": request.workflow.meta.version or 1,
+            "n8nCompatible": True,
+            "contents": ["workflow", "tests", "contracts", "environments"],
+        },
+        workflow=request.workflow,
+        tests=request.tests,
+        contracts=request.contracts,
+        environments=request.environments,
+    )
+
+
+def documentation(workflow: WorkflowDocument) -> DocumentationResponse:
+    nodes = "\n".join(f"- **{node.data.label}** (`{node.data.type}`): {node.data.subtitle or 'Configured action'}" for node in workflow.nodes)
+    connections = "\n".join(
+        f"- {next((n.data.label for n in workflow.nodes if n.id == edge.source), edge.source)} -> "
+        f"{next((n.data.label for n in workflow.nodes if n.id == edge.target), edge.target)}"
+        for edge in workflow.edges
+    )
+    return DocumentationResponse(markdown=f"# {workflow.name}\n\n{workflow.meta.description or 'n8n workflow'}\n\n## Nodes\n\n{nodes}\n\n## Connections\n\n{connections or '- None'}\n")
+
+
+def roi(workflow: WorkflowDocument, executions: int, minutes_saved: float, hourly_rate: float, sla_minutes: float) -> RoiResponse:
+    hours = executions * minutes_saved / 60
+    cost = estimate_cost(workflow, executions).estimated_monthly_usd
+    estimated_duration = max(1, len(workflow.nodes) * 250)
+    return RoiResponse(
+        hours_saved=round(hours, 2), labor_value_usd=round(hours * hourly_rate, 2),
+        estimated_operating_cost_usd=cost, net_value_usd=round(hours * hourly_rate - cost, 2),
+        estimated_duration_ms=estimated_duration,
+        sla_headroom_percent=round(max(0, 1 - estimated_duration / (sla_minutes * 60_000)) * 100, 2),
+    )
+
+
+def inspect_webhook(payload: dict[str, Any], redact: bool) -> WebhookInspectResponse:
+    result = deepcopy(payload)
+    schema = {path: _type_name(value) for path, value in _walk(payload) if not isinstance(value, (dict, list))}
+    redacted = []
+    if redact:
+        def scrub(value: Any, prefix: str = "") -> None:
+            if not isinstance(value, dict):
+                return
+            for key, nested in value.items():
+                path = f"{prefix}.{key}" if prefix else key
+                if _classification(path) in {"secret", "personal", "financial"}:
+                    value[key] = "[REDACTED]"
+                    redacted.append(path)
+                else:
+                    scrub(nested, path)
+        scrub(result)
+    return WebhookInspectResponse(payload=result, schema_map=schema, redacted_fields=redacted)
+
+
+def dependency_impact(workflow: WorkflowDocument, dependency: str) -> DependencyImpactResponse:
+    query = dependency.lower()
+    affected = [node for node in workflow.nodes if query in f"{node.data.type} {node.data.label} {node.data.parameters}".lower()]
+    affected_ids = {node.id for node in affected}
+    downstream_ids: set[str] = set()
+    frontier = list(affected_ids)
+    while frontier:
+        source = frontier.pop()
+        for edge in workflow.edges:
+            if edge.source == source and edge.target not in downstream_ids:
+                downstream_ids.add(edge.target)
+                frontier.append(edge.target)
+    labels = {node.id: node.data.label for node in workflow.nodes}
+    count = len(affected_ids | downstream_ids)
+    severity = "none" if count == 0 else "low" if count == 1 else "medium" if count < 5 else "high"
+    return DependencyImpactResponse(
+        affected_nodes=[node.data.label for node in affected],
+        downstream_nodes=[labels[node_id] for node_id in downstream_ids if node_id in labels],
+        severity=severity,
+    )
+
+
+def self_heal(workflow: WorkflowDocument, errors: list[str]) -> SelfHealResponse:
+    before = quality(workflow)
+    optimized = optimizer.optimize(workflow)
+    after = quality(optimized.workflow)
+    changes = [suggestion.title for suggestion in optimized.suggestions]
+    if errors:
+        changes.append("Attached execution errors as approval evidence")
+    return SelfHealResponse(
+        proposed_workflow=optimized.workflow,
+        changes=changes,
+        quality_before=before.overall,
+        quality_after=after.overall,
+    )

app/services/operations.py

@@ -0,0 +1,230 @@
+import json
+import re
+from time import perf_counter
+from typing import Any
+
+from app.models.workflow import (
+    CostEstimate,
+    NodeRunResult,
+    SimulationResponse,
+    TestAssertion,
+    TestCaseResult,
+    WorkflowDiff,
+    WorkflowDocument,
+    WorkflowTestCase,
+)
+
+
+EXPRESSION = re.compile(r"^=\{\{\s*\$json(?:\.([\w.]+))?\s*\}\}$")
+
+
+def _get_path(value: Any, path: str) -> Any:
+    current = value
+    for part in path.removeprefix("$.").split("."):
+        if not part:
+            continue
+        if isinstance(current, dict) and part in current:
+            current = current[part]
+        else:
+            return None
+    return current
+
+
+def _resolve(value: Any, item: dict[str, Any]) -> Any:
+    if isinstance(value, str):
+        match = EXPRESSION.match(value)
+        if match:
+            return _get_path(item, match.group(1) or "")
+        return value
+    if isinstance(value, list):
+        return [_resolve(entry, item) for entry in value]
+    if isinstance(value, dict):
+        return {key: _resolve(entry, item) for key, entry in value.items()}
+    return value
+
+
+def _ordered_nodes(workflow: WorkflowDocument):
+    nodes = {node.id: node for node in workflow.nodes}
+    indegree = {node_id: 0 for node_id in nodes}
+    outgoing: dict[str, list[str]] = {node_id: [] for node_id in nodes}
+    for edge in workflow.edges:
+        if edge.source in nodes and edge.target in nodes:
+            outgoing[edge.source].append(edge.target)
+            indegree[edge.target] += 1
+    queue = [node_id for node_id, degree in indegree.items() if degree == 0]
+    ordered = []
+    while queue:
+        node_id = queue.pop(0)
+        ordered.append(nodes[node_id])
+        for target in outgoing[node_id]:
+            indegree[target] -= 1
+            if indegree[target] == 0:
+                queue.append(target)
+    if len(ordered) != len(nodes):
+        raise ValueError("Workflow contains a cycle and cannot be simulated")
+    return ordered
+
+
+def simulate(workflow: WorkflowDocument, input_data: dict[str, Any]) -> SimulationResponse:
+    started = perf_counter()
+    item = dict(input_data)
+    trace: list[NodeRunResult] = []
+    warnings = [
+        "Simulation does not call external services or execute Code nodes; outputs are deterministic previews."
+    ]
+    try:
+        ordered = _ordered_nodes(workflow)
+    except ValueError as exc:
+        return SimulationResponse(
+            status="error", duration_ms=0, trace=[], output_data=item, warnings=[str(exc)]
+        )
+
+    for node in ordered:
+        node_started = perf_counter()
+        incoming = dict(item)
+        parameters = _resolve(node.data.parameters, incoming)
+        if node.data.disabled:
+            status = "skipped"
+            output = incoming
+        elif node.data.type == "n8n-nodes-base.set":
+            output = {**incoming, "parameters": parameters}
+            status = "success"
+        elif node.data.type == "n8n-nodes-base.code":
+            output = {**incoming, "_simulation": "Code execution skipped"}
+            status = "skipped"
+        else:
+            output = {
+                **incoming,
+                "_lastNode": node.data.label,
+                "_parameters": parameters,
+            }
+            status = "success"
+        duration = max(1, round((perf_counter() - node_started) * 1000))
+        trace.append(
+            NodeRunResult(
+                node_id=node.id,
+                node_name=node.data.label,
+                status=status,
+                duration_ms=duration,
+                input_data=incoming,
+                output_data=output,
+            )
+        )
+        item = output
+
+    return SimulationResponse(
+        status="success",
+        duration_ms=max(1, round((perf_counter() - started) * 1000)),
+        trace=trace,
+        output_data=item,
+        warnings=warnings,
+    )
+
+
+def replay_from_node(
+    workflow: WorkflowDocument, node_id: str, input_data: dict[str, Any]
+) -> SimulationResponse:
+    ids = {node.id for node in workflow.nodes}
+    if node_id not in ids:
+        return SimulationResponse(
+            status="error",
+            duration_ms=0,
+            trace=[],
+            output_data=input_data,
+            warnings=["Replay node was not found"],
+        )
+    reachable = {node_id}
+    frontier = [node_id]
+    while frontier:
+        source = frontier.pop()
+        for edge in workflow.edges:
+            if edge.source == source and edge.target not in reachable:
+                reachable.add(edge.target)
+                frontier.append(edge.target)
+    replay = workflow.model_copy(deep=True)
+    replay.nodes = [node for node in replay.nodes if node.id in reachable]
+    replay.edges = [
+        edge for edge in replay.edges if edge.source in reachable and edge.target in reachable
+    ]
+    result = simulate(replay, input_data)
+    result.warnings.append("Replay starts from captured input and suppresses real external side effects.")
+    return result
+
+
+def _assert(assertion: TestAssertion, output: dict[str, Any]) -> str | None:
+    actual = _get_path(output, assertion.path)
+    if assertion.operator == "exists":
+        passed = actual is not None
+    elif assertion.operator == "equals":
+        passed = actual == assertion.expected
+    elif assertion.operator == "not_equals":
+        passed = actual != assertion.expected
+    else:
+        passed = assertion.expected in actual if isinstance(actual, (str, list, dict)) else False
+    if passed:
+        return None
+    return f"{assertion.path} {assertion.operator} {json.dumps(assertion.expected)} (actual: {json.dumps(actual)})"
+
+
+def run_test(workflow: WorkflowDocument, case: WorkflowTestCase) -> TestCaseResult:
+    result = simulate(workflow, case.input_data)
+    failures = [_assert(assertion, result.output_data) for assertion in case.assertions]
+    failures = [failure for failure in failures if failure]
+    if result.status == "error":
+        failures.extend(result.warnings)
+    return TestCaseResult(
+        name=case.name,
+        passed=not failures,
+        failures=failures,
+        duration_ms=result.duration_ms,
+    )
+
+
+def estimate_cost(workflow: WorkflowDocument, executions: int) -> CostEstimate:
+    ai_nodes = sum("langchain" in node.data.type.lower() or "openai" in node.data.type.lower() for node in workflow.nodes)
+    api_nodes = sum(
+        node.data.type in {"n8n-nodes-base.httpRequest", "n8n-nodes-base.slack", "n8n-nodes-base.gmail"}
+        for node in workflow.nodes
+    )
+    ai_tokens = executions * ai_nodes * 1500
+    api_calls = executions * api_nodes
+    warnings = []
+    if executions > 100_000 and api_nodes:
+        warnings.append("High request volume: add batching, backoff, and provider-specific concurrency limits.")
+    return CostEstimate(
+        executions_per_month=executions,
+        estimated_api_calls=api_calls,
+        estimated_ai_tokens=ai_tokens,
+        estimated_monthly_usd=round(ai_tokens / 1_000_000 * 0.75, 2),
+        assumptions=[
+            "AI nodes average 1,500 tokens per execution.",
+            "External service nodes make one request per execution.",
+            "Estimate excludes n8n hosting and third-party subscription fees.",
+        ],
+        rate_limit_warnings=warnings,
+    )
+
+
+def diff_workflows(before: WorkflowDocument, after: WorkflowDocument) -> WorkflowDiff:
+    old = {node.id: node for node in before.nodes}
+    new = {node.id: node for node in after.nodes}
+    shared = old.keys() & new.keys()
+    old_edges = {(edge.source, edge.target, edge.sourceHandle, edge.targetHandle) for edge in before.edges}
+    new_edges = {(edge.source, edge.target, edge.sourceHandle, edge.targetHandle) for edge in after.edges}
+    return WorkflowDiff(
+        added_nodes=[new[node_id].data.label for node_id in new.keys() - old.keys()],
+        removed_nodes=[old[node_id].data.label for node_id in old.keys() - new.keys()],
+        modified_nodes=[
+            new[node_id].data.label
+            for node_id in shared
+            if old[node_id].data.model_dump(exclude={"issues"})
+            != new[node_id].data.model_dump(exclude={"issues"})
+        ],
+        moved_nodes=[
+            new[node_id].data.label
+            for node_id in shared
+            if old[node_id].position != new[node_id].position
+        ],
+        added_edges=len(new_edges - old_edges),
+        removed_edges=len(old_edges - new_edges),
+    )

app/services/optimizer.py

@@ -0,0 +1,83 @@
+from copy import deepcopy
+
+from app.models.workflow import (
+    OptimizationResponse,
+    OptimizationSuggestion,
+    WorkflowDocument,
+)
+
+
+class WorkflowOptimizer:
+    def optimize(self, workflow: WorkflowDocument) -> OptimizationResponse:
+        optimized = deepcopy(workflow)
+        suggestions: list[OptimizationSuggestion] = []
+
+        http_nodes = [
+            node for node in optimized.nodes if node.data.type == "n8n-nodes-base.httpRequest"
+        ]
+        for node in http_nodes:
+            options = node.data.parameters.setdefault("options", {})
+            if isinstance(options, dict) and "retry" not in options:
+                options.update({"retry": {"maxTries": 3, "waitBetweenTries": 1000}})
+                suggestions.append(
+                    OptimizationSuggestion(
+                        title="Add API retries",
+                        description=f"Added bounded retry behavior to {node.data.label}.",
+                        impact="high",
+                        nodeIds=[node.id],
+                    )
+                )
+
+        if not optimized.settings.get("saveExecutionProgress"):
+            optimized.settings["saveExecutionProgress"] = True
+            suggestions.append(
+                OptimizationSuggestion(
+                    title="Enable execution recovery",
+                    description="Save execution progress so long-running workflows can recover.",
+                    impact="medium",
+                )
+            )
+
+        linear_pairs = []
+        for edge in optimized.edges:
+            source = next((node for node in optimized.nodes if node.id == edge.source), None)
+            target = next((node for node in optimized.nodes if node.id == edge.target), None)
+            if source and target and source.data.type == target.data.type == "n8n-nodes-base.set":
+                linear_pairs.append((source, target))
+        if linear_pairs:
+            suggestions.append(
+                OptimizationSuggestion(
+                    title="Combine adjacent Edit Fields nodes",
+                    description="Adjacent field transformations can be handled by one node.",
+                    impact="medium",
+                    nodeIds=[node.id for pair in linear_pairs for node in pair],
+                )
+            )
+
+        branches = {}
+        for edge in optimized.edges:
+            branches[edge.source] = branches.get(edge.source, 0) + 1
+        parallel = [node_id for node_id, count in branches.items() if count > 1]
+        if parallel:
+            suggestions.append(
+                OptimizationSuggestion(
+                    title="Review parallel branches",
+                    description="Independent branches can execute concurrently; merge only when required.",
+                    impact="medium",
+                    nodeIds=parallel,
+                )
+            )
+
+        if not suggestions:
+            suggestions.append(
+                OptimizationSuggestion(
+                    title="Workflow is already compact",
+                    description="No deterministic optimization could be applied safely.",
+                    impact="low",
+                )
+            )
+
+        return OptimizationResponse(workflow=optimized, suggestions=suggestions)
+
+
+optimizer = WorkflowOptimizer()

app/services/remote.py

@@ -0,0 +1,66 @@
+import ipaddress
+import socket
+from urllib.parse import urlparse
+
+import httpx
+
+MAX_IMPORT_BYTES = 5_000_000
+ALLOWED_GITHUB_HOSTS = {"github.com", "raw.githubusercontent.com"}
+
+
+def _is_public_host(hostname: str) -> bool:
+    try:
+        addresses = socket.getaddrinfo(hostname, 443, type=socket.SOCK_STREAM)
+    except socket.gaierror:
+        return False
+    for address in addresses:
+        ip = ipaddress.ip_address(address[4][0])
+        if (
+            ip.is_private
+            or ip.is_loopback
+            or ip.is_link_local
+            or ip.is_multicast
+            or ip.is_reserved
+            or ip.is_unspecified
+        ):
+            return False
+    return True
+
+
+def normalize_github_url(url: str) -> str:
+    parsed = urlparse(url)
+    if parsed.hostname == "github.com" and "/blob/" in parsed.path:
+        parts = parsed.path.strip("/").split("/")
+        if len(parts) < 5:
+            raise ValueError("GitHub URL must point to a workflow JSON file")
+        owner, repository, _, branch, *path = parts
+        return (
+            f"https://raw.githubusercontent.com/{owner}/{repository}/"
+            f"{branch}/{'/'.join(path)}"
+        )
+    return url
+
+
+async def fetch_remote_json(url: str, github_only: bool = False) -> str:
+    target = normalize_github_url(url.strip())
+    parsed = urlparse(target)
+    if parsed.scheme != "https" or not parsed.hostname:
+        raise ValueError("Remote imports require an HTTPS URL")
+    if github_only and parsed.hostname not in ALLOWED_GITHUB_HOSTS:
+        raise ValueError("GitHub imports only accept github.com URLs")
+    if not _is_public_host(parsed.hostname):
+        raise ValueError("URL does not resolve to a public host")
+
+    async with httpx.AsyncClient(
+        timeout=httpx.Timeout(15, connect=5),
+        follow_redirects=False,
+        headers={"Accept": "application/json"},
+    ) as client:
+        response = await client.get(target)
+        response.raise_for_status()
+        if int(response.headers.get("content-length", "0") or 0) > MAX_IMPORT_BYTES:
+            raise ValueError("Remote workflow exceeds the 5 MB import limit")
+        content = response.content
+    if len(content) > MAX_IMPORT_BYTES:
+        raise ValueError("Remote workflow exceeds the 5 MB import limit")
+    return content.decode("utf-8")

app/services/repository.py

@@ -0,0 +1,164 @@
+from datetime import UTC, datetime
+from uuid import uuid4
+
+import httpx
+
+from app.core.config import Settings
+from app.models.catalog import ProjectSummary
+from app.models.workflow import SaveRequest, SaveResponse
+
+
+class WorkflowRepository:
+    def __init__(self, settings: Settings):
+        self.settings = settings
+
+    @property
+    def configured(self) -> bool:
+        return bool(self.settings.supabase_url and self.settings.supabase_service_role_key)
+
+    def _headers(self) -> dict[str, str]:
+        return {
+            "apikey": self.settings.supabase_service_role_key,
+            "Authorization": f"Bearer {self.settings.supabase_service_role_key}",
+            "Content-Type": "application/json",
+            "Prefer": "return=representation",
+        }
+
+    async def save(self, request: SaveRequest, user_id: str) -> SaveResponse:
+        now = datetime.now(UTC).isoformat()
+        workflow_id = str(uuid4())
+        if not self.configured:
+            return SaveResponse(id=workflow_id, version=1, saved_at=now)
+
+        base_url = f"{self.settings.supabase_url.rstrip('/')}/rest/v1"
+        async with httpx.AsyncClient(timeout=10) as client:
+            membership_response = await client.get(
+                (
+                    f"{base_url}/workspace_members"
+                    f"?select=workspace_id,role&user_id=eq.{user_id}"
+                ),
+                headers=self._headers(),
+            )
+            membership_response.raise_for_status()
+            memberships = membership_response.json()
+            editable_workspace_ids = {
+                item["workspace_id"]
+                for item in memberships
+                if item["role"] in {"owner", "admin", "editor"}
+            }
+            if not editable_workspace_ids:
+                raise ValueError("User does not have edit access to a workspace")
+
+            owner_id = user_id
+            workspace_id = next(iter(editable_workspace_ids))
+            if request.workflow.id:
+                existing_response = await client.get(
+                    (
+                        f"{base_url}/workflows"
+                        f"?select=id,owner_id,workspace_id&id=eq.{request.workflow.id}&limit=1"
+                    ),
+                    headers=self._headers(),
+                )
+                existing_response.raise_for_status()
+                existing = existing_response.json()
+                if existing:
+                    record = existing[0]
+                    if record["workspace_id"] not in editable_workspace_ids:
+                        raise ValueError("Workflow is not editable by this user")
+                    workflow_id = record["id"]
+                    workspace_id = record["workspace_id"]
+                    owner_id = record["owner_id"]
+
+            if request.project_id:
+                project_response = await client.get(
+                    (
+                        f"{base_url}/projects?select=workspace_id"
+                        f"&id=eq.{request.project_id}&limit=1"
+                    ),
+                    headers=self._headers(),
+                )
+                project_response.raise_for_status()
+                projects = project_response.json()
+                if (
+                    not projects
+                    or projects[0]["workspace_id"] not in editable_workspace_ids
+                    or (
+                        request.workflow.id
+                        and workflow_id == request.workflow.id
+                        and projects[0]["workspace_id"] != workspace_id
+                    )
+                ):
+                    raise ValueError("Project is not editable by this user")
+                workspace_id = projects[0]["workspace_id"]
+
+            definition = request.workflow.model_dump(mode="json")
+            definition["id"] = workflow_id
+            payload = {
+                "id": workflow_id,
+                "owner_id": owner_id,
+                "workspace_id": workspace_id,
+                "project_id": request.project_id,
+                "name": request.workflow.name,
+                "definition": definition,
+                "is_active": request.workflow.active,
+                "updated_at": now,
+            }
+            response = await client.post(
+                f"{base_url}/workflows?on_conflict=id",
+                headers={**self._headers(), "Prefer": "resolution=merge-duplicates,return=representation"},
+                json=payload,
+            )
+            response.raise_for_status()
+            version_response = await client.post(
+                f"{base_url}/workflow_versions",
+                headers=self._headers(),
+                json={
+                    "workflow_id": workflow_id,
+                    "created_by": user_id,
+                    "definition": definition,
+                    "change_summary": request.change_summary,
+                },
+            )
+            version_response.raise_for_status()
+            version_payload = version_response.json()[0]
+        return SaveResponse(
+            id=workflow_id,
+            version=version_payload["version_number"],
+            saved_at=version_payload["created_at"],
+        )
+
+    async def projects(self, user_id: str) -> list[ProjectSummary]:
+        if not self.configured:
+            return [
+                ProjectSummary(
+                    id="demo-finance",
+                    name="Finance Ops",
+                    description="Invoice and reporting automations",
+                    workflow_count=4,
+                    updated_at=datetime.now(UTC).isoformat(),
+                )
+            ]
+        base_url = f"{self.settings.supabase_url.rstrip('/')}/rest/v1"
+        async with httpx.AsyncClient(timeout=10) as client:
+            membership_response = await client.get(
+                f"{base_url}/workspace_members?select=workspace_id&user_id=eq.{user_id}",
+                headers=self._headers(),
+            )
+            membership_response.raise_for_status()
+            workspace_ids = [
+                item["workspace_id"] for item in membership_response.json()
+            ]
+            if not workspace_ids:
+                return []
+            workspace_filter = ",".join(workspace_ids)
+            response = await client.get(
+                (
+                    f"{base_url}/projects"
+                    "?select=id,name,description,updated_at"
+                    f"&workspace_id=in.({workspace_filter})"
+                    "&order=updated_at.desc"
+                ),
+                headers=self._headers(),
+            )
+            response.raise_for_status()
+        return [ProjectSummary(**item, workflow_count=0) for item in response.json()]

app/services/validation.py

@@ -0,0 +1,215 @@
+import re
+from collections import defaultdict
+from typing import Protocol
+
+from app.models.workflow import ValidationIssue, ValidationResult, WorkflowDocument
+from app.services.catalog import SUPPORTED_NODE_TYPES
+
+EXPRESSION_RE = re.compile(r"^=.*\{\{.+\}\}.*$", re.DOTALL)
+
+
+class ValidationRule(Protocol):
+    code: str
+
+    def validate(self, workflow: WorkflowDocument) -> list[ValidationIssue]: ...
+
+
+class TriggerRule:
+    code = "missing_trigger"
+
+    def validate(self, workflow: WorkflowDocument) -> list[ValidationIssue]:
+        if any(node.data.category == "trigger" for node in workflow.nodes):
+            return []
+        return [
+            ValidationIssue(
+                code=self.code,
+                severity="error",
+                message="Workflow does not have a trigger node.",
+                suggestion="Add a Webhook, Schedule Trigger, or service trigger.",
+            )
+        ]
+
+
+class CredentialRule:
+    code = "missing_credentials"
+
+    def validate(self, workflow: WorkflowDocument) -> list[ValidationIssue]:
+        issues = []
+        for node in workflow.nodes:
+            for credential in (node.data.credentials or {}).values():
+                if credential.id.startswith("__"):
+                    issues.append(
+                        ValidationIssue(
+                            code=self.code,
+                            severity="error",
+                            nodeId=node.id,
+                            message=f"{node.data.label} has an unconfigured credential.",
+                            suggestion=f"Select a {credential.type} credential before activation.",
+                        )
+                    )
+        return issues
+
+
+class ConnectionRule:
+    code = "broken_connection"
+
+    def validate(self, workflow: WorkflowDocument) -> list[ValidationIssue]:
+        ids = {node.id for node in workflow.nodes}
+        issues = []
+        for edge in workflow.edges:
+            if edge.source not in ids or edge.target not in ids:
+                issues.append(
+                    ValidationIssue(
+                        code=self.code,
+                        severity="error",
+                        message=f"Connection {edge.id} references a missing node.",
+                        suggestion="Delete the connection or reconnect it to an existing node.",
+                    )
+                )
+        connected = {edge.source for edge in workflow.edges} | {
+            edge.target for edge in workflow.edges
+        }
+        for node in workflow.nodes:
+            if len(workflow.nodes) > 1 and node.id not in connected:
+                issues.append(
+                    ValidationIssue(
+                        code="disconnected_node",
+                        severity="warning",
+                        nodeId=node.id,
+                        message=f"{node.data.label} is not connected.",
+                        suggestion="Connect or remove this node.",
+                    )
+                )
+        return issues
+
+
+class LoopRule:
+    code = "infinite_loop"
+
+    def validate(self, workflow: WorkflowDocument) -> list[ValidationIssue]:
+        graph: dict[str, list[str]] = defaultdict(list)
+        for edge in workflow.edges:
+            graph[edge.source].append(edge.target)
+        visiting: set[str] = set()
+        visited: set[str] = set()
+
+        def has_cycle(node_id: str) -> bool:
+            if node_id in visiting:
+                return True
+            if node_id in visited:
+                return False
+            visiting.add(node_id)
+            if any(has_cycle(target) for target in graph[node_id]):
+                return True
+            visiting.remove(node_id)
+            visited.add(node_id)
+            return False
+
+        if any(has_cycle(node.id) for node in workflow.nodes if node.id not in visited):
+            return [
+                ValidationIssue(
+                    code=self.code,
+                    severity="error",
+                    message="Workflow contains a cycle that may execute indefinitely.",
+                    suggestion="Break the cycle or add an explicit loop termination condition.",
+                )
+            ]
+        return []
+
+
+class ParameterRule:
+    code = "empty_parameters"
+
+    def validate(self, workflow: WorkflowDocument) -> list[ValidationIssue]:
+        issues = []
+        for node in workflow.nodes:
+            if node.data.category != "trigger" and not node.data.parameters:
+                issues.append(
+                    ValidationIssue(
+                        code=self.code,
+                        severity="warning",
+                        nodeId=node.id,
+                        message=f"{node.data.label} has no parameters.",
+                        suggestion="Configure the required operation and resource fields.",
+                    )
+                )
+        return issues
+
+
+class ExpressionRule:
+    code = "invalid_expression"
+
+    def validate(self, workflow: WorkflowDocument) -> list[ValidationIssue]:
+        issues = []
+
+        def inspect(value, node_id: str, label: str) -> None:
+            if isinstance(value, dict):
+                for nested in value.values():
+                    inspect(nested, node_id, label)
+            elif isinstance(value, list):
+                for nested in value:
+                    inspect(nested, node_id, label)
+            elif isinstance(value, str) and ("{{" in value or "}}" in value):
+                if value.count("{{") != value.count("}}") or not EXPRESSION_RE.match(value):
+                    issues.append(
+                        ValidationIssue(
+                            code=self.code,
+                            severity="error",
+                            nodeId=node_id,
+                            message=f"{label} contains an invalid n8n expression.",
+                            suggestion="Expressions should start with = and use balanced {{ }} braces.",
+                        )
+                    )
+
+        for node in workflow.nodes:
+            inspect(node.data.parameters, node.id, node.data.label)
+        return issues
+
+
+class SupportedNodeRule:
+    code = "unsupported_node"
+
+    def validate(self, workflow: WorkflowDocument) -> list[ValidationIssue]:
+        return [
+            ValidationIssue(
+                code=self.code,
+                severity="warning",
+                nodeId=node.id,
+                message=f"{node.data.label} is not in the bundled node catalog.",
+                suggestion="Confirm that the node package is installed on the target n8n instance.",
+            )
+            for node in workflow.nodes
+            if node.data.type not in SUPPORTED_NODE_TYPES
+            and not node.data.type.startswith("@n8n/n8n-nodes-langchain.")
+        ]
+
+
+class Validator:
+    def __init__(self) -> None:
+        self.rules: list[ValidationRule] = []
+
+    def register(self, rule: ValidationRule) -> None:
+        self.rules.append(rule)
+
+    def validate(self, workflow: WorkflowDocument) -> ValidationResult:
+        issues = [issue for rule in self.rules for issue in rule.validate(workflow)]
+        deductions = {"error": 18, "warning": 6, "info": 1}
+        score = max(0, 100 - sum(deductions[issue.severity] for issue in issues))
+        return ValidationResult(
+            valid=not any(issue.severity == "error" for issue in issues),
+            score=score,
+            issues=issues,
+        )
+
+
+validator = Validator()
+for validation_rule in (
+    TriggerRule(),
+    CredentialRule(),
+    ConnectionRule(),
+    LoopRule(),
+    ParameterRule(),
+    ExpressionRule(),
+    SupportedNodeRule(),
+):
+    validator.register(validation_rule)

requirements.txt

@@ -1,3 +1,9 @@
-google-genai==1.5.0
-gradio
-pydantic==2.10.6
+fastapi==0.139.0
+uvicorn[standard]==0.51.0
+pydantic==2.13.4
+pydantic-settings==2.14.2
+httpx==0.28.1
+PyJWT[crypto]==2.10.1
+python-multipart==0.0.20
+pytest==8.3.4
+pytest-asyncio==0.25.0

tests/conftest.py

@@ -0,0 +1,14 @@
+import os
+
+os.environ["ENVIRONMENT"] = "test"
+os.environ["AUTH_REQUIRED"] = "false"
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app.main import app
+
+
+@pytest.fixture
+def client():
+    return TestClient(app)

tests/test_api.py

@@ -0,0 +1,241 @@
+def test_health(client):
+    response = client.get("/health")
+    assert response.status_code == 200
+    assert response.json()["status"] == "ok"
+
+
+def test_generate_and_validate_workflow(client):
+    generated = client.post(
+        "/generate-workflow",
+        json={
+            "prompt": (
+                "Read Gmail invoice attachments, extract the data, store it in "
+                "Supabase, and send a Slack notification"
+            )
+        },
+    )
+    assert generated.status_code == 200
+    workflow = generated.json()["workflow"]
+    assert len(workflow["nodes"]) == 4
+    assert len(workflow["edges"]) == 3
+
+    validation = client.post("/validate", json={"workflow": workflow})
+    assert validation.status_code == 200
+    body = validation.json()
+    assert body["valid"] is False
+    assert any(issue["code"] == "missing_credentials" for issue in body["issues"])
+
+
+def test_cycle_detection(client):
+    workflow = {
+        "name": "Cycle",
+        "active": False,
+        "nodes": [
+            {
+                "id": "a",
+                "type": "workflow",
+                "position": {"x": 0, "y": 0},
+                "data": {
+                    "label": "Webhook",
+                    "type": "n8n-nodes-base.webhook",
+                    "typeVersion": 2,
+                    "category": "trigger",
+                    "parameters": {"path": "test"},
+                },
+            },
+            {
+                "id": "b",
+                "type": "workflow",
+                "position": {"x": 300, "y": 0},
+                "data": {
+                    "label": "Edit Fields",
+                    "type": "n8n-nodes-base.set",
+                    "typeVersion": 3,
+                    "category": "core",
+                    "parameters": {"value": 1},
+                },
+            },
+        ],
+        "edges": [
+            {"id": "a-b", "source": "a", "target": "b"},
+            {"id": "b-a", "source": "b", "target": "a"},
+        ],
+        "settings": {},
+        "meta": {},
+    }
+    response = client.post("/validate", json={"workflow": workflow})
+    assert response.status_code == 200
+    assert any(issue["code"] == "infinite_loop" for issue in response.json()["issues"])
+
+
+def test_n8n_import_export_round_trip(client):
+    n8n = {
+        "name": "Round trip",
+        "nodes": [
+            {
+                "id": "one",
+                "name": "Webhook",
+                "type": "n8n-nodes-base.webhook",
+                "typeVersion": 2,
+                "position": [10, 20],
+                "parameters": {"path": "incoming"},
+            },
+            {
+                "id": "two",
+                "name": "Edit Fields",
+                "type": "n8n-nodes-base.set",
+                "typeVersion": 3,
+                "position": [300, 20],
+                "parameters": {"assignments": {}},
+            },
+        ],
+        "connections": {
+            "Webhook": {
+                "main": [[{"node": "Edit Fields", "type": "main", "index": 0}]]
+            }
+        },
+        "settings": {},
+    }
+    imported = client.post(
+        "/import",
+        json={"content": __import__("json").dumps(n8n), "source": "json"},
+    )
+    assert imported.status_code == 200
+    workflow = imported.json()
+    assert len(workflow["edges"]) == 1
+
+    exported = client.post(
+        "/export", json={"workflow": workflow, "format": "n8n"}
+    )
+    assert exported.status_code == 200
+    assert exported.json()["connections"]["Webhook"]["main"][0][0]["node"] == "Edit Fields"
+
+
+def test_simulation_testing_cost_and_diff(client):
+    generated = client.post(
+        "/generate-workflow",
+        json={"prompt": "Create a webhook workflow that stores data in Supabase"},
+    ).json()["workflow"]
+
+    simulation = client.post(
+        "/simulate",
+        json={"workflow": generated, "input_data": {"email": "a@example.com"}},
+    )
+    assert simulation.status_code == 200
+    assert simulation.json()["status"] == "success"
+    assert len(simulation.json()["trace"]) == len(generated["nodes"])
+
+    expected_last = generated["nodes"][-1]["data"]["label"]
+    tests = client.post(
+        "/test-workflow",
+        json={
+            "workflow": generated,
+            "cases": [
+                {
+                    "name": "happy path",
+                    "input_data": {},
+                    "assertions": [
+                        {"path": "_lastNode", "operator": "equals", "expected": expected_last}
+                    ],
+                }
+            ],
+        },
+    )
+    assert tests.status_code == 200
+    assert tests.json()["passed"] == 1
+
+    cost = client.post(
+        "/estimate-cost",
+        json={"workflow": generated, "executions_per_month": 10_000},
+    )
+    assert cost.status_code == 200
+    assert cost.json()["executions_per_month"] == 10_000
+
+    changed = {**generated, "name": "Changed", "nodes": generated["nodes"][:-1]}
+    diff = client.post("/diff", json={"before": generated, "after": changed})
+    assert diff.status_code == 200
+    assert len(diff.json()["removed_nodes"]) == 1
+
+
+def test_local_share_link_round_trip(client):
+    workflow = client.post(
+        "/generate-workflow",
+        json={"prompt": "Create a webhook workflow that sends a Slack message"},
+    ).json()["workflow"]
+    workflow["id"] = "local-workflow"
+    shared = client.post(
+        "/shares",
+        json={
+            "workflow_id": workflow["id"],
+            "workflow": workflow,
+            "permission": "copy",
+            "expires_in_days": 7,
+        },
+    )
+    assert shared.status_code == 200
+    token = shared.json()["url"].rsplit("/", 1)[-1]
+    public = client.get(f"/shares/{token}")
+    assert public.status_code == 200
+    assert public.json()["permission"] == "copy"
+    assert public.json()["workflow"]["name"] == workflow["name"]
+
+
+def test_intelligence_release_and_package_endpoints(client):
+    workflow = client.post(
+        "/generate-workflow",
+        json={"prompt": "Read customer email invoices and store amounts in Supabase"},
+    ).json()["workflow"]
+
+    quality = client.post("/quality", json={"workflow": workflow})
+    assert quality.status_code == 200
+    assert 0 <= quality.json()["overall"] <= 100
+
+    lineage = client.post("/lineage", json={"workflow": workflow})
+    assert lineage.status_code == 200
+    assert "fields" in lineage.json()
+
+    contract = client.post(
+        "/contract-test",
+        json={
+            "workflow": workflow,
+            "sample_data": {"email": "a@example.com", "amount": 15},
+            "expected_schema": {"email": "string", "amount": "number"},
+        },
+    )
+    assert contract.status_code == 200
+    assert contract.json()["valid"] is True
+
+    release = client.post(
+        "/release-plan",
+        json={"workflow": workflow, "strategy": "shadow", "traffic_percentage": 0},
+    )
+    assert release.status_code == 200
+    assert release.json()["requires_approval"] is True
+
+    package = client.post(
+        "/package",
+        json={"workflow": workflow, "tests": [], "contracts": {}, "environments": {}},
+    )
+    assert package.status_code == 200
+    assert package.json()["manifest"]["format"] == "flowforge.workflow-package/v1"
+
+
+def test_webhook_redaction_and_replay(client):
+    workflow = client.post(
+        "/generate-workflow",
+        json={"prompt": "Create a webhook that sends a notification"},
+    ).json()["workflow"]
+    inspected = client.post(
+        "/inspect-webhook",
+        json={"payload": {"email": "a@example.com", "token": "secret", "event": "paid"}},
+    )
+    assert inspected.status_code == 200
+    assert inspected.json()["payload"]["email"] == "[REDACTED]"
+    assert inspected.json()["payload"]["token"] == "[REDACTED]"
+
+    replay = client.post(
+        "/replay",
+        json={"workflow": workflow, "node_id": workflow["nodes"][-1]["id"], "input_data": {}},
+    )
+    assert replay.status_code == 200
+    assert len(replay.json()["trace"]) == 1