Delete app

app/__init__.py

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

app/api/routes.py

@@ -1,482 +0,0 @@
-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.core.errors import ServiceConfigurationError
-from app.models.catalog import DashboardSummary, 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
-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:
-    try:
-        provider = providers.get(request.provider or settings.ai_provider)
-        return await provider.generate(request.prompt, request.model)
-    except ValueError as exc:
-        raise HTTPException(status_code=422, detail=str(exc)) from exc
-    except (RuntimeError, httpx.HTTPError) as exc:
-        raise HTTPException(status_code=502, detail=f"Workflow generation failed: {exc}") from exc
-
-
-@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,
-    settings: AppSettings,
-    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]:
-    try:
-        items, total = await WorkflowRepository(settings).templates(
-            user.id,
-            query=q.strip(),
-            category=category,
-            limit=limit,
-            offset=offset,
-        )
-        return {"items": items, "total": total}
-    except ServiceConfigurationError as exc:
-        raise HTTPException(status_code=503, detail=str(exc)) from exc
-
-
-@router.get("/templates/{template_id}", response_model=WorkflowDocument)
-async def get_template(
-    template_id: str, user: User, settings: AppSettings
-) -> WorkflowDocument:
-    try:
-        return await WorkflowRepository(settings).template(template_id, user.id)
-    except ServiceConfigurationError as exc:
-        raise HTTPException(status_code=503, detail=str(exc)) from exc
-    except ValueError as exc:
-        raise HTTPException(status_code=404, detail=str(exc)) from exc
-
-
-@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]:
-    try:
-        return await WorkflowRepository(settings).projects(user.id)
-    except ServiceConfigurationError as exc:
-        raise HTTPException(status_code=503, detail=str(exc)) from exc
-
-
-@router.get("/dashboard", response_model=DashboardSummary)
-async def dashboard(user: User, settings: AppSettings) -> DashboardSummary:
-    try:
-        return await WorkflowRepository(settings).dashboard(user.id, user.email)
-    except ServiceConfigurationError as exc:
-        raise HTTPException(status_code=503, detail=str(exc)) from exc
-
-
-@router.get("/workflows/{workflow_id}", response_model=WorkflowDocument)
-async def get_workflow(
-    workflow_id: str, user: User, settings: AppSettings
-) -> WorkflowDocument:
-    try:
-        return await WorkflowRepository(settings).workflow(workflow_id, user.id)
-    except ServiceConfigurationError as exc:
-        raise HTTPException(status_code=503, detail=str(exc)) from exc
-    except ValueError as exc:
-        raise HTTPException(status_code=404, detail=str(exc)) from exc
-
-
-@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 ServiceConfigurationError as exc:
-        raise HTTPException(status_code=503, detail=str(exc)) from exc
-    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 ServiceConfigurationError as exc:
-        raise HTTPException(status_code=503, detail=str(exc)) from exc
-    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 ServiceConfigurationError as exc:
-        raise HTTPException(status_code=503, detail=str(exc)) from exc
-    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 ServiceConfigurationError as exc:
-        raise HTTPException(status_code=503, detail=str(exc)) from exc
-    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 ServiceConfigurationError as exc:
-        raise HTTPException(status_code=503, detail=str(exc)) from exc
-    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 ServiceConfigurationError as exc:
-        raise HTTPException(status_code=503, detail=str(exc)) from exc
-    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 ServiceConfigurationError as exc:
-        raise HTTPException(status_code=503, detail=str(exc)) from exc
-    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

@@ -1,72 +0,0 @@
-from functools import lru_cache
-from typing import Literal
-
-from pydantic import Field, model_validator
-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: Literal["openai", "gemini", "openrouter", "deterministic"] = "openai"
-    openai_api_key: str = ""
-    openai_model: str = Field(
-        default="gpt-5.5", pattern=r"^[A-Za-z0-9][A-Za-z0-9._:/-]*$"
-    )
-    gemini_api_key: str = ""
-    gemini_model: str = Field(
-        default="gemini-2.5-pro", pattern=r"^[A-Za-z0-9][A-Za-z0-9._:/-]*$"
-    )
-    openrouter_api_key: str = ""
-    openrouter_model: str = Field(
-        default="openai/gpt-5.5", pattern=r"^[A-Za-z0-9][A-Za-z0-9._:/-]*$"
-    )
-    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",
-    )
-
-    @model_validator(mode="after")
-    def validate_production_configuration(self) -> "Settings":
-        if self.environment != "production":
-            return self
-        missing = []
-        if not self.auth_required:
-            missing.append("AUTH_REQUIRED=true")
-        if not self.supabase_url:
-            missing.append("SUPABASE_URL")
-        if not self.supabase_service_role_key:
-            missing.append("SUPABASE_SERVICE_ROLE_KEY")
-        provider_keys = {
-            "openai": (self.openai_api_key, "OPENAI_API_KEY"),
-            "gemini": (self.gemini_api_key, "GEMINI_API_KEY"),
-            "openrouter": (self.openrouter_api_key, "OPENROUTER_API_KEY"),
-        }
-        provider_key = provider_keys.get(self.ai_provider)
-        if provider_key and not provider_key[0]:
-            missing.append(provider_key[1])
-        if missing:
-            raise ValueError(
-                "Production configuration is incomplete: " + ", ".join(missing)
-            )
-        return self
-
-    @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/errors.py

@@ -1,3 +0,0 @@
-class ServiceConfigurationError(RuntimeError):
-    """Raised when an operation requires an unconfigured external service."""
-

app/core/rate_limit.py

@@ -1,45 +0,0 @@
-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

@@ -1,65 +0,0 @@
-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

@@ -1,36 +0,0 @@
-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

@@ -1,67 +0,0 @@
-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
-
-
-class DashboardUser(BaseModel):
-    name: str
-    email: str | None = None
-    avatar_url: str | None = None
-
-
-class DashboardStats(BaseModel):
-    workflows: int = 0
-    projects: int = 0
-    active_workflows: int = 0
-    executions: int = 0
-    success_rate: float | None = None
-    ai_generations: int = 0
-
-
-class DashboardWorkflow(BaseModel):
-    id: str
-    name: str
-    project_name: str | None = None
-    updated_at: str
-    node_count: int = 0
-    is_active: bool = False
-    favorite: bool = False
-
-
-class DashboardSummary(BaseModel):
-    workspace_name: str
-    user: DashboardUser
-    stats: DashboardStats
-    workflows: list[DashboardWorkflow] = Field(default_factory=list)
-    templates: list[TemplateSummary] = Field(default_factory=list)

app/models/workflow.py

@@ -1,465 +0,0 @@
-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 CredentialReference(BaseModel):
-    id: str | None = None
-    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, CredentialReference] | 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: Literal["openai", "gemini", "openrouter", "deterministic"] | None = None
-    model: str | None = Field(
-        default=None,
-        min_length=1,
-        max_length=200,
-        pattern=r"^[A-Za-z0-9][A-Za-z0-9._:/-]*$",
-    )
-
-
-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
-    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

@@ -1,178 +0,0 @@
-import json
-from typing import Any, Protocol
-
-from app.models.workflow import (
-    CredentialReference,
-    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: CredentialReference(
-                    id=str(value["id"]) if value.get("id") else None,
-                    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} if value.id else {}),
-                            "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

@@ -1,131 +0,0 @@
-from app.models.catalog import NodeDefinition
-
-
-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-5.5"},
-        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}

app/services/chat.py

@@ -1,84 +0,0 @@
-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

@@ -1,248 +0,0 @@
-import hashlib
-import secrets
-from datetime import UTC, datetime, timedelta
-from uuid import uuid4
-
-import httpx
-
-from app.core.config import Settings
-from app.core.errors import ServiceConfigurationError
-from app.models.workflow import (
-    ShareRequest,
-    ShareResponse,
-    SharedWorkflowResponse,
-    VersionSummary,
-    WorkflowComment,
-    WorkflowDocument,
-)
-
-
-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,
-        }
-
-    def _require_configured(self) -> None:
-        if not self.configured:
-            raise ServiceConfigurationError(
-                "Supabase persistence is not configured on the API server."
-            )
-
-    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:
-        self._require_configured()
-        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())
-        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:
-        self._require_configured()
-        token_hash = hashlib.sha256(token.encode()).hexdigest()
-        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]:
-        self._require_configured()
-        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:
-        self._require_configured()
-        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]:
-        self._require_configured()
-        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:
-        self._require_configured()
-        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

@@ -1,43 +0,0 @@
-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

@@ -1,411 +0,0 @@
-import json
-import re
-from dataclasses import dataclass
-from typing import Protocol
-
-import httpx
-
-from app.core.config import get_settings
-from app.models.workflow import (
-    CredentialReference,
-    GenerateWorkflowResponse,
-    Position,
-    WorkflowDocument,
-    WorkflowEdge,
-    WorkflowMeta,
-    WorkflowNode,
-    WorkflowNodeData,
-)
-
-WORKFLOW_SYSTEM_PROMPT = (
-    "You generate production-ready n8n workflow documents for FlowForge. "
-    "Return only valid JSON with keys workflow, explanation, and warnings. "
-    "The workflow must contain name, active, nodes, edges, settings, and meta. "
-    "Each node must contain id, position {x,y}, and data with label, type, "
-    "typeVersion, category, parameters, and optional credentials. "
-    "Credential objects must include name and type and omit credential IDs. "
-    "Use only these categories: trigger, core, ai, database, communication, cloud, developer. "
-    "Use n8n expressions only when they start with = and have balanced {{ }} braces. "
-    "Never invent secret values, API keys, private URLs, or credential IDs. "
-    "For an unspecified endpoint use an n8n environment expression such as "
-    "={{ $env.SERVICE_BASE_URL }} and mention it in warnings."
-)
-
-
-def _parse_provider_response(content: str, provider: str, model: str) -> GenerateWorkflowResponse:
-    try:
-        result = GenerateWorkflowResponse.model_validate(json.loads(content))
-    except (TypeError, json.JSONDecodeError, ValueError) as exc:
-        raise ValueError(f"{provider} returned an invalid workflow document") from exc
-    result.workflow.meta.generatedBy = f"{provider} {model}"
-    return result
-
-
-class WorkflowProvider(Protocol):
-    async def generate(self, prompt: str, model: str | None = None) -> 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, CredentialReference] | None:
-    if not types:
-        return None
-    return {
-        credential_type: CredentialReference(
-            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, model: str | None = None) -> 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": "={{ $env.API_BASE_URL }}",
-                        "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 deterministic 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=[
-                "Select credentials for each connected service before activation.",
-                "Review generated expressions with representative execution data.",
-            ],
-        )
-
-
-class OpenAIWorkflowProvider:
-    """Generate normalized workflow documents through OpenAI structured JSON output."""
-
-    async def generate(self, prompt: str, model: str | None = None) -> GenerateWorkflowResponse:
-        settings = get_settings()
-        if not settings.openai_api_key:
-            raise RuntimeError("OPENAI_API_KEY is required when AI_PROVIDER=openai.")
-        selected_model = model or settings.openai_model
-        payload = {
-            "model": selected_model,
-            "messages": [
-                {"role": "system", "content": WORKFLOW_SYSTEM_PROMPT},
-                {"role": "user", "content": prompt},
-            ],
-            "response_format": {"type": "json_object"},
-            "max_completion_tokens": 12000,
-        }
-        async with httpx.AsyncClient(timeout=60) as client:
-            response = await client.post(
-                "https://api.openai.com/v1/chat/completions",
-                headers={
-                    "Authorization": f"Bearer {settings.openai_api_key}",
-                    "Content-Type": "application/json",
-                },
-                json=payload,
-            )
-            response.raise_for_status()
-        try:
-            content = response.json()["choices"][0]["message"]["content"]
-        except (KeyError, TypeError, IndexError) as exc:
-            raise ValueError("OpenAI returned an invalid response envelope") from exc
-        return _parse_provider_response(content, "OpenAI", selected_model)
-
-
-class GeminiWorkflowProvider:
-    """Generate workflow documents through the Gemini generateContent API."""
-
-    async def generate(self, prompt: str, model: str | None = None) -> GenerateWorkflowResponse:
-        settings = get_settings()
-        if not settings.gemini_api_key:
-            raise RuntimeError("GEMINI_API_KEY is required when AI_PROVIDER=gemini.")
-        selected_model = model or settings.gemini_model
-        payload = {
-            "systemInstruction": {"parts": [{"text": WORKFLOW_SYSTEM_PROMPT}]},
-            "contents": [{"role": "user", "parts": [{"text": prompt}]}],
-            "generationConfig": {
-                "responseMimeType": "application/json",
-                "temperature": 0.2,
-                "maxOutputTokens": 12000,
-            },
-        }
-        async with httpx.AsyncClient(timeout=60) as client:
-            response = await client.post(
-                f"https://generativelanguage.googleapis.com/v1beta/models/{selected_model}:generateContent",
-                headers={
-                    "x-goog-api-key": settings.gemini_api_key,
-                    "Content-Type": "application/json",
-                },
-                json=payload,
-            )
-            response.raise_for_status()
-        try:
-            parts = response.json()["candidates"][0]["content"]["parts"]
-            content = "".join(part.get("text", "") for part in parts)
-        except (KeyError, TypeError, IndexError) as exc:
-            raise ValueError("Gemini returned an invalid response envelope") from exc
-        return _parse_provider_response(content, "Gemini", selected_model)
-
-
-class OpenRouterWorkflowProvider:
-    """Generate workflow documents through OpenRouter's chat completions API."""
-
-    async def generate(self, prompt: str, model: str | None = None) -> GenerateWorkflowResponse:
-        settings = get_settings()
-        if not settings.openrouter_api_key:
-            raise RuntimeError(
-                "OPENROUTER_API_KEY is required when AI_PROVIDER=openrouter."
-            )
-        selected_model = model or settings.openrouter_model
-        payload = {
-            "model": selected_model,
-            "messages": [
-                {"role": "system", "content": WORKFLOW_SYSTEM_PROMPT},
-                {"role": "user", "content": prompt},
-            ],
-            "response_format": {"type": "json_object"},
-            "max_tokens": 12000,
-        }
-        async with httpx.AsyncClient(timeout=60) as client:
-            response = await client.post(
-                "https://openrouter.ai/api/v1/chat/completions",
-                headers={
-                    "Authorization": f"Bearer {settings.openrouter_api_key}",
-                    "Content-Type": "application/json",
-                    "HTTP-Referer": settings.allowed_origins[0],
-                    "X-Title": settings.app_name,
-                },
-                json=payload,
-            )
-            response.raise_for_status()
-        try:
-            content = response.json()["choices"][0]["message"]["content"]
-        except (KeyError, TypeError, IndexError) as exc:
-            raise ValueError("OpenRouter returned an invalid response envelope") from exc
-        return _parse_provider_response(content, "OpenRouter", selected_model)
-
-
-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:
-        provider = self._providers.get(name)
-        if not provider:
-            raise ValueError(f"Unsupported AI provider: {name}")
-        return provider
-
-
-providers = ProviderRegistry()
-providers.register("openai", OpenAIWorkflowProvider())
-providers.register("gemini", GeminiWorkflowProvider())
-providers.register("openrouter", OpenRouterWorkflowProvider())
-providers.register("deterministic", RuleBasedWorkflowProvider())

app/services/intelligence.py

@@ -1,300 +0,0 @@
-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

@@ -1,230 +0,0 @@
-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

@@ -1,83 +0,0 @@
-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

@@ -1,66 +0,0 @@
-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

@@ -1,411 +0,0 @@
-import re
-from datetime import UTC, datetime
-from uuid import uuid4
-
-import httpx
-
-from app.core.config import Settings
-from app.core.errors import ServiceConfigurationError
-from app.models.catalog import (
-    DashboardStats,
-    DashboardSummary,
-    DashboardUser,
-    DashboardWorkflow,
-    ProjectSummary,
-    TemplateSummary,
-)
-from app.models.workflow import SaveRequest, SaveResponse, WorkflowDocument
-
-
-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",
-        }
-
-    def _require_configured(self) -> None:
-        if not self.configured:
-            raise ServiceConfigurationError(
-                "Supabase persistence is not configured on the API server."
-            )
-
-    @property
-    def base_url(self) -> str:
-        return f"{self.settings.supabase_url.rstrip('/')}/rest/v1"
-
-    async def _workspace_ids(self, client: httpx.AsyncClient, user_id: str) -> list[str]:
-        response = await client.get(
-            f"{self.base_url}/workspace_members?select=workspace_id&user_id=eq.{user_id}",
-            headers=self._headers(),
-        )
-        response.raise_for_status()
-        return [row["workspace_id"] for row in response.json()]
-
-    async def _count(self, client: httpx.AsyncClient, path: str) -> int:
-        response = await client.get(
-            f"{self.base_url}/{path}",
-            headers={**self._headers(), "Prefer": "count=exact"},
-        )
-        response.raise_for_status()
-        content_range = response.headers.get("content-range", "")
-        try:
-            return int(content_range.rsplit("/", 1)[1])
-        except (IndexError, ValueError):
-            return len(response.json())
-
-    @staticmethod
-    def _response_count(response: httpx.Response) -> int:
-        content_range = response.headers.get("content-range", "")
-        try:
-            return int(content_range.rsplit("/", 1)[1])
-        except (IndexError, ValueError):
-            return len(response.json())
-
-    async def save(self, request: SaveRequest, user_id: str) -> SaveResponse:
-        self._require_configured()
-        now = datetime.now(UTC).isoformat()
-        workflow_id = str(uuid4())
-        base_url = self.base_url
-        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]:
-        self._require_configured()
-        base_url = self.base_url
-        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()
-            project_rows = response.json()
-            project_ids = [item["id"] for item in project_rows]
-            workflow_counts: dict[str, int] = {}
-            if project_ids:
-                project_filter = ",".join(project_ids)
-                workflow_response = await client.get(
-                    f"{base_url}/workflows?select=project_id&project_id=in.({project_filter})&is_archived=eq.false",
-                    headers=self._headers(),
-                )
-                workflow_response.raise_for_status()
-                for workflow in workflow_response.json():
-                    project_id = workflow.get("project_id")
-                    if project_id:
-                        workflow_counts[project_id] = workflow_counts.get(project_id, 0) + 1
-        return [
-            ProjectSummary(
-                **item,
-                workflow_count=workflow_counts.get(item["id"], 0),
-            )
-            for item in project_rows
-        ]
-
-    async def workflow(self, workflow_id: str, user_id: str) -> WorkflowDocument:
-        self._require_configured()
-        async with httpx.AsyncClient(timeout=10) as client:
-            workspace_ids = await self._workspace_ids(client, user_id)
-            response = await client.get(
-                f"{self.base_url}/workflows?select=workspace_id,definition&id=eq.{workflow_id}&is_archived=eq.false&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")
-        definition = dict(rows[0]["definition"])
-        definition["id"] = workflow_id
-        return WorkflowDocument.model_validate(definition)
-
-    async def template(self, template_id: str, user_id: str) -> WorkflowDocument:
-        self._require_configured()
-        async with httpx.AsyncClient(timeout=10) as client:
-            response = await client.get(
-                f"{self.base_url}/templates?select=name,definition,is_public,owner_id&id=eq.{template_id}&limit=1",
-                headers=self._headers(),
-            )
-            response.raise_for_status()
-            rows = response.json()
-        if not rows or (not rows[0]["is_public"] and rows[0]["owner_id"] != user_id):
-            raise ValueError("Template is not accessible by this user")
-        definition = dict(rows[0]["definition"])
-        definition.pop("id", None)
-        definition["name"] = rows[0]["name"]
-        return WorkflowDocument.model_validate(definition)
-
-    async def templates(
-        self,
-        user_id: str,
-        query: str = "",
-        category: str | None = None,
-        limit: int = 24,
-        offset: int = 0,
-    ) -> tuple[list[TemplateSummary], int]:
-        self._require_configured()
-        params: dict[str, str | int] = {
-            "select": "id,name,description,category,definition,use_count,tags",
-            "or": f"(is_public.eq.true,owner_id.eq.{user_id})",
-            "order": "use_count.desc",
-            "limit": limit,
-            "offset": offset,
-        }
-        if query:
-            safe_query = re.sub(r"[^\w\s-]", " ", query).strip()
-            if safe_query:
-                params["and"] = (
-                    f"(or(name.ilike.*{safe_query}*,description.ilike.*{safe_query}*))"
-                )
-        if category:
-            safe_category = re.sub(r"[^\w\s-]", "", category).strip()
-            if safe_category:
-                params["category"] = f"eq.{safe_category}"
-        async with httpx.AsyncClient(timeout=10) as client:
-            response = await client.get(
-                f"{self.base_url}/templates",
-                params=params,
-                headers={**self._headers(), "Prefer": "count=exact"},
-            )
-            response.raise_for_status()
-        items = [
-            TemplateSummary(
-                id=row["id"],
-                name=row["name"],
-                description=row.get("description") or "",
-                category=row["category"],
-                node_count=len((row.get("definition") or {}).get("nodes", [])),
-                use_count=row.get("use_count", 0),
-                tags=row.get("tags") or [],
-            )
-            for row in response.json()
-        ]
-        content_range = response.headers.get("content-range", "")
-        try:
-            total = int(content_range.rsplit("/", 1)[1])
-        except (IndexError, ValueError):
-            total = len(items)
-        return items, total
-
-    async def dashboard(self, user_id: str, email: str | None) -> DashboardSummary:
-        self._require_configured()
-        async with httpx.AsyncClient(timeout=15) as client:
-            workspace_ids = await self._workspace_ids(client, user_id)
-            profile_response = await client.get(
-                f"{self.base_url}/users?select=full_name,email,avatar_url&id=eq.{user_id}&limit=1",
-                headers=self._headers(),
-            )
-            profile_response.raise_for_status()
-            profile_rows = profile_response.json()
-            profile = profile_rows[0] if profile_rows else {}
-            display_name = profile.get("full_name") or (email or "Workspace member").split("@", 1)[0]
-
-            if not workspace_ids:
-                return DashboardSummary(
-                    workspace_name="No workspace",
-                    user=DashboardUser(
-                        name=display_name,
-                        email=profile.get("email") or email,
-                        avatar_url=profile.get("avatar_url"),
-                    ),
-                    stats=DashboardStats(),
-                )
-
-            workspace_filter = ",".join(workspace_ids)
-            workspace_response = await client.get(
-                f"{self.base_url}/workspaces?select=id,name&id=in.({workspace_filter})&order=created_at.asc",
-                headers=self._headers(),
-            )
-            project_response = await client.get(
-                f"{self.base_url}/projects?select=id,name&workspace_id=in.({workspace_filter})",
-                headers=self._headers(),
-            )
-            workflow_response = await client.get(
-                f"{self.base_url}/workflows?select=id,name,definition,is_active,updated_at,project_id&workspace_id=in.({workspace_filter})&is_archived=eq.false&order=updated_at.desc&limit=100",
-                headers={**self._headers(), "Prefer": "count=exact"},
-            )
-            favorite_response = await client.get(
-                f"{self.base_url}/favorites?select=workflow_id&user_id=eq.{user_id}&kind=eq.workflow",
-                headers=self._headers(),
-            )
-            for response in (
-                workspace_response,
-                project_response,
-                workflow_response,
-                favorite_response,
-            ):
-                response.raise_for_status()
-
-            projects = {row["id"]: row["name"] for row in project_response.json()}
-            favorites = {row["workflow_id"] for row in favorite_response.json()}
-            workflow_rows = workflow_response.json()
-            workflow_ids = [row["id"] for row in workflow_rows]
-            workflow_total = self._response_count(workflow_response)
-            active_workflows = await self._count(
-                client,
-                f"workflows?select=id&workspace_id=in.({workspace_filter})&is_archived=eq.false&is_active=eq.true&limit=1",
-            )
-            executions = successes = 0
-            if workflow_ids:
-                workflow_filter = ",".join(workflow_ids)
-                executions = await self._count(
-                    client,
-                    f"workflow_runs?select=id&workflow_id=in.({workflow_filter})&limit=1",
-                )
-                successes = await self._count(
-                    client,
-                    f"workflow_runs?select=id&workflow_id=in.({workflow_filter})&status=eq.success&limit=1",
-                )
-            ai_generations = await self._count(
-                client,
-                f"ai_history?select=id&user_id=eq.{user_id}&role=eq.assistant&limit=1",
-            )
-            template_items, _ = await self.templates(user_id, limit=6)
-
-        workspaces = workspace_response.json()
-        workspace_name = workspaces[0]["name"] if len(workspaces) == 1 else "All workspaces"
-        return DashboardSummary(
-            workspace_name=workspace_name,
-            user=DashboardUser(
-                name=display_name,
-                email=profile.get("email") or email,
-                avatar_url=profile.get("avatar_url"),
-            ),
-            stats=DashboardStats(
-                workflows=workflow_total,
-                projects=len(projects),
-                active_workflows=active_workflows,
-                executions=executions,
-                success_rate=round(successes / executions * 100, 1) if executions else None,
-                ai_generations=ai_generations,
-            ),
-            workflows=[
-                DashboardWorkflow(
-                    id=row["id"],
-                    name=row["name"],
-                    project_name=projects.get(row.get("project_id")),
-                    updated_at=row["updated_at"],
-                    node_count=len((row.get("definition") or {}).get("nodes", [])),
-                    is_active=bool(row["is_active"]),
-                    favorite=row["id"] in favorites,
-                )
-                for row in workflow_rows
-            ],
-            templates=template_items,
-        )

app/services/validation.py

@@ -1,215 +0,0 @@
-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 not credential.id:
-                    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)