diff --git a/backend/app/__init__.py b/backend/app/__init__.py index a26b270f..19e297fd 100644 --- a/backend/app/__init__.py +++ b/backend/app/__init__.py @@ -332,9 +332,8 @@ def create_app(config_name=None): from app.api.metrics import metrics_bp app.register_blueprint(metrics_bp, url_prefix='/api/v1/metrics') - # Register blueprints - Workflows - from app.api.workflows import workflows_bp - app.register_blueprint(workflows_bp, url_prefix='/api/v1/workflows') + # The /api/v1/workflows blueprint (React-Flow Workflow Builder) was retired + # in plan 45 Phase 4 -- the Automations extension (tramo) replaces it. # Register blueprints - Servers (Multi-server management) from app.api.servers import servers_bp @@ -620,8 +619,7 @@ def create_app(config_name=None): # scheduled backups). from app.services.deployment_job_service import DeploymentJobService DeploymentJobService.register_jobs() - from app.services.workflow_engine import WorkflowEngine - WorkflowEngine.register_jobs() + # WorkflowEngine.register_jobs() removed in plan 45 Phase 4 (engine retired). from app.services.backup_service import BackupService BackupService.register_jobs() from app.services.backup_policy_service import BackupPolicyService diff --git a/backend/app/api/workflows.py b/backend/app/api/workflows.py deleted file mode 100644 index 442c82fb..00000000 --- a/backend/app/api/workflows.py +++ /dev/null @@ -1,373 +0,0 @@ -import json -import secrets -from datetime import datetime -from flask import Blueprint, request, jsonify -from flask_jwt_extended import jwt_required, get_jwt_identity -from app import db -from app.models import Workflow, User, WorkflowExecution, WorkflowLog - -workflows_bp = Blueprint('workflows', __name__) - - -def _validate_workflow_graph(data): - """Validate workflow graph for cycles if nodes/edges are present.""" - nodes = data.get('nodes', []) - edges = data.get('edges', []) - if nodes and edges: - from app.services.workflow_engine import WorkflowEngine - err = WorkflowEngine.validate_graph(nodes, edges) - if err: - return err - return None - - -@workflows_bp.route('', methods=['GET']) -@jwt_required() -def get_workflows(): - """Get all workflows for the current user.""" - current_user_id = get_jwt_identity() - user = User.query.get(current_user_id) - - if user.role == 'admin': - workflows = Workflow.query.all() - else: - workflows = Workflow.query.filter_by(user_id=current_user_id).all() - - return jsonify({ - 'workflows': [w.to_dict() for w in workflows] - }) - - -@workflows_bp.route('/', methods=['GET']) -@jwt_required() -def get_workflow(workflow_id): - """Get a single workflow by ID.""" - current_user_id = get_jwt_identity() - user = User.query.get(current_user_id) - workflow = Workflow.query.get(workflow_id) - - if not workflow: - return jsonify({'error': 'Workflow not found'}), 404 - - if user.role != 'admin' and workflow.user_id != current_user_id: - return jsonify({'error': 'Access denied'}), 403 - - return jsonify(workflow.to_dict()) - - -@workflows_bp.route('', methods=['POST']) -@jwt_required() -def create_workflow(): - """Create a new workflow.""" - current_user_id = get_jwt_identity() - data = request.get_json() - - if not data: - return jsonify({'error': 'No data provided'}), 400 - - name = data.get('name') - if not name: - return jsonify({'error': 'Name is required'}), 400 - - # Validate graph for cycles - cycle_err = _validate_workflow_graph(data) - if cycle_err: - return jsonify({'error': cycle_err}), 400 - - # Auto-generate webhook_id if trigger is webhook - trigger_config = data.get('trigger_config', {}) - if data.get('trigger_type') == 'webhook' and not trigger_config.get('webhook_id'): - trigger_config['webhook_id'] = secrets.token_urlsafe(24) - - workflow = Workflow( - name=name, - description=data.get('description', ''), - nodes=json.dumps(data.get('nodes', [])), - edges=json.dumps(data.get('edges', [])), - viewport=json.dumps(data.get('viewport')) if data.get('viewport') else None, - is_active=data.get('is_active', False), - trigger_type=data.get('trigger_type', 'manual'), - trigger_config=json.dumps(trigger_config), - user_id=current_user_id - ) - - db.session.add(workflow) - db.session.commit() - - return jsonify({ - 'message': 'Workflow created successfully', - 'workflow': workflow.to_dict() - }), 201 - - -@workflows_bp.route('/', methods=['PUT']) -@jwt_required() -def update_workflow(workflow_id): - """Update an existing workflow.""" - current_user_id = get_jwt_identity() - user = User.query.get(current_user_id) - workflow = Workflow.query.get(workflow_id) - - if not workflow: - return jsonify({'error': 'Workflow not found'}), 404 - - if user.role != 'admin' and workflow.user_id != current_user_id: - return jsonify({'error': 'Access denied'}), 403 - - data = request.get_json() - if not data: - return jsonify({'error': 'No data provided'}), 400 - - # Validate graph for cycles - cycle_err = _validate_workflow_graph(data) - if cycle_err: - return jsonify({'error': cycle_err}), 400 - - if 'name' in data: - workflow.name = data['name'] - if 'description' in data: - workflow.description = data['description'] - if 'nodes' in data: - workflow.nodes = json.dumps(data['nodes']) - if 'edges' in data: - workflow.edges = json.dumps(data['edges']) - if 'viewport' in data: - workflow.viewport = json.dumps(data['viewport']) if data['viewport'] else None - - # Automation fields - if 'is_active' in data: - workflow.is_active = data['is_active'] - if 'trigger_type' in data: - workflow.trigger_type = data['trigger_type'] - if 'trigger_config' in data: - trigger_config = data['trigger_config'] - # Auto-generate webhook_id if switching to webhook trigger - if data.get('trigger_type') == 'webhook' and not trigger_config.get('webhook_id'): - existing = json.loads(workflow.trigger_config) if workflow.trigger_config else {} - trigger_config['webhook_id'] = existing.get('webhook_id') or secrets.token_urlsafe(24) - workflow.trigger_config = json.dumps(trigger_config) - - db.session.commit() - - return jsonify({ - 'message': 'Workflow updated successfully', - 'workflow': workflow.to_dict() - }) - - -@workflows_bp.route('/', methods=['DELETE']) -@jwt_required() -def delete_workflow(workflow_id): - """Delete a workflow.""" - current_user_id = get_jwt_identity() - user = User.query.get(current_user_id) - workflow = Workflow.query.get(workflow_id) - - if not workflow: - return jsonify({'error': 'Workflow not found'}), 404 - - if user.role != 'admin' and workflow.user_id != current_user_id: - return jsonify({'error': 'Access denied'}), 403 - - db.session.delete(workflow) - db.session.commit() - - return jsonify({'message': 'Workflow deleted successfully'}) - - -@workflows_bp.route('//execute', methods=['POST']) -@jwt_required() -def execute_workflow(workflow_id): - """Trigger a workflow execution manually.""" - from app.services.workflow_engine import WorkflowEngine - - current_user_id = get_jwt_identity() - user = User.query.get(current_user_id) - workflow = Workflow.query.get(workflow_id) - - if not workflow: - return jsonify({'error': 'Workflow not found'}), 404 - - if user.role != 'admin' and workflow.user_id != current_user_id: - return jsonify({'error': 'Access denied'}), 403 - - data = request.get_json() or {} - context = data.get('context', {}) - - execution_id = WorkflowEngine.enqueue_execution( - workflow_id=workflow_id, - trigger_type='manual', - context=context - ) - - return jsonify({ - 'message': 'Workflow execution started', - 'execution_id': execution_id - }) - - -@workflows_bp.route('//executions', methods=['GET']) -@jwt_required() -def get_workflow_executions(workflow_id): - """Get execution history for a workflow.""" - current_user_id = get_jwt_identity() - user = User.query.get(current_user_id) - workflow = Workflow.query.get(workflow_id) - - if not workflow: - return jsonify({'error': 'Workflow not found'}), 404 - - if user.role != 'admin' and workflow.user_id != current_user_id: - return jsonify({'error': 'Access denied'}), 403 - - executions = WorkflowExecution.query.filter_by(workflow_id=workflow_id).order_by(WorkflowExecution.started_at.desc()).all() - - return jsonify({ - 'executions': [e.to_dict() for e in executions] - }) - - -@workflows_bp.route('/executions/', methods=['GET']) -@jwt_required() -def get_execution_details(execution_id): - """Get details for a specific execution.""" - current_user_id = get_jwt_identity() - user = User.query.get(current_user_id) - execution = WorkflowExecution.query.get(execution_id) - - if not execution: - return jsonify({'error': 'Execution not found'}), 404 - - workflow = execution.workflow - if user.role != 'admin' and workflow.user_id != current_user_id: - return jsonify({'error': 'Access denied'}), 403 - - return jsonify(execution.to_dict()) - - -@workflows_bp.route('/executions//logs', methods=['GET']) -@jwt_required() -def get_execution_logs(execution_id): - """Get logs for a specific execution.""" - current_user_id = get_jwt_identity() - user = User.query.get(current_user_id) - execution = WorkflowExecution.query.get(execution_id) - - if not execution: - return jsonify({'error': 'Execution not found'}), 404 - - workflow = execution.workflow - if user.role != 'admin' and workflow.user_id != current_user_id: - return jsonify({'error': 'Access denied'}), 403 - - logs = WorkflowLog.query.filter_by(execution_id=execution_id).order_by(WorkflowLog.timestamp.asc()).all() - - return jsonify({ - 'logs': [l.to_dict() for l in logs] - }) - - -@workflows_bp.route('//deploy', methods=['POST']) -@jwt_required() -def deploy_workflow(workflow_id): - """ - Deploy all resources from a workflow. - """ - from app.services.workflow_service import WorkflowService - - current_user_id = get_jwt_identity() - - result = WorkflowService.deploy_workflow(workflow_id, current_user_id) - - if result.get('success'): - return jsonify(result), 200 - elif result.get('error') == 'Workflow not found': - return jsonify(result), 404 - elif result.get('error') == 'Access denied': - return jsonify(result), 403 - else: - # Partial success or errors - return 200 with error details - return jsonify(result), 200 - - -@workflows_bp.route('/hooks/', methods=['POST']) -def webhook_trigger(webhook_id): - """ - Public webhook endpoint to trigger a workflow. - No JWT required — the webhook_id acts as the authentication token. - """ - from app.services.workflow_engine import WorkflowEngine, CycleDetectedError - - # Find the workflow with this webhook_id - workflows = Workflow.query.filter_by( - is_active=True, - trigger_type='webhook' - ).all() - - target_workflow = None - for wf in workflows: - try: - config = json.loads(wf.trigger_config) if wf.trigger_config else {} - if config.get('webhook_id') == webhook_id: - target_workflow = wf - break - except (json.JSONDecodeError, TypeError): - continue - - if not target_workflow: - return jsonify({'error': 'Webhook not found'}), 404 - - # Build context from the incoming request - context = { - 'webhook_id': webhook_id, - 'method': request.method, - 'headers': dict(request.headers), - 'triggered_at': datetime.utcnow().isoformat() - } - - # Include request body - if request.is_json: - context['body'] = request.get_json(silent=True) or {} - else: - body = request.get_data(as_text=True) - context['body'] = body[:8192] if body else '' - - # Include query parameters - if request.args: - context['query'] = dict(request.args) - - try: - execution_id = WorkflowEngine.enqueue_execution( - workflow_id=target_workflow.id, - trigger_type='webhook', - context=context - ) - return jsonify({ - 'message': 'Workflow triggered', - 'execution_id': execution_id, - 'workflow': target_workflow.name - }), 200 - except CycleDetectedError as e: - return jsonify({'error': str(e)}), 400 - except Exception as e: - return jsonify({'error': f'Execution failed: {str(e)}'}), 500 - - -@workflows_bp.route('/validate', methods=['POST']) -@jwt_required() -def validate_workflow(): - """Validate a workflow graph for cycles and other issues.""" - from app.services.workflow_engine import WorkflowEngine - - data = request.get_json() - if not data: - return jsonify({'error': 'No data provided'}), 400 - - nodes = data.get('nodes', []) - edges = data.get('edges', []) - - err = WorkflowEngine.validate_graph(nodes, edges) - if err: - return jsonify({'valid': False, 'error': err}), 200 - - return jsonify({'valid': True}), 200 diff --git a/backend/app/jobs/builtin_handlers.py b/backend/app/jobs/builtin_handlers.py index e77e3afb..388e61f1 100644 --- a/backend/app/jobs/builtin_handlers.py +++ b/backend/app/jobs/builtin_handlers.py @@ -61,47 +61,16 @@ def check_auto_sync_schedules(): def check_workflow_schedules(): - """Check all active workflows with cron triggers and run those that are due.""" - from app.models.workflow import Workflow - from app.services.workflow_engine import WorkflowEngine - from datetime import datetime - import json + """Retired no-op (plan 45 Phase 4). - try: - from croniter import croniter - except ImportError: - logger.debug('croniter not installed, skipping workflow schedule check') - return - - workflows = Workflow.query.filter_by(is_active=True, trigger_type='cron').all() - if not workflows: - return - - now = datetime.utcnow() - - for workflow in workflows: - try: - config = json.loads(workflow.trigger_config) if workflow.trigger_config else {} - cron_expr = config.get('cron') - if not cron_expr or not croniter.is_valid(cron_expr): - continue - cron = croniter(cron_expr, now) - prev_run = cron.get_prev(datetime) - seconds_since_due = (now - prev_run).total_seconds() - # Don't run multiple times for the same slot. - if 0 < seconds_since_due <= 90: - if workflow.last_run_at: - seconds_since_last_run = (now - workflow.last_run_at).total_seconds() - if seconds_since_last_run < 110: - continue - logger.info(f'Scheduled workflow triggered: {workflow.name} (ID: {workflow.id})') - WorkflowEngine.enqueue_execution( - workflow_id=workflow.id, - trigger_type='cron', - context={'scheduled_at': prev_run.isoformat()}, - ) - except Exception as e: - logger.error(f'Workflow schedule check failed for workflow {workflow.id}: {e}') + The React-Flow Workflow Builder and its execution engine were removed in + favour of the Automations extension (tramo), whose own in-process scheduler + drives ``cron-trigger`` nodes inside the managed engine. The legacy + ``workflows`` tables are kept for read-only export, so this scheduled handler + is retained as a no-op (avoids an orphaned-handler error for any persisted + ``workflow-schedules`` schedule row) but no longer executes anything. + """ + return # How often the per-site WordPress health poller runs (seconds). diff --git a/backend/app/services/docker_service.py b/backend/app/services/docker_service.py index dc31de51..ce8f87b3 100644 --- a/backend/app/services/docker_service.py +++ b/backend/app/services/docker_service.py @@ -263,13 +263,10 @@ class DockerService: capture_output=True, text=True ) if result.returncode == 0: - try: - from app.services.workflow_engine import WorkflowEventBus - WorkflowEventBus.emit('app_stopped', { - 'container_id': container_id - }) - except Exception as e: - logger.error(f"Failed to emit app_stopped event: {e}") + # The panel emits the `app.stopped` event via the audit trail + # (AuditService.log('app.stop') -> EventService.emit_for_audit), + # so this low-level helper no longer emits its own event -- doing + # so would double-fire it for Automations triggers (plan 45 Ph4). return {'success': True} return {'success': False, 'error': result.stderr} except Exception as e: diff --git a/backend/app/services/environment_health_service.py b/backend/app/services/environment_health_service.py index 93a0f6fb..85c9d85a 100644 --- a/backend/app/services/environment_health_service.py +++ b/backend/app/services/environment_health_service.py @@ -100,15 +100,17 @@ class EnvironmentHealthService: site.last_health_check = datetime.utcnow() db.session.commit() - # Emit event for workflow triggers on every failing poll (back-compat for - # existing event-trigger workflows keyed on 'health_check_failed'). + # Emit a panel event on every failing poll so Automations (tramo) + # workflows can trigger on it through the events bridge (plan 45 Ph4, + # ported from the retired Workflow Builder's 'health_check_failed'). if overall in ('unhealthy', 'degraded'): try: - from app.services.workflow_engine import WorkflowEventBus - WorkflowEventBus.emit('health_check_failed', { + from app.services.event_service import EventService + EventService.emit('health.check_failed', { + 'event': 'health.check_failed', 'site_id': site_id, 'overall_status': overall, - 'checks': checks + 'checks': checks, }) except Exception: pass diff --git a/backend/app/services/event_service.py b/backend/app/services/event_service.py index 5322e527..368ee770 100644 --- a/backend/app/services/event_service.py +++ b/backend/app/services/event_service.py @@ -57,6 +57,12 @@ EVENT_CATALOG = [ {'type': 'wordpress.update_rolled_back', 'category': 'WordPress', 'description': 'A WordPress update was auto-rolled-back'}, {'type': 'wordpress.deployed', 'category': 'WordPress', 'description': 'A WordPress git deploy completed'}, {'type': 'wordpress.deploy_failed', 'category': 'WordPress', 'description': 'A WordPress git deploy failed'}, + # Ported from the retired Workflow Builder event bus (plan 45 Phase 4) so + # Automations (tramo) workflows can trigger on them via the events bridge. + {'type': 'health.check_failed', 'category': 'Monitoring', 'description': 'An environment health check failed'}, + {'type': 'git.push', 'category': 'Deployments', 'description': 'A git push was received for an app'}, + {'type': 'monitor.high_cpu', 'category': 'Monitoring', 'description': 'A server crossed its high-CPU alert threshold'}, + {'type': 'monitor.high_memory', 'category': 'Monitoring', 'description': 'A server crossed its high-memory alert threshold'}, ] diff --git a/backend/app/services/git_service.py b/backend/app/services/git_service.py index 7d2138aa..f788d277 100644 --- a/backend/app/services/git_service.py +++ b/backend/app/services/git_service.py @@ -436,13 +436,16 @@ class GitService: 'message': f'Ignoring push to {ref}, configured branch is {branch}' } - # Emit event for workflow triggers + # Emit a panel event so Automations (tramo) workflows can trigger on a + # git push through the events bridge (plan 45 Ph4, ported from the + # retired Workflow Builder's 'git_push'). try: - from app.services.workflow_engine import WorkflowEventBus - WorkflowEventBus.emit('git_push', { + from app.services.event_service import EventService + EventService.emit('git.push', { + 'event': 'git.push', 'app_id': app_id, 'branch': branch, - 'ref': ref + 'ref': ref, }) except Exception: pass diff --git a/backend/app/services/monitoring_service.py b/backend/app/services/monitoring_service.py index 413b60f1..0b4ed4d2 100644 --- a/backend/app/services/monitoring_service.py +++ b/backend/app/services/monitoring_service.py @@ -303,24 +303,28 @@ class MonitoringService: commit=True, ) - # Emit events for workflow triggers + # Emit panel events so Automations (tramo) workflows can trigger on a + # high-CPU/memory alert through the events bridge (plan 45 Ph4, ported + # from the retired Workflow Builder's 'high_cpu'/'high_memory'). try: - from app.services.workflow_engine import WorkflowEventBus + from app.services.event_service import EventService for alert in alerts_to_send: if alert['type'] == 'cpu': - WorkflowEventBus.emit('high_cpu', { + EventService.emit('monitor.high_cpu', { + 'event': 'monitor.high_cpu', 'percent': alert.get('value'), 'threshold': alert.get('threshold'), - 'severity': alert.get('severity') + 'severity': alert.get('severity'), }) elif alert['type'] == 'memory': - WorkflowEventBus.emit('high_memory', { + EventService.emit('monitor.high_memory', { + 'event': 'monitor.high_memory', 'percent': alert.get('value'), 'threshold': alert.get('threshold'), - 'severity': alert.get('severity') + 'severity': alert.get('severity'), }) except Exception: - logger.exception("Error emitting workflow events for alerts") + logger.exception("Error emitting monitor events for alerts") @classmethod def log_alert(cls, alerts: List[Dict]) -> None: diff --git a/backend/app/services/workflow_engine.py b/backend/app/services/workflow_engine.py deleted file mode 100644 index bfd071e1..00000000 --- a/backend/app/services/workflow_engine.py +++ /dev/null @@ -1,1001 +0,0 @@ -""" -Advanced Workflow & Automation Engine - -Executes event-driven workflows with DAG-based execution, logic branching, -variable interpolation, timeouts, retries, and script sandboxing. -""" - -import json -import logging -import os -import re -import signal -import subprocess -import threading -import traceback -from collections import deque -from datetime import datetime -from typing import Dict, List, Any, Optional, Set, Tuple - -from app import db -from app.models import Workflow, WorkflowExecution, WorkflowLog, User -from app.services.workflow_service import WorkflowService -from app.services.docker_service import DockerService -from app.services.database_service import DatabaseService -from app.services.notification_service import NotificationService - -logger = logging.getLogger(__name__) - -# Defaults for node execution -DEFAULT_TIMEOUT = 300 # 5 minutes -MAX_TIMEOUT = 3600 # 1 hour -DEFAULT_RETRY_COUNT = 0 -MAX_RETRY_COUNT = 5 -DEFAULT_RETRY_DELAY = 5 # seconds -MAX_OUTPUT_SIZE = 1024 * 512 # 512 KB - -# Unified job kinds for asynchronous workflow execution + event dispatch -# (see WorkflowEngine.enqueue_execution / register_jobs). -WORKFLOW_JOB_KIND = 'workflow.execute' -WORKFLOW_DISPATCH_JOB_KIND = 'workflow.dispatch' - - -def _parse_version(s: str) -> tuple: - """Turn '3.11.4' / 'v20.10.0' / '8.2' into a comparable tuple of ints. - Trailing non-numeric parts are dropped silently.""" - if not s: - return () - s = s.lstrip('vV').strip() - parts = [] - for part in s.split('.'): - digits = ''.join(c for c in part if c.isdigit()) - if not digits: - break - parts.append(int(digits)) - return tuple(parts) - - -def _runtime_satisfies(actual: str, spec: str) -> bool: - """Compare actual version against a spec like '>=3.11', '>=18', - '==8.2', or just '3.11' (treated as >=). Naive but covers the - realistic gate cases without an extra dep.""" - spec = spec.strip() - op = '>=' - for prefix in ('>=', '<=', '==', '>', '<'): - if spec.startswith(prefix): - op = prefix - spec = spec[len(prefix):].strip() - break - a = _parse_version(actual) - b = _parse_version(spec) - if not a or not b: - return False - if op == '>=': - return a >= b - if op == '<=': - return a <= b - if op == '==': - return a == b - if op == '>': - return a > b - if op == '<': - return a < b - return False - - -class CycleDetectedError(Exception): - """Raised when a cycle is detected in the workflow graph.""" - pass - - -class NodeTimeoutError(Exception): - """Raised when a node exceeds its timeout.""" - pass - - -class WorkflowEngine: - """Engine for executing advanced workflows with DAG support.""" - - # ------------------------------------------------------------------ - # Public API - # ------------------------------------------------------------------ - - @staticmethod - def validate_graph(nodes: List[Dict], edges: List[Dict]) -> Optional[str]: - """ - Validate a workflow graph for cycles. - - Returns None if valid, or an error message string if a cycle is found. - """ - adj: Dict[str, List[str]] = {} - node_ids = {n['id'] for n in nodes} - - for edge in edges: - src = edge['source'] - if src not in adj: - adj[src] = [] - adj[src].append(edge['target']) - - # Kahn's algorithm for cycle detection - in_degree = {nid: 0 for nid in node_ids} - for src, targets in adj.items(): - for t in targets: - if t in in_degree: - in_degree[t] += 1 - - queue = deque(nid for nid, deg in in_degree.items() if deg == 0) - visited_count = 0 - - while queue: - nid = queue.popleft() - visited_count += 1 - for neighbor in adj.get(nid, []): - if neighbor in in_degree: - in_degree[neighbor] -= 1 - if in_degree[neighbor] == 0: - queue.append(neighbor) - - if visited_count < len(node_ids): - # Find nodes involved in the cycle for a helpful message - cycle_nodes = [nid for nid, deg in in_degree.items() if deg > 0] - node_labels = {} - for n in nodes: - node_labels[n['id']] = n.get('data', {}).get('label', n['id']) - cycle_labels = [node_labels.get(nid, nid) for nid in cycle_nodes[:5]] - return f"Cycle detected involving: {', '.join(cycle_labels)}" - - return None - - @staticmethod - def execute_workflow(workflow_id: int, trigger_type: str = 'manual', - context: Dict[str, Any] = None) -> int: - """Create and SYNCHRONOUSLY run a workflow. Returns the execution id. - - Kept for callers that want to block on the result (and tests). Async - callers (API/webhook/cron/event) use enqueue_execution instead. - """ - execution = WorkflowEngine._create_execution(workflow_id, trigger_type, context) - WorkflowEngine.run_execution(execution.id) - return execution.id - - @staticmethod - def enqueue_execution(workflow_id: int, trigger_type: str = 'manual', - context: Dict[str, Any] = None) -> int: - """Create the execution row and run it asynchronously via the unified job - system (kind ``workflow.execute``). Returns the execution id immediately, - so triggers don't block on the DAG. Graph validation still happens here, - so an invalid workflow raises synchronously (callers can 400).""" - execution = WorkflowEngine._create_execution(workflow_id, trigger_type, context) - from app.jobs.service import JobService - JobService.enqueue( - WORKFLOW_JOB_KIND, - payload={'execution_id': execution.id}, - max_attempts=1, # workflows aren't idempotent (deploy/script/notify) - owner_type='workflow', - owner_id=workflow_id, - ) - return execution.id - - @staticmethod - def _create_execution(workflow_id: int, trigger_type: str = 'manual', - context: Dict[str, Any] = None) -> WorkflowExecution: - """Validate the workflow graph and persist a new (running) execution row.""" - workflow = Workflow.query.get(workflow_id) - if not workflow: - raise ValueError(f"Workflow {workflow_id} not found") - - nodes = json.loads(workflow.nodes) if workflow.nodes else [] - edges = json.loads(workflow.edges) if workflow.edges else [] - - # Validate graph before creating the execution - cycle_err = WorkflowEngine.validate_graph(nodes, edges) - if cycle_err: - raise CycleDetectedError(cycle_err) - - execution = WorkflowExecution( - workflow_id=workflow_id, - trigger_type=trigger_type, - status='running', - context=json.dumps(context or {}), - started_at=datetime.utcnow() - ) - db.session.add(execution) - db.session.commit() - - workflow.last_run_at = execution.started_at - db.session.commit() - return execution - - @staticmethod - def run_execution(execution_id: int) -> int: - """Run an already-created execution to completion. The synchronous core - shared by execute_workflow and the workflow.execute job handler.""" - execution = WorkflowExecution.query.get(execution_id) - if not execution: - raise ValueError(f"WorkflowExecution {execution_id} not found") - workflow = execution.workflow - nodes = json.loads(workflow.nodes) if workflow.nodes else [] - edges = json.loads(workflow.edges) if workflow.edges else [] - - try: - WorkflowEngine._run_execution(execution.id, nodes, edges) - except Exception as e: - WorkflowEngine._log(execution.id, f"Engine Error: {str(e)}", level='ERROR') - execution.status = 'failed' - execution.completed_at = datetime.utcnow() - workflow.last_status = 'failed' - db.session.commit() - - return execution.id - - @staticmethod - def _run_workflow_job(job): - """Unified-job handler for ``workflow.execute``. Runs the queued execution; - a failed run is raised so the unified job is marked failed too (the - WorkflowExecution row carries the detailed per-node status/logs).""" - execution_id = (job.get_payload() or {}).get('execution_id') - if not execution_id: - raise ValueError('workflow.execute job missing execution_id') - WorkflowEngine.run_execution(execution_id) - execution = WorkflowExecution.query.get(execution_id) - if execution and execution.status == 'failed': - raise RuntimeError(f'Workflow execution {execution_id} failed') - return {'execution_id': execution_id, - 'status': execution.status if execution else None} - - @staticmethod - def register_jobs(): - """Register the workflow job handlers with the unified job registry. - Called once at app startup (see app/__init__.py).""" - from app.jobs import registry - registry.register(WORKFLOW_JOB_KIND, WorkflowEngine._run_workflow_job, replace=True) - registry.register(WORKFLOW_DISPATCH_JOB_KIND, WorkflowEventBus.dispatch_event, replace=True) - - # ------------------------------------------------------------------ - # DAG Execution - # ------------------------------------------------------------------ - - @staticmethod - def _run_execution(execution_id: int, nodes: List[Dict], edges: List[Dict]): - """Run the workflow using topological DAG execution with branch support.""" - execution = WorkflowExecution.query.get(execution_id) - workflow = execution.workflow - - WorkflowEngine._log(execution_id, f"Starting workflow: {workflow.name}") - - if not nodes: - WorkflowEngine._log(execution_id, "Workflow has no nodes", level='WARNING') - execution.status = 'success' - execution.completed_at = datetime.utcnow() - db.session.commit() - return - - # Build graph structures - node_map = {n['id']: n for n in nodes} - adj: Dict[str, List[Tuple[str, str]]] = {} # source -> [(target, sourceHandle)] - in_degree: Dict[str, int] = {n['id']: 0 for n in nodes} - - for edge in edges: - src = edge['source'] - tgt = edge['target'] - src_handle = edge.get('sourceHandle', 'output') - if src not in adj: - adj[src] = [] - adj[src].append((tgt, src_handle)) - if tgt in in_degree: - in_degree[tgt] += 1 - - # Start with root nodes (no incoming edges) - ready = deque(nid for nid, deg in in_degree.items() if deg == 0) - if not ready: - ready.append(nodes[0]['id']) - - context = json.loads(execution.context) if execution.context else {} - results: Dict[str, Dict] = {} - processed: Set[str] = set() - # Track which branches are active (for logic_if gating) - # Maps node_id -> set of sourceHandles that were activated - active_branches: Dict[str, Set[str]] = {} - failed = False - - while ready and not failed: - node_id = ready.popleft() - - if node_id in processed: - continue - - node = node_map.get(node_id) - if not node: - continue - - # Check if this node is gated by a logic_if branch - if not WorkflowEngine._is_node_reachable(node_id, edges, active_branches, processed): - processed.add(node_id) - # Still decrement successors so they can become ready - for tgt, _ in adj.get(node_id, []): - if tgt in in_degree: - in_degree[tgt] -= 1 - if in_degree[tgt] == 0: - ready.append(tgt) - continue - - node_label = node.get('data', {}).get('label', node_id) - WorkflowEngine._log(execution_id, f"Executing node: {node_label} ({node['type']})", node_id=node_id) - - try: - node_result = WorkflowEngine._execute_node_with_retry( - node, edges, execution, context, results - ) - results[node_id] = node_result - - if not node_result.get('success', True): - WorkflowEngine._log( - execution_id, - f"Node failed: {node_result.get('error', 'unknown')}", - level='ERROR', node_id=node_id - ) - if node_result.get('critical', True): - failed = True - break - - # For logic_if nodes, record which branch was taken - if node['type'] == 'logic_if': - branch = node_result.get('branch', 'true') - if node_id not in active_branches: - active_branches[node_id] = set() - active_branches[node_id].add(branch) - - # Enqueue successor nodes whose in-degree reaches 0 - for tgt, src_handle in adj.get(node_id, []): - if tgt in in_degree: - in_degree[tgt] -= 1 - if in_degree[tgt] == 0: - ready.append(tgt) - - except Exception as e: - WorkflowEngine._log(execution_id, f"Node Execution Error: {str(e)}", level='ERROR', node_id=node_id) - WorkflowEngine._log(execution_id, traceback.format_exc(), level='DEBUG', node_id=node_id) - failed = True - break - - processed.add(node_id) - - execution.status = 'failed' if failed else 'success' - execution.results = json.dumps(results) - execution.completed_at = datetime.utcnow() - workflow.last_status = execution.status - db.session.commit() - - WorkflowEngine._log(execution_id, f"Workflow finished with status: {execution.status}") - - @staticmethod - def _is_node_reachable(node_id: str, edges: List[Dict], - active_branches: Dict[str, Set[str]], - processed: Set[str]) -> bool: - """ - Check if a node should execute based on logic_if branching. - - A node is unreachable if ALL its incoming edges from logic_if nodes - come through inactive branches. - """ - incoming_from_logic = [] - - for edge in edges: - if edge['target'] != node_id: - continue - src = edge['source'] - src_handle = edge.get('sourceHandle', 'output') - - # Only gate on logic_if nodes that have already been processed - if src in active_branches: - incoming_from_logic.append((src, src_handle)) - - if not incoming_from_logic: - return True # No logic_if gating, always reachable - - # Reachable if at least one logic_if branch leading here is active - for src, src_handle in incoming_from_logic: - if src_handle in active_branches[src]: - return True - - return False - - # ------------------------------------------------------------------ - # Node Execution with Retry - # ------------------------------------------------------------------ - - @staticmethod - def _execute_node_with_retry(node: Dict, edges: List, execution: WorkflowExecution, - context: Dict, results: Dict) -> Dict: - """Execute a node with retry support.""" - node_data = node.get('data', {}) - retry_count = min(int(node_data.get('retryCount', DEFAULT_RETRY_COUNT)), MAX_RETRY_COUNT) - retry_delay = max(1, int(node_data.get('retryDelay', DEFAULT_RETRY_DELAY))) - - last_result = None - for attempt in range(retry_count + 1): - if attempt > 0: - WorkflowEngine._log( - execution.id, - f"Retry {attempt}/{retry_count} after {retry_delay}s", - node_id=node['id'] - ) - import time - time.sleep(retry_delay) - - last_result = WorkflowEngine._execute_node(node, edges, execution, context, results) - - if last_result.get('success', True): - return last_result - - return last_result - - # ------------------------------------------------------------------ - # Node Execution - # ------------------------------------------------------------------ - - @staticmethod - def _execute_node(node: Dict, edges: List, execution: WorkflowExecution, - context: Dict, results: Dict) -> Dict: - """Execute a single node and return its results.""" - node_type = node.get('type') - node_data = node.get('data', {}) - - if node_type == 'trigger': - return {'success': True, 'output': context} - - elif node_type in ('database', 'dockerApp', 'service', 'domain'): - res = WorkflowService.deploy_node(node, edges, execution.workflow.user_id, results) - return res - - elif node_type == 'notification': - return WorkflowEngine._execute_notification(node, execution, context, results) - - elif node_type == 'script': - return WorkflowEngine._execute_script(node, execution, context, results) - - elif node_type == 'logic_if': - return WorkflowEngine._execute_logic_if(node, context, results) - - # Phase 4 — fleet workflow primitives. Each routes to an agent - # via the registry; on failure they short-circuit the workflow - # so a missing capability or wrong runtime doesn't silently - # mutate the target host. - elif node_type == 'agent_command': - return WorkflowEngine._execute_agent_command(node, execution, context, results) - elif node_type == 'capability_gate': - return WorkflowEngine._execute_capability_gate(node) - elif node_type == 'runtime_gate': - return WorkflowEngine._execute_runtime_gate(node) - - return {'success': True, 'message': f"Node type {node_type} passed through"} - - # ------------------------------------------------------------------ - # Phase 4: Agent Command + Gate Execution - # ------------------------------------------------------------------ - - @staticmethod - def _resolve_target(node_data: Dict) -> Optional[str]: - """Pull a `{kind, server_id}` target from node_data and return - the server_id for agent dispatch, or None for the local panel - host. Raises if the target dict is malformed.""" - target = node_data.get('target') or {'kind': 'local'} - kind = target.get('kind') - if kind == 'local' or kind is None: - return None - if kind == 'agent': - sid = target.get('server_id') - if not sid: - raise ValueError("agent target missing server_id") - return sid - # 'agent_group' lands when multi-target execution does — for - # now reject so a user authoring a template against a - # not-yet-implemented mode gets a clear error instead of - # silent passthrough. - raise ValueError(f"unsupported target kind: {kind}") - - @staticmethod - def _execute_agent_command(node: Dict, execution: WorkflowExecution, - context: Dict, results: Dict) -> Dict: - """Dispatch an action to an agent and return the result. The - node's data carries the action verb, params, and target — see - plan §Phase 4 for the schema.""" - from app.services.agent_registry import agent_registry - - node_data = node.get('data', {}) - action = (node_data.get('action') or '').strip() - if not action: - return {'success': False, 'error': 'agent_command node missing action'} - - try: - server_id = WorkflowEngine._resolve_target(node_data) - except ValueError as e: - return {'success': False, 'error': str(e)} - - if server_id is None: - # Local target on an agent_command node is unusual — the - # equivalent panel-local action belongs in the original - # node types. Treat as a config error rather than guessing. - return { - 'success': False, - 'error': "agent_command requires an agent target; use the matching local node type instead", - } - - # Interpolate params so templates can reference earlier results. - raw_params = node_data.get('params') or {} - params: Dict[str, Any] = {} - for k, v in raw_params.items(): - if isinstance(v, str): - params[k] = WorkflowEngine._interpolate(v, context, results, execution) - else: - params[k] = v - - timeout = float(node_data.get('timeout_s') or DEFAULT_TIMEOUT) - timeout = min(timeout, MAX_TIMEOUT) - - WorkflowEngine._log( - execution.id, - f"agent {server_id}: {action} {params}", - node_id=node.get('id'), - ) - - result = agent_registry.send_command( - server_id=server_id, - action=action, - params=params, - user_id=execution.workflow.user_id if execution.workflow else None, - timeout=timeout, - ) - - # Translate agent_registry's envelope into the engine's - # standard {success, output|error} shape so downstream nodes - # can branch on it uniformly. - if not result.get('success'): - on_failure = (node_data.get('on_failure') or 'abort').lower() - return { - 'success': False, - 'error': result.get('error') or 'agent command failed', - 'code': result.get('code'), - 'critical': on_failure == 'abort', - } - return { - 'success': True, - 'output': result.get('data'), - 'action': action, - 'server_id': server_id, - } - - @staticmethod - def _execute_capability_gate(node: Dict) -> Dict: - """Pass the workflow only if the target server reports every - listed capability. A gate failure aborts by default — gates - exist precisely to prevent the next step from running on an - unsupported host.""" - from app.services.agent_registry import agent_registry - - node_data = node.get('data', {}) - try: - server_id = WorkflowEngine._resolve_target(node_data) - except ValueError as e: - return {'success': False, 'error': str(e), 'critical': True} - - require = node_data.get('require') or [] - if isinstance(require, str): - require = [require] - require = [str(r).strip() for r in require if str(r).strip()] - if not require: - return {'success': True, 'message': 'no capabilities required'} - - if server_id is None: - # Panel host — every "capability" we ask about is implicitly - # available (it's the panel itself). Tighter modelling can - # come later; refusing here would break local-only workflows - # that want to use the same gate node for clarity. - return {'success': True, 'satisfied': require, 'target': 'local'} - - caps = agent_registry.get_capabilities(server_id) or {} - missing = [r for r in require if not caps.get(r)] - if missing: - on_missing = (node_data.get('on_missing') or 'abort').lower() - return { - 'success': False, - 'error': f"target missing capabilities: {', '.join(missing)}", - 'missing': missing, - 'satisfied': [r for r in require if caps.get(r)], - 'critical': on_missing == 'abort', - } - return {'success': True, 'satisfied': require, 'server_id': server_id} - - @staticmethod - def _execute_runtime_gate(node: Dict) -> Dict: - """Pass the workflow only if every required runtime is present - at >= the requested version. Comparison is naive lexicographic - on tuples-of-ints — good enough for "python >= 3.11" style - gates without dragging in a semver dependency.""" - from app.services.agent_registry import agent_registry - - node_data = node.get('data', {}) - try: - server_id = WorkflowEngine._resolve_target(node_data) - except ValueError as e: - return {'success': False, 'error': str(e), 'critical': True} - - require = node_data.get('require') or {} - if not isinstance(require, dict) or not require: - return {'success': True, 'message': 'no runtimes required'} - - if server_id is None: - # Local panel — we don't track runtime versions for the - # panel itself; accept and move on. - return {'success': True, 'target': 'local'} - - # agent_registry stores the runtime map alongside capabilities. - agent = agent_registry._agents.get(server_id) if hasattr(agent_registry, '_agents') else None - runtimes = dict(getattr(agent, 'runtimes', {})) if agent else {} - - unmet = [] - for name, spec in require.items(): - actual = runtimes.get(name) or '' - if not _runtime_satisfies(actual, str(spec)): - unmet.append({'runtime': name, 'required': spec, 'actual': actual or 'not installed'}) - - if unmet: - return { - 'success': False, - 'error': 'runtime requirements not met', - 'unmet': unmet, - 'critical': True, - } - return {'success': True, 'satisfied': require} - - # ------------------------------------------------------------------ - # Logic If Evaluation - # ------------------------------------------------------------------ - - @staticmethod - def _execute_logic_if(node: Dict, context: Dict, results: Dict) -> Dict: - """ - Evaluate a logic_if condition. - - The condition is a Python expression that has access to: - - results: dict of {node_id: node_result} - - context: the workflow execution context - """ - node_data = node.get('data', {}) - condition = node_data.get('condition', '').strip() - - if not condition: - return {'success': True, 'branch': 'true'} - - # Build a safe evaluation namespace - eval_globals = {"__builtins__": {}} - eval_locals = { - 'results': results, - 'context': context, - # Expose common helpers - 'len': len, - 'str': str, - 'int': int, - 'float': float, - 'bool': bool, - 'abs': abs, - 'min': min, - 'max': max, - 'any': any, - 'all': all, - 'isinstance': isinstance, - } - - try: - result = eval(condition, eval_globals, eval_locals) - branch = 'true' if result else 'false' - return { - 'success': True, - 'branch': branch, - 'condition': condition, - 'evaluated': bool(result) - } - except Exception as e: - return { - 'success': False, - 'error': f"Condition evaluation failed: {str(e)}", - 'condition': condition, - 'branch': 'false', - 'critical': False # Don't kill the whole workflow for a bad condition - } - - # ------------------------------------------------------------------ - # Variable Interpolation - # ------------------------------------------------------------------ - - @staticmethod - def _interpolate(text: str, context: Dict, results: Dict, - execution: Optional[WorkflowExecution] = None) -> str: - """ - Replace variable placeholders in text. - - Supported syntax: - - ${node_id.field} — access a specific field from a node's result - - ${node_id.output} — shorthand for the output field - - {{workflow_name}} — built-in workflow variables - - {{execution_id}} — current execution ID - - {{started_at}} — execution start time - - {{context.field}} — access context fields - """ - if not text or not isinstance(text, str): - return text - - # Replace ${node_id.field} patterns - def replace_node_var(match): - node_id = match.group(1) - field = match.group(2) - node_result = results.get(node_id, {}) - if field == 'output' and 'output' not in node_result: - # Try stdout as fallback for script nodes - return str(node_result.get('stdout', '')) - return str(node_result.get(field, '')) - - text = re.sub(r'\$\{([^.}]+)\.([^}]+)\}', replace_node_var, text) - - # Replace {{builtin}} patterns - builtins = { - 'workflow_name': execution.workflow.name if execution else '', - 'execution_id': str(execution.id) if execution else '', - 'started_at': execution.started_at.isoformat() if execution and execution.started_at else '', - 'trigger_type': execution.trigger_type if execution else '', - } - - # Add context.* variables - for key, value in context.items(): - builtins[f'context.{key}'] = str(value) - - for key, value in builtins.items(): - text = text.replace('{{' + key + '}}', value) - - # Replace {{node_id.field}} as alternative syntax - def replace_node_var_braces(match): - node_id = match.group(1) - field = match.group(2) - node_result = results.get(node_id, {}) - return str(node_result.get(field, '')) - - text = re.sub(r'\{\{([^.}]+)\.([^}]+)\}\}', replace_node_var_braces, text) - - return text - - # ------------------------------------------------------------------ - # Notification Node - # ------------------------------------------------------------------ - - @staticmethod - def _execute_notification(node: Dict, execution: WorkflowExecution, - context: Dict, results: Dict) -> Dict: - """Execute a notification node with variable interpolation.""" - node_data = node.get('data', {}) - channel = node_data.get('channel', 'system') - message = node_data.get('message', 'Workflow notification') - - # Interpolate variables in the message - message = WorkflowEngine._interpolate(message, context, results, execution) - - title = f"Workflow: {execution.workflow.name}" - - # Build alert in the format NotificationService expects - alerts = [{ - 'type': 'workflow', - 'severity': 'info', - 'message': message, - 'value': '', - 'threshold': '' - }] - - try: - if channel == 'system' or channel == 'all': - result = NotificationService.send_all(alerts) - elif channel == 'discord': - config = NotificationService.get_config().get('discord', {}) - result = NotificationService.send_discord(alerts, config) - elif channel == 'slack': - config = NotificationService.get_config().get('slack', {}) - result = NotificationService.send_slack(alerts, config) - elif channel == 'email': - config = NotificationService.get_config().get('email', {}) - result = NotificationService.send_email(alerts, config) - elif channel == 'telegram': - config = NotificationService.get_config().get('telegram', {}) - result = NotificationService.send_telegram(alerts, config) - else: - result = NotificationService.send_all(alerts) - - return {'success': result.get('success', True), 'channel': channel} - except Exception as e: - return {'success': False, 'error': str(e), 'critical': False} - - # ------------------------------------------------------------------ - # Script Node (Sandboxed) - # ------------------------------------------------------------------ - - @staticmethod - def _execute_script(node: Dict, execution: WorkflowExecution, - context: Dict, results: Dict) -> Dict: - """Execute a script node with timeout, output limits, and variable interpolation.""" - node_data = node.get('data', {}) - script_type = node_data.get('language', 'bash') - content = node_data.get('content', '') - timeout = min(int(node_data.get('timeout', DEFAULT_TIMEOUT)), MAX_TIMEOUT) - - if not content.strip(): - return {'success': True, 'stdout': '', 'stderr': '', 'returncode': 0} - - # Interpolate variables in script content - content = WorkflowEngine._interpolate(content, context, results, execution) - - # Build environment with node results available as env vars - env = os.environ.copy() - env['WORKFLOW_ID'] = str(execution.workflow_id) - env['EXECUTION_ID'] = str(execution.id) - env['TRIGGER_TYPE'] = execution.trigger_type or 'manual' - - for nid, nresult in results.items(): - safe_id = re.sub(r'[^a-zA-Z0-9_]', '_', nid).upper() - if isinstance(nresult, dict): - stdout = nresult.get('stdout', nresult.get('output', '')) - if isinstance(stdout, str): - env[f'NODE_{safe_id}_OUTPUT'] = stdout[:4096] - rc = nresult.get('returncode') - if rc is not None: - env[f'NODE_{safe_id}_RC'] = str(rc) - - try: - if script_type == 'bash': - cmd = ['bash', '-c', content] - elif script_type == 'python': - cmd = ['python3', '-c', content] - else: - return {'success': False, 'error': f"Unsupported script language: {script_type}"} - - proc = subprocess.run( - cmd, - capture_output=True, - text=True, - timeout=timeout, - env=env, - cwd='/tmp' if os.name != 'nt' else None - ) - - stdout = proc.stdout[:MAX_OUTPUT_SIZE] if proc.stdout else '' - stderr = proc.stderr[:MAX_OUTPUT_SIZE] if proc.stderr else '' - - return { - 'success': proc.returncode == 0, - 'stdout': stdout, - 'stderr': stderr, - 'returncode': proc.returncode - } - - except subprocess.TimeoutExpired: - return { - 'success': False, - 'error': f"Script timed out after {timeout}s", - 'stdout': '', - 'stderr': '', - 'returncode': -1 - } - except FileNotFoundError: - fallback = 'python' if script_type == 'python' else script_type - try: - proc = subprocess.run( - [fallback, '-c', content] if script_type == 'python' else content, - capture_output=True, text=True, timeout=timeout, - env=env, shell=(script_type == 'bash'), - cwd='/tmp' if os.name != 'nt' else None - ) - stdout = proc.stdout[:MAX_OUTPUT_SIZE] if proc.stdout else '' - stderr = proc.stderr[:MAX_OUTPUT_SIZE] if proc.stderr else '' - return { - 'success': proc.returncode == 0, - 'stdout': stdout, 'stderr': stderr, - 'returncode': proc.returncode - } - except Exception as e: - return {'success': False, 'error': str(e)} - except Exception as e: - return {'success': False, 'error': str(e)} - - # ------------------------------------------------------------------ - # Logging - # ------------------------------------------------------------------ - - @staticmethod - def _log(execution_id: int, message: str, level: str = 'INFO', node_id: str = None): - """Add a log entry for an execution.""" - log_entry = WorkflowLog( - execution_id=execution_id, - level=level, - message=message, - node_id=node_id - ) - db.session.add(log_entry) - db.session.commit() - logger.info(f"[{level}] Workflow {execution_id}: {message}") - - # Keep backward-compatible alias - log = _log - - -# ------------------------------------------------------------------ -# Event Bus for workflow triggers -# ------------------------------------------------------------------ - -class WorkflowEventBus: - """ - Simple in-process event bus for triggering workflows on system events. - - Events are emitted by services (monitoring, health checks, git deploy) - and matched against workflows with trigger_type='event'. - """ - - _listeners_lock = threading.Lock() - - @staticmethod - def emit(event_type: str, data: Dict[str, Any] = None): - """Emit a system event that may trigger event-subscribed workflows. - - Non-blocking: enqueues a single ``workflow.dispatch`` job that fans out to - the matching workflows on the unified job system (replacing the former - per-event daemon thread). Best-effort — never raises into the caller. - - Args: - event_type: One of health_check_failed, high_cpu, high_memory, - git_push, app_stopped, or any custom string. - data: Event payload passed as workflow context. - """ - try: - from app.jobs.service import JobService - JobService.enqueue( - WORKFLOW_DISPATCH_JOB_KIND, - payload={'event_type': event_type, 'data': data or {}}, - max_attempts=1, - owner_type='workflow_event', - owner_id=event_type, - ) - except Exception as e: - # No app context, or the queue is unavailable — never break the caller. - logger.warning(f"WorkflowEventBus.emit could not enqueue '{event_type}': {e}") - - @staticmethod - def dispatch_event(job): - """Unified-job handler for ``workflow.dispatch`` — find workflows - subscribed to the event and enqueue a workflow.execute job for each - (honoring the per-workflow 60s cooldown).""" - payload = job.get_payload() or {} - event_type = payload.get('event_type') - data = payload.get('data') or {} - if not event_type: - return {'event_type': None, 'triggered': 0} - - workflows = Workflow.query.filter_by(is_active=True, trigger_type='event').all() - triggered = 0 - for workflow in workflows: - try: - config = json.loads(workflow.trigger_config) if workflow.trigger_config else {} - if config.get('eventType', '') != event_type: - continue - # Cooldown: don't re-trigger within 60 seconds. - if workflow.last_run_at and \ - (datetime.utcnow() - workflow.last_run_at).total_seconds() < 60: - continue - logger.info(f"Event '{event_type}' triggering workflow: {workflow.name}") - WorkflowEngine.enqueue_execution( - workflow_id=workflow.id, - trigger_type='event', - context={ - 'event_type': event_type, - 'event_data': data, - 'triggered_at': datetime.utcnow().isoformat(), - }, - ) - triggered += 1 - except Exception as e: - logger.error(f"Event trigger failed for workflow {workflow.id}: {e}") - return {'event_type': event_type, 'triggered': triggered} diff --git a/backend/app/services/workflow_service.py b/backend/app/services/workflow_service.py deleted file mode 100644 index 0c0b8f8e..00000000 --- a/backend/app/services/workflow_service.py +++ /dev/null @@ -1,621 +0,0 @@ -""" -Workflow Deployment Service - -Orchestrates deployment of workflows by converting visual nodes and edges -into actual infrastructure resources (Docker apps, databases, domains). -""" - -import json -import os -from app import db -from app.models import Workflow, Application, Domain, User -from app.services.docker_service import DockerService -from app.services.database_service import DatabaseService -from app import paths - - -class WorkflowService: - """Service for deploying workflows to infrastructure.""" - - # Base path for app deployments - APP_BASE_PATH = paths.APPS_DIR - - @staticmethod - def deploy_workflow(workflow_id, user_id): - """ - Deploy all resources from a workflow. - - Args: - workflow_id: ID of the workflow to deploy - user_id: ID of the user deploying - - Returns: - dict with results, errors, and updated workflow - """ - workflow = Workflow.query.get(workflow_id) - if not workflow: - return {'success': False, 'error': 'Workflow not found'} - - # Verify user access - user = User.query.get(user_id) - if not user: - return {'success': False, 'error': 'User not found'} - - if user.role != 'admin' and workflow.user_id != user_id: - return {'success': False, 'error': 'Access denied'} - - # Parse workflow data - nodes = json.loads(workflow.nodes) if workflow.nodes else [] - edges = json.loads(workflow.edges) if workflow.edges else [] - - if not nodes: - return {'success': False, 'error': 'Workflow has no nodes to deploy'} - - # Get deployment order - ordered_nodes = WorkflowService.get_deployment_order(nodes, edges) - - # Track results and created resources - results = [] - errors = [] - created_resources = {} # nodeId -> resourceId mapping - - # Deploy each node - for node in ordered_nodes: - result = WorkflowService.deploy_node( - node, edges, user_id, created_resources - ) - results.append(result) - - if result['success']: - created_resources[node['id']] = { - 'type': node['type'], - 'resourceId': result.get('resourceId'), - 'resourceName': result.get('resourceName'), - 'credentials': result.get('credentials'), # Store DB credentials for apps - 'rootPath': result.get('rootPath') - } - else: - errors.append({ - 'nodeId': node['id'], - 'type': node['type'], - 'error': result.get('error') - }) - - # Update workflow nodes with created resource IDs - updated_nodes = WorkflowService.update_nodes_with_ids(nodes, created_resources) - workflow.nodes = json.dumps(updated_nodes) - db.session.commit() - - return { - 'success': len(errors) == 0, - 'message': 'Deployment completed' if len(errors) == 0 else 'Deployment completed with errors', - 'results': results, - 'errors': errors, - 'workflow': workflow.to_dict() - } - - @staticmethod - def get_deployment_order(nodes, edges): - """ - Determine deployment order using topological sort. - - Order: databases -> apps -> domains - - Args: - nodes: List of workflow nodes - edges: List of workflow edges - - Returns: - List of nodes in deployment order - """ - # Separate by type - databases = [n for n in nodes if n['type'] == 'database'] - apps = [n for n in nodes if n['type'] in ('dockerApp', 'service')] - domains = [n for n in nodes if n['type'] == 'domain'] - - # Order: databases first, then apps, then domains - return databases + apps + domains - - @staticmethod - def deploy_node(node, edges, user_id, created_resources): - """ - Deploy a single node based on its type. - - Args: - node: Node to deploy - edges: All workflow edges - user_id: User ID for ownership - created_resources: Already created resources mapping - - Returns: - dict with success, resourceId, etc. - """ - node_type = node.get('type') - node_data = node.get('data', {}) - - try: - if node_type == 'database': - return WorkflowService.deploy_database(node, user_id) - elif node_type == 'dockerApp': - return WorkflowService.deploy_docker_app(node, edges, user_id, created_resources) - elif node_type == 'service': - return WorkflowService.deploy_service(node, edges, user_id, created_resources) - elif node_type == 'domain': - return WorkflowService.deploy_domain(node, edges, user_id, created_resources) - else: - return { - 'nodeId': node['id'], - 'type': node_type, - 'success': False, - 'error': f'Unknown node type: {node_type}' - } - except Exception as e: - return { - 'nodeId': node['id'], - 'type': node_type, - 'success': False, - 'error': str(e) - } - - @staticmethod - def deploy_database(node, user_id): - """ - Deploy a database node. - - For MVP: Creates database using existing database service. - Note: Databases don't have a model in the current system, - so we track by name rather than ID. - - Args: - node: Database node - user_id: User ID - - Returns: - dict with deployment result - """ - node_data = node.get('data', {}) - db_name = node_data.get('name', '').replace(' ', '_').replace('-', '_').lower() - db_type = node_data.get('type', 'mysql') - - if not db_name: - return { - 'nodeId': node['id'], - 'type': 'database', - 'success': False, - 'error': 'Database name is required' - } - - # Check if already deployed (has dbName set) - if node_data.get('deployed'): - return { - 'nodeId': node['id'], - 'type': 'database', - 'success': True, - 'resourceId': None, - 'resourceName': db_name, - 'message': 'Database already deployed' - } - - try: - if db_type == 'mysql': - # Create MySQL database with user - result = DatabaseService.mysql_create_database( - db_name, - charset='utf8mb4', - collation='utf8mb4_unicode_ci' - ) - - if result.get('success'): - # Also create a user for the database - password = DatabaseService.generate_password() - user_result = DatabaseService.mysql_create_user( - username=db_name, - password=password, - host='%' - ) - - if user_result.get('success'): - # Grant privileges - DatabaseService.mysql_grant_privileges( - username=db_name, - database=db_name, - privileges='ALL', - host='%' - ) - - return { - 'nodeId': node['id'], - 'type': 'database', - 'success': True, - 'resourceId': None, - 'resourceName': db_name, - 'credentials': { - 'database': db_name, - 'username': db_name, - 'password': password, - 'host': 'localhost', - 'port': 3306 - } - } - else: - return { - 'nodeId': node['id'], - 'type': 'database', - 'success': False, - 'error': result.get('error', 'Failed to create MySQL database') - } - - elif db_type == 'postgresql': - result = DatabaseService.pg_create_database(db_name) - - if result.get('success'): - password = DatabaseService.generate_password() - user_result = DatabaseService.pg_create_user( - username=db_name, - password=password - ) - - if user_result.get('success'): - DatabaseService.pg_grant_privileges( - username=db_name, - database=db_name, - privileges='ALL' - ) - - return { - 'nodeId': node['id'], - 'type': 'database', - 'success': True, - 'resourceId': None, - 'resourceName': db_name, - 'credentials': { - 'database': db_name, - 'username': db_name, - 'password': password, - 'host': 'localhost', - 'port': 5432 - } - } - else: - return { - 'nodeId': node['id'], - 'type': 'database', - 'success': False, - 'error': result.get('error', 'Failed to create PostgreSQL database') - } - - else: - # MongoDB and Redis would be handled as Docker containers - return { - 'nodeId': node['id'], - 'type': 'database', - 'success': False, - 'error': f'{db_type} databases should be deployed as Docker containers' - } - - except Exception as e: - return { - 'nodeId': node['id'], - 'type': 'database', - 'success': False, - 'error': str(e) - } - - @staticmethod - def deploy_docker_app(node, edges, user_id, created_resources): - """ - Deploy a Docker app node. - - Args: - node: Docker app node - edges: Workflow edges (to find database connections) - user_id: User ID - created_resources: Already created resources - - Returns: - dict with deployment result - """ - node_data = node.get('data', {}) - app_name = node_data.get('name', '').replace(' ', '-').lower() - image = node_data.get('image', 'nginx:latest') - ports = node_data.get('ports', []) - memory = node_data.get('memory') - - if not app_name: - return { - 'nodeId': node['id'], - 'type': 'dockerApp', - 'success': False, - 'error': 'App name is required' - } - - # Check if already deployed - existing_app_id = node_data.get('appId') - if existing_app_id: - app = Application.query.get(existing_app_id) - if app: - # Update existing app - if image: - app.docker_image = image - if ports: - # Extract host port from first port mapping - first_port = ports[0] if ports else None - if first_port and ':' in first_port: - app.port = int(first_port.split(':')[0]) - db.session.commit() - - return { - 'nodeId': node['id'], - 'type': 'dockerApp', - 'success': True, - 'resourceId': app.id, - 'resourceName': app.name, - 'message': 'App updated' - } - - # Find connected database for environment variables - env_vars = {} - for edge in edges: - if edge.get('source') == node['id']: - # This app connects to something - target_id = edge.get('target') - target_resource = created_resources.get(target_id) - - if target_resource and target_resource['type'] == 'database': - # Get database credentials from the deployment result - credentials = target_resource.get('credentials', {}) - if credentials: - env_vars.update({ - 'DB_HOST': credentials.get('host', 'localhost'), - 'DB_NAME': credentials.get('database', ''), - 'DB_USER': credentials.get('username', ''), - 'DB_PASSWORD': credentials.get('password', ''), - 'DB_PORT': str(credentials.get('port', 3306)) - }) - else: - # Fallback to naming convention - db_name = target_resource.get('resourceName', '') - env_vars.update({ - 'DB_HOST': 'localhost', - 'DB_NAME': db_name, - 'DB_USER': db_name, - 'DB_PORT': '3306' - }) - - # Create the application record - app = Application( - name=app_name, - app_type='docker', - status='stopped', - docker_image=image, - user_id=user_id - ) - - # Extract host port from first port mapping - if ports: - first_port = ports[0] - if ':' in first_port: - app.port = int(first_port.split(':')[0]) - else: - app.port = int(first_port) - - # Set root path - app.root_path = os.path.join(WorkflowService.APP_BASE_PATH, app_name) - - db.session.add(app) - db.session.commit() - - # Create Docker compose files - docker_result = DockerService.create_docker_app( - app_path=app.root_path, - app_name=app_name, - image=image, - ports=ports if ports else None, - env=env_vars if env_vars else None - ) - - if not docker_result.get('success'): - # Rollback application creation - db.session.delete(app) - db.session.commit() - return { - 'nodeId': node['id'], - 'type': 'dockerApp', - 'success': False, - 'error': docker_result.get('error', 'Failed to create Docker app') - } - - # Start the containers - start_result = DockerService.compose_up(app.root_path, detach=True) - if start_result.get('success'): - app.status = 'running' - db.session.commit() - else: - # App created but containers didn't start - still return success - # but include warning - return { - 'nodeId': node['id'], - 'type': 'dockerApp', - 'success': True, - 'resourceId': app.id, - 'resourceName': app.name, - 'port': app.port, - 'rootPath': app.root_path, - 'warning': f'App created but containers failed to start: {start_result.get("error", "Unknown error")}' - } - - return { - 'nodeId': node['id'], - 'type': 'dockerApp', - 'success': True, - 'resourceId': app.id, - 'resourceName': app.name, - 'port': app.port, - 'rootPath': app.root_path, - 'status': 'running' - } - - @staticmethod - def deploy_service(node, edges, user_id, created_resources): - """ - Deploy a service node. - - Services are similar to Docker apps but may have specialized handling. - For now, treat them as Docker apps. - """ - return WorkflowService.deploy_docker_app(node, edges, user_id, created_resources) - - @staticmethod - def deploy_domain(node, edges, user_id, created_resources): - """ - Deploy a domain node. - - Domains must be connected to an app. Find the connected app - from edges and create the domain linked to that app. - - Args: - node: Domain node - edges: Workflow edges - user_id: User ID - created_resources: Already created resources - - Returns: - dict with deployment result - """ - node_data = node.get('data', {}) - domain_name = node_data.get('name', '') - ssl_status = node_data.get('ssl', 'none') - - if not domain_name: - return { - 'nodeId': node['id'], - 'type': 'domain', - 'success': False, - 'error': 'Domain name is required' - } - - # Check if already deployed - existing_domain_id = node_data.get('domainId') - if existing_domain_id: - domain = Domain.query.get(existing_domain_id) - if domain: - return { - 'nodeId': node['id'], - 'type': 'domain', - 'success': True, - 'resourceId': domain.id, - 'resourceName': domain.name, - 'message': 'Domain already exists' - } - - # Find the app this domain connects to - target_app_id = None - for edge in edges: - if edge.get('source') == node['id']: - target_id = edge.get('target') - target_resource = created_resources.get(target_id) - - if target_resource and target_resource['type'] in ('dockerApp', 'service'): - target_app_id = target_resource.get('resourceId') - break - - if not target_app_id: - return { - 'nodeId': node['id'], - 'type': 'domain', - 'success': False, - 'error': 'Domain must be connected to an app' - } - - # Verify the app exists - app = Application.query.get(target_app_id) - if not app: - return { - 'nodeId': node['id'], - 'type': 'domain', - 'success': False, - 'error': 'Connected app not found' - } - - # Check if domain already exists - existing = Domain.query.filter_by(name=domain_name).first() - if existing: - return { - 'nodeId': node['id'], - 'type': 'domain', - 'success': False, - 'error': f'Domain {domain_name} already exists' - } - - # Create the domain - domain = Domain( - name=domain_name, - is_primary=True, - ssl_enabled=ssl_status == 'valid', - ssl_auto_renew=True, - application_id=target_app_id - ) - - db.session.add(domain) - db.session.commit() - - # Create nginx config - from app.services.nginx_service import NginxService - - if app.port: - all_domains = [d.name for d in Domain.query.filter_by(application_id=target_app_id).all()] - nginx_result = NginxService.create_site( - name=app.name, - app_type='docker', - domains=all_domains, - root_path=app.root_path or '', - port=app.port - ) - - if nginx_result.get('success'): - NginxService.enable_site(app.name) - - return { - 'nodeId': node['id'], - 'type': 'domain', - 'success': True, - 'resourceId': domain.id, - 'resourceName': domain.name, - 'applicationId': target_app_id - } - - @staticmethod - def update_nodes_with_ids(nodes, created_resources): - """ - Update node data with created resource IDs. - - Args: - nodes: Original nodes list - created_resources: Mapping of nodeId to resource info - - Returns: - Updated nodes list - """ - updated = [] - for node in nodes: - node_copy = dict(node) - resource = created_resources.get(node['id']) - - if resource: - data = dict(node_copy.get('data', {})) - - if resource['type'] == 'dockerApp' or resource['type'] == 'service': - data['appId'] = resource.get('resourceId') - data['deployed'] = True - elif resource['type'] == 'domain': - data['domainId'] = resource.get('resourceId') - data['deployed'] = True - elif resource['type'] == 'database': - data['deployed'] = True - data['dbName'] = resource.get('resourceName') - - node_copy['data'] = data - - updated.append(node_copy) - - return updated diff --git a/backend/tests/test_jobs.py b/backend/tests/test_jobs.py index 8c02b09e..e22bbf0f 100644 --- a/backend/tests/test_jobs.py +++ b/backend/tests/test_jobs.py @@ -392,71 +392,6 @@ class TestDeploymentInstallHandler: assert unified.correlation_id == dep.correlation_id -class TestWorkflowExecutionHandler: - """Phase 6 — workflow executions run as a 'workflow.execute' unified job - instead of inline/threaded. Empty-node workflows run for real to success; - the failure path patches _run_execution to blow up.""" - - def _make_workflow(self, nodes='[]', edges='[]'): - from app.models.workflow import Workflow - wf = Workflow(name='wf-test', user_id=1, nodes=nodes, edges=edges, - is_active=True, trigger_type='manual') - db.session.add(wf) - db.session.commit() - return wf - - def test_register_jobs_adds_handler(self, app): - from app.services.workflow_engine import WorkflowEngine, WORKFLOW_JOB_KIND - WorkflowEngine.register_jobs() - assert WORKFLOW_JOB_KIND == 'workflow.execute' - assert registry.is_registered('workflow.execute') - - def test_enqueue_creates_running_row_and_job(self, app): - from app.services.workflow_engine import WorkflowEngine - from app.models.workflow import WorkflowExecution - wf = self._make_workflow() - - execution_id = WorkflowEngine.enqueue_execution(wf.id, trigger_type='manual') - - ex = WorkflowExecution.query.get(execution_id) - assert ex is not None and ex.status == 'running' - unified = Job.query.filter_by(kind='workflow.execute', owner_id=str(wf.id)).first() - assert unified is not None - assert unified.get_payload() == {'execution_id': execution_id} - assert unified.max_attempts == 1 - - def test_handler_runs_empty_workflow_to_success(self, app): - from app.services.workflow_engine import WorkflowEngine - from app.models.workflow import WorkflowExecution - WorkflowEngine.register_jobs() - wf = self._make_workflow(nodes='[]', edges='[]') - - execution_id = WorkflowEngine.enqueue_execution(wf.id) - _drain_once() - - assert WorkflowExecution.query.get(execution_id).status == 'success' - unified = Job.query.filter_by(kind='workflow.execute').first() - assert unified.status == Job.STATUS_SUCCEEDED - - def test_handler_failure_marks_job_failed(self, app, monkeypatch): - from app.services.workflow_engine import WorkflowEngine - from app.models.workflow import WorkflowExecution - - def boom(execution_id, nodes, edges): - raise RuntimeError('node blew up') - - monkeypatch.setattr(WorkflowEngine, '_run_execution', staticmethod(boom)) - WorkflowEngine.register_jobs() - wf = self._make_workflow(nodes='[{"id":"a","type":"trigger","data":{}}]', edges='[]') - - execution_id = WorkflowEngine.enqueue_execution(wf.id) - _drain_once() - - assert WorkflowExecution.query.get(execution_id).status == 'failed' - unified = Job.query.filter_by(kind='workflow.execute').first() - assert unified.status == Job.STATUS_FAILED - - class TestBackupScheduleHandler: """Phase 7 — scheduled backups run as 'backup.run' unified jobs, enqueued by the builtin.backup_scheduler tick (replacing the orphaned daemon loop). @@ -522,46 +457,7 @@ class TestBackupScheduleHandler: assert Job.query.get(bad.id).status == Job.STATUS_FAILED -class TestWorkflowEventDispatch: - """Event-bus cleanup — WorkflowEventBus.emit enqueues a 'workflow.dispatch' - job instead of spawning a per-event thread; the handler fans out to matching - event-subscribed workflows.""" - - def _make_event_workflow(self, event_type): - import json as _json - from app.models.workflow import Workflow - wf = Workflow(name=f'evt-{event_type}', user_id=1, nodes='[]', edges='[]', - is_active=True, trigger_type='event', - trigger_config=_json.dumps({'eventType': event_type})) - db.session.add(wf) - db.session.commit() - return wf - - def test_register_adds_dispatch_handler(self, app): - from app.services.workflow_engine import WorkflowEngine - WorkflowEngine.register_jobs() - assert registry.is_registered('workflow.dispatch') - - def test_emit_enqueues_dispatch_job(self, app): - from app.services.workflow_engine import WorkflowEventBus - WorkflowEventBus.emit('high_cpu', {'value': 95}) - evt_jobs = Job.query.filter_by(kind='workflow.dispatch').all() - assert len(evt_jobs) == 1 - assert evt_jobs[0].get_payload() == {'event_type': 'high_cpu', 'data': {'value': 95}} - assert evt_jobs[0].owner_id == 'high_cpu' - - def test_dispatch_fans_out_to_matching_workflow_only(self, app): - from app.services.workflow_engine import WorkflowEngine, WorkflowEventBus - from app.models.workflow import WorkflowExecution - WorkflowEngine.register_jobs() - match = self._make_event_workflow('high_cpu') - self._make_event_workflow('git_push') # different event — must be ignored - - WorkflowEventBus.emit('high_cpu', {'value': 95}) - _drain_once() # process workflow.dispatch -> enqueues workflow.execute - - execs = WorkflowExecution.query.all() - assert len(execs) == 1 - assert execs[0].workflow_id == match.id - assert execs[0].trigger_type == 'event' - assert Job.query.filter_by(kind='workflow.execute').count() == 1 +# TestWorkflowEventDispatch removed in plan 45 Phase 4: the React-Flow Workflow +# Builder engine (WorkflowEngine / WorkflowEventBus) was retired. Its four event +# emitters were ported to EventService (health.check_failed, git.push, +# monitor.high_cpu/high_memory); those are covered in test_tramo_extension.py.