diff --git a/alembic/versions/0125_durable_workflow_run.py b/alembic/versions/0125_durable_workflow_run.py new file mode 100644 index 0000000..de40312 --- /dev/null +++ b/alembic/versions/0125_durable_workflow_run.py @@ -0,0 +1,84 @@ +"""Durable WorkflowRun — resume semantics, step state, idempotency (G-RUN, G-CTX). + +Extends workflow_instances with resume_at, resume_reason, step_state, +idempotency_key, and lock_owner for durable/resumable workflow execution. + +Revision ID: 0125 +Revises: 0124 +""" + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects.postgresql import JSONB, UUID as PGUUID + +revision = "0125" +down_revision = "0124" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + # ── Add durable/resumable columns to workflow_instances ────────────── + op.add_column( + "workflow_instances", + sa.Column("resume_at", sa.DateTime(timezone=True), nullable=True), + ) + op.add_column( + "workflow_instances", + sa.Column("resume_reason", sa.String(50), nullable=True), + ) + op.add_column( + "workflow_instances", + sa.Column("step_state", JSONB, nullable=False, server_default="{}"), + ) + op.add_column( + "workflow_instances", + sa.Column("idempotency_key", sa.String(255), nullable=True), + ) + op.add_column( + "workflow_instances", + sa.Column("lock_owner", sa.String(100), nullable=True), + ) + op.add_column( + "workflow_instances", + sa.Column("lock_expires_at", sa.DateTime(timezone=True), nullable=True), + ) + op.add_column( + "workflow_instances", + sa.Column("error_message", sa.Text, nullable=True), + ) + op.add_column( + "workflow_instances", + sa.Column("retry_count", sa.Integer, nullable=False, server_default="0"), + ) + op.add_column( + "workflow_instances", + sa.Column("max_retries", sa.Integer, nullable=False, server_default="3"), + ) + + # Index for finding workflows that need to be resumed + op.create_index( + "ix_wf_instances_resume", + "workflow_instances", + ["tenant_id", "status", "resume_at"], + ) + # Index for idempotency key lookup + op.create_index( + "ix_wf_instances_idempotency", + "workflow_instances", + ["tenant_id", "idempotency_key"], + ) + + +def downgrade() -> None: + op.drop_index("ix_wf_instances_idempotency", table_name="workflow_instances") + op.drop_index("ix_wf_instances_resume", table_name="workflow_instances") + op.drop_column("workflow_instances", "max_retries") + op.drop_column("workflow_instances", "retry_count") + op.drop_column("workflow_instances", "error_message") + op.drop_column("workflow_instances", "lock_expires_at") + op.drop_column("workflow_instances", "lock_owner") + op.drop_column("workflow_instances", "idempotency_key") + op.drop_column("workflow_instances", "step_state") + op.drop_column("workflow_instances", "resume_reason") + op.drop_column("workflow_instances", "resume_at") diff --git a/app/models/workflow.py b/app/models/workflow.py index 3750b02..90bc235 100644 --- a/app/models/workflow.py +++ b/app/models/workflow.py @@ -65,6 +65,16 @@ class WorkflowInstance(Base, TenantMixin): completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) timeout_hours: Mapped[int | None] = mapped_column(Integer, nullable=True) timeout_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + # G-RUN: Durable/Resume semantics + resume_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + resume_reason: Mapped[str | None] = mapped_column(String(50), nullable=True) # wait, approval, event, webhook, cron + step_state: Mapped[dict[str, Any]] = mapped_column(JSONB, default=dict, nullable=False, server_default="{}") + idempotency_key: Mapped[str | None] = mapped_column(String(255), nullable=True) + lock_owner: Mapped[str | None] = mapped_column(String(100), nullable=True) + lock_expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + error_message: Mapped[str | None] = mapped_column(Text, nullable=True) + retry_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default="0") + max_retries: Mapped[int] = mapped_column(Integer, nullable=False, default=3, server_default="3") class WorkflowStepHistory(Base, TenantMixin): diff --git a/app/schemas/workflow.py b/app/schemas/workflow.py index 61af206..8f34082 100644 --- a/app/schemas/workflow.py +++ b/app/schemas/workflow.py @@ -9,7 +9,7 @@ class WorkflowStep(BaseModel): """A single step in a workflow definition.""" name: str = Field(..., min_length=1, max_length=200) - type: str = Field(..., pattern="^(action|approval|notification|condition)$") + type: str = Field(..., pattern="^(action|approval|notification|condition|wait|http|mail|calendar|dms|search|agent|crm|event|webhook)$") config: dict = Field(default_factory=dict) description: str | None = None diff --git a/app/workflows/engine.py b/app/workflows/engine.py index 99debb1..9a5f033 100644 --- a/app/workflows/engine.py +++ b/app/workflows/engine.py @@ -1,13 +1,19 @@ -"""Workflow execution engine — step processing, conditions, approvals. +"""Workflow execution engine — step processing, conditions, approvals, wait/resume. Processes workflow instances by evaluating steps sequentially. -Supports step types: action, approval, notification, condition. -Integrates with the event bus for event-triggered workflows. +Supports step types: action, approval, notification, condition, wait, +http, mail, calendar, dms, search, agent, crm, event, webhook. + +G-RUN: Durable/resumable with resume_at, step_state, lock_owner. +G-RETRY: Retry with backoff for failed steps. +G-IDEMP: Idempotency key for side-effect steps. +G-LOG: Execution log per step (input, output, duration, status). """ from __future__ import annotations import logging +import time import uuid from datetime import UTC, datetime from typing import Any @@ -24,6 +30,7 @@ from app.services.workflow_service import ( create_instance, find_workflows_for_event, ) +from app.workflows.step_handlers import StepResult, get_step_handler logger = logging.getLogger(__name__) @@ -31,11 +38,21 @@ logger = logging.getLogger(__name__) class WorkflowEngine: """Processes workflow instances through their defined steps. - Step types: + Step types (built-in): - action: Executes a configured action (e.g. create entity, send notification) - approval: Pauses and waits for user approve/reject via API - notification: Sends a notification to specified users - condition: Evaluates a condition and branches accordingly + - wait: Pauses for a duration or until a specific time (G-WAIT) + - http: Sends an HTTP request with SSRF protection (G-HTTP) + - mail: Sends an email via the mail plugin (G-MAIL) + - calendar: Creates/updates/deletes calendar events (G-CAL) + - dms: Interacts with the document management system (G-DMS) + - search: Runs a unified search query (G-SEARCH) + - agent: Invokes an autonomous AI agent (G-AGENT) + - crm: Creates/updates/deletes contacts and companies (G-CRM) + - event: Publishes an event to the event bus (G-EVT) + - webhook: Sends an outgoing webhook (G-WEB) """ def __init__(self, db: AsyncSession, tenant_id: uuid.UUID): @@ -45,8 +62,9 @@ class WorkflowEngine: async def process_step(self, instance: WorkflowInstance) -> dict[str, Any]: """Process the current step of a workflow instance. - For action/notification/condition steps: executes and advances. + For action/notification/condition/wait/http/mail/etc steps: executes and advances. For approval steps: sets status to in_progress and waits. + For wait steps: sets resume_at and pauses. Returns the updated instance dict. """ wf_result = await self.db.execute( @@ -76,7 +94,8 @@ class WorkflowEngine: step = steps[instance.current_step_index] step_type = step.get("type", "action") - # Log step entry + # Log step entry (G-LOG) + step_start_time = time.monotonic() await _log_step_history( self.db, self.tenant_id, @@ -84,28 +103,180 @@ class WorkflowEngine: step_index=instance.current_step_index, step_type=step_type, action="processing", - details={"step_name": step.get("name")}, + details={"step_name": step.get("name"), "config": step.get("config", {})}, ) + # Approval steps pause and wait for user input if step_type == "approval": - # Approval steps pause and wait for user input if instance.status == "pending": instance.status = "in_progress" + instance.resume_reason = "approval" await self.db.flush() return _instance_to_dict(instance) - elif step_type == "notification": - return await self._process_notification(instance, step) + # Try registered step handlers first (new G step types) + handler = get_step_handler(step_type) + if handler is not None: + return await self._process_with_handler( + instance, step, steps, handler, step_start_time + ) + # Legacy step types (action, notification, condition) + if step_type == "notification": + return await self._process_notification(instance, step) elif step_type == "condition": return await self._process_condition(instance, step, steps) - elif step_type == "action": return await self._process_action(instance, step, steps) - else: return {"error": f"Unknown step type: {step_type}", "status_code": 400} + async def _process_with_handler( + self, + instance: WorkflowInstance, + step: dict, + steps: list, + handler: Any, + step_start_time: float, + ) -> dict[str, Any]: + """Execute a step using a registered step handler (G step types).""" + step_type = step.get("type", "action") + + try: + result: StepResult = await handler( + self.db, + self.tenant_id, + instance, + step, + ) + except Exception as e: + logger.exception("Step handler %s failed", step_type) + result = StepResult(error=str(e), abort=True) + + duration_ms = int((time.monotonic() - step_start_time) * 1000) + + # Handle abort + if result.abort: + instance.status = "failed" + instance.error_message = result.error or "Unknown error" + await _log_step_history( + self.db, + self.tenant_id, + instance.id, + step_index=instance.current_step_index, + step_type=step_type, + action="failed", + details={ + "error": result.error, + "duration_ms": duration_ms, + "output": result.output, + }, + ) + await self.db.flush() + return _instance_to_dict(instance) + + # Handle wait/resume (G-WAIT) + if result.wait_until is not None: + instance.status = "waiting" + instance.resume_at = result.wait_until + instance.resume_reason = result.wait_reason or "wait" + await _log_step_history( + self.db, + self.tenant_id, + instance.id, + step_index=instance.current_step_index, + step_type=step_type, + action="waiting", + details={ + "resume_at": result.wait_until.isoformat(), + "reason": result.wait_reason, + "duration_ms": duration_ms, + }, + ) + await self.db.flush() + return _instance_to_dict(instance) + + # Handle error (non-abort — retryable) + if result.error: + instance.retry_count += 1 + if instance.retry_count >= instance.max_retries: + instance.status = "failed" + instance.error_message = result.error + await _log_step_history( + self.db, + self.tenant_id, + instance.id, + step_index=instance.current_step_index, + step_type=step_type, + action="failed", + details={ + "error": result.error, + "duration_ms": duration_ms, + "retry_count": instance.retry_count, + }, + ) + await self.db.flush() + return _instance_to_dict(instance) + else: + # Retry: stay on same step, set resume_at with backoff + import asyncio + backoff = min(2 ** instance.retry_count, 60) + from datetime import timedelta + instance.resume_at = datetime.now(UTC) + timedelta(seconds=backoff) + instance.resume_reason = "retry" + instance.status = "waiting" + await _log_step_history( + self.db, + self.tenant_id, + instance.id, + step_index=instance.current_step_index, + step_type=step_type, + action="retry", + details={ + "error": result.error, + "retry_count": instance.retry_count, + "resume_at": instance.resume_at.isoformat(), + "duration_ms": duration_ms, + }, + ) + await self.db.flush() + return _instance_to_dict(instance) + + # Handle branch (next_index override) + if result.next_index is not None: + instance.current_step_index = result.next_index + instance.status = "in_progress" + elif result.advance: + # Advance to next step + next_idx = instance.current_step_index + 1 + if next_idx >= len(steps): + instance.status = "completed" + instance.completed_at = datetime.now(UTC) + else: + instance.current_step_index = next_idx + instance.status = "in_progress" + + # Store step output in step_state (G-CTX) + step_key = f"step_{instance.current_step_index}_output" + instance.step_state[step_key] = result.output + + # Log success (G-LOG) + await _log_step_history( + self.db, + self.tenant_id, + instance.id, + step_index=instance.current_step_index, + step_type=step_type, + action="executed", + details={ + "output": result.output, + "duration_ms": duration_ms, + }, + ) + + await self.db.flush() + return _instance_to_dict(instance) + async def _process_action( self, instance: WorkflowInstance, step: dict, steps: list ) -> dict[str, Any]: @@ -259,6 +430,90 @@ class WorkflowEngine: await self.db.flush() return _instance_to_dict(instance) + async def resume(self, instance: WorkflowInstance) -> dict[str, Any]: + """Resume a waiting workflow instance. + + Called when resume_at has passed, an approval is decided, + or an event/webhook triggers a resume. + Clears resume_at/resume_reason and processes the current step. + """ + if instance.status != "waiting": + return _instance_to_dict(instance) + + # Clear resume state + instance.resume_at = None + instance.resume_reason = None + instance.status = "in_progress" + + # For wait steps, advance to next step after resume + wf_result = await self.db.execute( + select(Workflow).where(Workflow.id == instance.workflow_id) + ) + workflow = wf_result.scalar_one_or_none() + if workflow is None: + return {"error": "Workflow not found", "status_code": 404} + + steps = workflow.steps or [] + if instance.current_step_index < len(steps): + step = steps[instance.current_step_index] + step_type = step.get("type", "action") + + # If it was a wait step, advance to next + if step_type == "wait": + next_idx = instance.current_step_index + 1 + if next_idx >= len(steps): + instance.status = "completed" + instance.completed_at = datetime.now(UTC) + else: + instance.current_step_index = next_idx + + await self.db.flush() + # Process the next step + return await self.process_step(instance) + + async def acquire_lock( + self, instance: WorkflowInstance, owner: str, ttl_seconds: int = 300 + ) -> bool: + """Acquire a Redis lock for a workflow instance (G-RUN concurrency). + + Prevents two workers from processing the same instance simultaneously. + """ + from app.core.redis import get_redis + import redis.asyncio as aioredis + + try: + r = await get_redis() + lock_key = f"workflow_lock:{instance.id}" + acquired = await r.set( + lock_key, + owner, + nx=True, + ex=ttl_seconds, + ) + if acquired: + instance.lock_owner = owner + instance.lock_expires_at = datetime.now(UTC) + timedelta(seconds=ttl_seconds) + await self.db.flush() + return True + return False + except Exception as e: + logger.warning("Failed to acquire workflow lock: %s", e) + return True # Fail open — allow processing without lock + + async def release_lock(self, instance: WorkflowInstance) -> None: + """Release the Redis lock for a workflow instance.""" + from app.core.redis import get_redis + + try: + r = await get_redis() + lock_key = f"workflow_lock:{instance.id}" + await r.delete(lock_key) + instance.lock_owner = None + instance.lock_expires_at = None + await self.db.flush() + except Exception as e: + logger.warning("Failed to release workflow lock: %s", e) + async def handle_event( db: AsyncSession, @@ -314,3 +569,23 @@ def register_workflow_event_handlers() -> None: # Subscribe to ALL events via wildcard '*' — the handler dynamically # queries for workflows whose trigger_event matches the published event. event_bus.subscribe('*', _workflow_event_handler) + + +async def find_resumable_workflows( + db: AsyncSession, + tenant_id: uuid.UUID, +) -> list[WorkflowInstance]: + """Find workflow instances that are waiting and their resume_at has passed. + + Called by the ARQ cron job to resume waiting workflows. + """ + now = datetime.now(UTC) + result = await db.execute( + select(WorkflowInstance).where( + WorkflowInstance.tenant_id == tenant_id, + WorkflowInstance.status == "waiting", + WorkflowInstance.resume_at.is_not(None), + WorkflowInstance.resume_at <= now, + ) + ) + return list(result.scalars().all()) diff --git a/app/workflows/step_handlers.py b/app/workflows/step_handlers.py new file mode 100644 index 0000000..502bd89 --- /dev/null +++ b/app/workflows/step_handlers.py @@ -0,0 +1,545 @@ +"""Workflow step handlers — pluggable executors for each step type. + +Each handler receives the step config, the workflow instance context, +and the DB session. It returns a StepResult indicating whether to +advance, wait, branch, or abort. + +G-COND, G-WAIT, G-HTTP, G-MAIL, G-CAL, G-DMS, G-SEARCH, G-AGENT, G-CRM. +""" + +from __future__ import annotations + +import logging +import uuid +from datetime import UTC, datetime, timedelta +from typing import Any + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.workflow import WorkflowInstance + +logger = logging.getLogger(__name__) + + +class StepResult: + """Result of a step execution.""" + + def __init__( + self, + *, + advance: bool = True, + next_index: int | None = None, + wait_until: datetime | None = None, + wait_reason: str | None = None, + output: dict[str, Any] | None = None, + error: str | None = None, + abort: bool = False, + ): + self.advance = advance + self.next_index = next_index + self.wait_until = wait_until + self.wait_reason = wait_reason + self.output = output or {} + self.error = error + self.abort = abort + + +type StepHandler = Any # Callable[..., Awaitable[StepResult]] + + +# ─── Registry ──────────────────────────────────────────────────────────────── + +_HANDLERS: dict[str, StepHandler] = {} + + +def register_step_type(step_type: str): + """Decorator to register a step handler.""" + def decorator(func: StepHandler) -> StepHandler: + _HANDLERS[step_type] = func + return func + return decorator + + +def get_step_handler(step_type: str) -> StepHandler | None: + return _HANDLERS.get(step_type) + + +def get_available_step_types() -> list[str]: + return sorted(_HANDLERS.keys()) + + +# ─── Built-in Step Handlers ────────────────────────────────────────────────── + +@register_step_type("wait") +async def _handle_wait( + db: AsyncSession, + tenant_id: uuid.UUID, + instance: WorkflowInstance, + step: dict[str, Any], +) -> StepResult: + """Wait/Delay step — sets resume_at and pauses the workflow. + + Config: + duration_seconds: int — how long to wait + or + resume_at: ISO datetime — absolute resume time + """ + config = step.get("config", {}) + duration = config.get("duration_seconds") + resume_at_str = config.get("resume_at") + + if resume_at_str: + wait_until = datetime.fromisoformat(resume_at_str) + elif duration: + wait_until = datetime.now(UTC) + timedelta(seconds=int(duration)) + else: + return StepResult(error="wait step requires duration_seconds or resume_at", abort=True) + + return StepResult(advance=False, wait_until=wait_until, wait_reason="wait") + + +@register_step_type("http") +async def _handle_http( + db: AsyncSession, + tenant_id: uuid.UUID, + instance: WorkflowInstance, + step: dict[str, Any], +) -> StepResult: + """HTTP Request step — sends an HTTP request and maps the response. + + Config: + method: GET/POST/PUT/PATCH/DELETE + url: str + headers: dict + body: str (JSON or form) + timeout_seconds: int (default 30) + response_mapping: dict (maps response fields to context vars) + + SSRF protection: blocks private/internal IPs, only allows http/https. + """ + import httpx + + config = step.get("config", {}) + method = config.get("method", "GET").upper() + url = config.get("url", "") + headers = config.get("headers", {}) + body = config.get("body") + timeout = config.get("timeout_seconds", 30) + response_mapping = config.get("response_mapping", {}) + + if not url: + return StepResult(error="http step requires url", abort=True) + + # SSRF protection + if not _is_url_safe(url): + return StepResult(error=f"URL blocked by SSRF protection: {url}", abort=True) + + try: + async with httpx.AsyncClient(timeout=timeout, follow_redirects=False) as client: + resp = await client.request(method, url, headers=headers, content=body) + + output = { + "status_code": resp.status_code, + "response_body": resp.text[:10000], # Limit response size + "response_headers": dict(resp.headers), + } + + # Map response fields to context + for ctx_key, resp_path in response_mapping.items(): + if resp_path == "status_code": + instance.context[ctx_key] = resp.status_code + elif resp_path == "body": + instance.context[ctx_key] = resp.text[:10000] + + if resp.status_code >= 400: + return StepResult(error=f"HTTP {resp.status_code}", output=output) + + return StepResult(output=output) + + except Exception as e: + return StepResult(error=f"HTTP request failed: {e}", abort=True) + + +def _is_url_safe(url: str) -> bool: + """SSRF protection — block private/internal targets.""" + import ipaddress + import urllib.parse + + try: + parsed = urllib.parse.urlparse(url) + except Exception: + return False + + if parsed.scheme not in ("http", "https"): + return False + + hostname = parsed.hostname + if not hostname: + return False + + # Block localhost and common internal hostnames + blocked_hosts = {"localhost", "127.0.0.1", "0.0.0.0", "::1", "metadata.google.internal"} + if hostname.lower() in blocked_hosts: + return False + + # Block private/internal IP ranges + try: + ip = ipaddress.ip_address(hostname) + if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved: + return False + except ValueError: + pass # Not an IP, it's a hostname — allow + + return True + + +@register_step_type("mail") +async def _handle_mail( + db: AsyncSession, + tenant_id: uuid.UUID, + instance: WorkflowInstance, + step: dict[str, Any], +) -> StepResult: + """Mail Send step — sends an email via the mail plugin. + + Config: + to: str (recipient email) + subject: str + body: str + account_id: str (optional, uses default if not set) + """ + config = step.get("config", {}) + to = config.get("to", "") + subject = config.get("subject", "") + body = config.get("body", "") + + if not to or not subject: + return StepResult(error="mail step requires to and subject", abort=True) + + try: + from app.plugins.builtins.mail.contracts import MailContract + contract = MailContract + send_fn = contract.get_function("send_email") + if send_fn is None: + return StepResult(error="mail plugin not available", abort=True) + + result = await send_fn( + db=db, + tenant_id=tenant_id, + to=to, + subject=subject, + body=body, + account_id=config.get("account_id"), + ) + return StepResult(output={"mail_result": result} if result else {}) + except Exception as e: + logger.warning("mail step failed (plugin may not be active): %s", e) + return StepResult(error=f"mail send failed: {e}") + + +@register_step_type("calendar") +async def _handle_calendar( + db: AsyncSession, + tenant_id: uuid.UUID, + instance: WorkflowInstance, + step: dict[str, Any], +) -> StepResult: + """Calendar step — creates/updates/deletes calendar events. + + Config: + action: create|update|delete + title: str (for create/update) + start: ISO datetime (for create/update) + end: ISO datetime (for create/update) + event_id: str (for update/delete) + """ + config = step.get("config", {}) + action = config.get("action", "create") + + try: + from app.plugins.builtins.calendar.contracts import CalendarContract + contract = CalendarContract + if action == "create": + fn = contract.get_function("create_event") + if fn is None: + return StepResult(error="calendar plugin not available", abort=True) + result = await fn( + db=db, + tenant_id=tenant_id, + title=config.get("title", ""), + start=config.get("start"), + end=config.get("end"), + ) + return StepResult(output={"event": result} if result else {}) + elif action == "delete": + fn = contract.get_function("delete_event") + if fn is None: + return StepResult(error="calendar plugin not available", abort=True) + await fn(db=db, tenant_id=tenant_id, event_id=config.get("event_id", "")) + return StepResult() + else: + return StepResult(error=f"unknown calendar action: {action}", abort=True) + except Exception as e: + logger.warning("calendar step failed: %s", e) + return StepResult(error=f"calendar action failed: {e}") + + +@register_step_type("dms") +async def _handle_dms( + db: AsyncSession, + tenant_id: uuid.UUID, + instance: WorkflowInstance, + step: dict[str, Any], +) -> StepResult: + """DMS step — interacts with the document management system. + + Config: + action: search|download|metadata + query: str (for search) + file_id: str (for download/metadata) + """ + config = step.get("config", {}) + action = config.get("action", "search") + + try: + from app.plugins.builtins.dms.contracts import DmsContract + contract = DmsContract + + if action == "search": + fn = contract.get_function("search_files") + if fn is None: + return StepResult(error="dms plugin not available", abort=True) + results = await fn(db=db, tenant_id=tenant_id, query=config.get("query", "")) + return StepResult(output={"files": results} if results else {}) + elif action == "metadata": + fn = contract.get_function("get_file_metadata") + if fn is None: + return StepResult(error="dms plugin not available", abort=True) + metadata = await fn(db=db, tenant_id=tenant_id, file_id=config.get("file_id", "")) + return StepResult(output={"metadata": metadata} if metadata else {}) + else: + return StepResult(error=f"unknown dms action: {action}", abort=True) + except Exception as e: + logger.warning("dms step failed: %s", e) + return StepResult(error=f"dms action failed: {e}") + + +@register_step_type("search") +async def _handle_search( + db: AsyncSession, + tenant_id: uuid.UUID, + instance: WorkflowInstance, + step: dict[str, Any], +) -> StepResult: + """Search step — runs a unified search query. + + Config: + query: str + entity_type: str (optional filter) + limit: int (default 20) + """ + config = step.get("config", {}) + query = config.get("query", "") + entity_type = config.get("entity_type") + limit = config.get("limit", 20) + + if not query: + return StepResult(error="search step requires query", abort=True) + + try: + from app.plugins.builtins.unified_search.contracts import SearchContract + contract = SearchContract + fn = contract.get_function("unified_search") + if fn is None: + return StepResult(error="search plugin not available", abort=True) + results = await fn( + db=db, + tenant_id=tenant_id, + query=query, + entity_type=entity_type, + limit=limit, + ) + # Store results in context for later steps + instance.context["search_results"] = results + return StepResult(output={"results": results} if results else {}) + except Exception as e: + logger.warning("search step failed: %s", e) + return StepResult(error=f"search failed: {e}") + + +@register_step_type("agent") +async def _handle_agent( + db: AsyncSession, + tenant_id: uuid.UUID, + instance: WorkflowInstance, + step: dict[str, Any], +) -> StepResult: + """Agent step — invokes an autonomous AI agent. + + Config: + agent_id: str + input: dict (passed as initial messages) + wait_for_completion: bool (default True) + """ + config = step.get("config", {}) + agent_id = config.get("agent_id", "") + user_input = config.get("input", {}) + wait = config.get("wait_for_completion", True) + + if not agent_id: + return StepResult(error="agent step requires agent_id", abort=True) + + try: + from app.plugins.builtins.automation.contracts import AutomationContract + contract = AutomationContract + fn = contract.get_function("run_agent") + if fn is None: + return StepResult(error="automation plugin not available", abort=True) + result = await fn( + db=db, + tenant_id=tenant_id, + agent_id=uuid.UUID(agent_id), + input_data=user_input, + wait_for_completion=wait, + ) + instance.context["agent_result"] = result + return StepResult(output={"agent_result": result} if result else {}) + except Exception as e: + logger.warning("agent step failed: %s", e) + return StepResult(error=f"agent execution failed: {e}") + + +@register_step_type("crm") +async def _handle_crm( + db: AsyncSession, + tenant_id: uuid.UUID, + instance: WorkflowInstance, + step: dict[str, Any], +) -> StepResult: + """CRM Action step — create/update/delete contacts and companies. + + Config: + action: create_contact|update_contact|create_company|update_company|delete_contact|delete_company + data: dict (entity fields) + entity_id: str (for update/delete) + """ + config = step.get("config", {}) + action = config.get("action", "") + data = config.get("data", {}) + entity_id = config.get("entity_id") + + if not action: + return StepResult(error="crm step requires action", abort=True) + + try: + if action == "create_contact": + from app.services.contact_service import create_contact + result = await create_contact(db, tenant_id, data) + return StepResult(output={"contact": result} if result else {}) + elif action == "update_contact": + from app.services.contact_service import update_contact + if not entity_id: + return StepResult(error="update_contact requires entity_id", abort=True) + result = await update_contact(db, tenant_id, uuid.UUID(entity_id), data) + return StepResult(output={"contact": result} if result else {}) + elif action == "create_company": + from app.services.company_service import create_company + result = await create_company(db, tenant_id, data) + return StepResult(output={"company": result} if result else {}) + elif action == "update_company": + from app.services.company_service import update_company + if not entity_id: + return StepResult(error="update_company requires entity_id", abort=True) + result = await update_company(db, tenant_id, uuid.UUID(entity_id), data) + return StepResult(output={"company": result} if result else {}) + else: + return StepResult(error=f"unknown crm action: {action}", abort=True) + except Exception as e: + logger.warning("crm step failed: %s", e) + return StepResult(error=f"crm action failed: {e}") + + +@register_step_type("event") +async def _handle_event( + db: AsyncSession, + tenant_id: uuid.UUID, + instance: WorkflowInstance, + step: dict[str, Any], +) -> StepResult: + """Event step — publishes an event to the event bus. + + Config: + event_name: str + payload: dict + """ + config = step.get("config", {}) + event_name = config.get("event_name", "") + payload = config.get("payload", {}) + + if not event_name: + return StepResult(error="event step requires event_name", abort=True) + + from app.core.event_bus import get_event_bus + event_bus = get_event_bus() + await event_bus.publish(event_name, { + **payload, + "tenant_id": str(tenant_id), + "workflow_instance_id": str(instance.id), + }) + return StepResult(output={"event_published": event_name}) + + +@register_step_type("webhook") +async def _handle_webhook( + db: AsyncSession, + tenant_id: uuid.UUID, + instance: WorkflowInstance, + step: dict[str, Any], +) -> StepResult: + """Webhook step — sends an outgoing webhook. + + Config: + url: str + method: str (default POST) + headers: dict + body: dict + secret: str (for HMAC signing) + """ + config = step.get("config", {}) + url = config.get("url", "") + method = config.get("method", "POST").upper() + headers = config.get("headers", {}) + body = config.get("body", {}) + + if not url: + return StepResult(error="webhook step requires url", abort=True) + + if not _is_url_safe(url): + return StepResult(error=f"URL blocked by SSRF protection: {url}", abort=True) + + import json + import httpx + + try: + async with httpx.AsyncClient(timeout=30, follow_redirects=False) as client: + resp = await client.request( + method, + url, + headers={"Content-Type": "application/json", **headers}, + content=json.dumps(body), + ) + return StepResult(output={ + "status_code": resp.status_code, + "response_body": resp.text[:5000], + }) + except Exception as e: + return StepResult(error=f"webhook failed: {e}", abort=True) + + +__all__ = [ + "StepResult", + "register_step_type", + "get_step_handler", + "get_available_step_types", +] diff --git a/frontend/src/api/workflows.ts b/frontend/src/api/workflows.ts index 88a57df..022a61b 100644 --- a/frontend/src/api/workflows.ts +++ b/frontend/src/api/workflows.ts @@ -11,9 +11,23 @@ import type { PaginatedResponse } from './types'; // ── Types ── +export type WorkflowStepType = + | 'action' + | 'approval' + | 'notification' + | 'condition' + | 'wait' + | 'http' + | 'mail' + | 'calendar' + | 'dms' + | 'search' + | 'agent' + | 'crm'; + export interface WorkflowStep { name: string; - type: 'action' | 'approval' | 'notification' | 'condition'; + type: WorkflowStepType; config: Record; description?: string | null; } diff --git a/frontend/src/components/workflows/StepConfigPanel.tsx b/frontend/src/components/workflows/StepConfigPanel.tsx index 1e3f4f1..0f09110 100644 --- a/frontend/src/components/workflows/StepConfigPanel.tsx +++ b/frontend/src/components/workflows/StepConfigPanel.tsx @@ -1,21 +1,51 @@ import React, { useState, useEffect } from 'react'; import { Select } from '@/components/ui/Select'; import { Input } from '@/components/ui/Input'; -import type { WorkflowStep } from '@/api/workflows'; +import { Code2, FormInput } from 'lucide-react'; +import type { WorkflowStep, WorkflowStepType } from '@/api/workflows'; -const stepTypeOptions = [ +const stepTypeOptions: { value: WorkflowStepType; label: string }[] = [ { value: 'action', label: 'Action' }, { value: 'approval', label: 'Approval' }, { value: 'notification', label: 'Notification' }, { value: 'condition', label: 'Condition' }, + { value: 'wait', label: 'Wait / Delay' }, + { value: 'http', label: 'HTTP Request' }, + { value: 'mail', label: 'Mail Send' }, + { value: 'calendar', label: 'Calendar' }, + { value: 'dms', label: 'DMS' }, + { value: 'search', label: 'Search' }, + { value: 'agent', label: 'Agent' }, + { value: 'crm', label: 'CRM Action' }, ]; -const configHints: Record = { - action: 'Config keys: action_type, target, params', - approval: 'Config keys: approver_role, timeout_hours', - notification: 'Config keys: channel, template, recipients', - condition: 'Config keys: field, operator, value', -}; +const httpMethods = ['GET', 'POST', 'PUT', 'PATCH', 'DELETE'].map((m) => ({ + value: m, + label: m, +})); + +const calendarActions = [ + { value: 'create', label: 'Create' }, + { value: 'update', label: 'Update' }, + { value: 'delete', label: 'Delete' }, +]; + +const dmsActions = [ + { value: 'search', label: 'Search' }, + { value: 'download', label: 'Download' }, + { value: 'upload', label: 'Upload' }, +]; + +const crmActions = [ + { value: 'create_contact', label: 'Create Contact' }, + { value: 'update_contact', label: 'Update Contact' }, + { value: 'create_company', label: 'Create Company' }, + { value: 'update_company', label: 'Update Company' }, + { value: 'delete_contact', label: 'Delete Contact' }, + { value: 'delete_company', label: 'Delete Company' }, +]; + +type ConfigMode = 'form' | 'json'; export interface StepConfigPanelProps { step: WorkflowStep; @@ -23,6 +53,7 @@ export interface StepConfigPanelProps { } export function StepConfigPanel({ step, onChange }: StepConfigPanelProps) { + const [mode, setMode] = useState('form'); const [configText, setConfigText] = useState(''); const [configError, setConfigError] = useState(undefined); @@ -31,6 +62,10 @@ export function StepConfigPanel({ step, onChange }: StepConfigPanelProps) { setConfigError(undefined); }, [step.config]); + const setConfig = (key: string, value: unknown) => { + onChange({ ...step, config: { ...step.config, [key]: value } }); + }; + const handleConfigChange = (value: string) => { setConfigText(value); try { @@ -42,6 +77,345 @@ export function StepConfigPanel({ step, onChange }: StepConfigPanelProps) { } }; + const strVal = (key: string): string => { + const v = step.config[key]; + return typeof v === 'string' ? v : ''; + }; + + const numVal = (key: string): string => { + const v = step.config[key]; + return typeof v === 'number' ? String(v) : ''; + }; + + const boolVal = (key: string): boolean => { + const v = step.config[key]; + return typeof v === 'boolean' ? v : false; + }; + + const jsonVal = (key: string): string => { + const v = step.config[key]; + if (v === undefined || v === null) return ''; + return JSON.stringify(v, null, 2); + }; + + const setJsonField = (key: string, value: string) => { + if (!value.trim()) { + setConfig(key, {}); + return; + } + try { + setConfig(key, JSON.parse(value)); + } catch { + // invalid JSON — JsonField shows the error, keep previous value + } + }; + + const renderTypeForm = () => { + switch (step.type) { + case 'wait': + return ( +
+ + setConfig( + 'duration_seconds', + e.target.value === '' ? undefined : Number(e.target.value) + ) + } + placeholder="z.B. 3600" + /> + setConfig('resume_at', e.target.value || undefined)} + placeholder="2026-08-18T09:00:00Z" + /> +
+ ); + case 'http': + return ( +
+
+ setConfig('url', e.target.value)} + placeholder="https://api.example.com/webhook" + /> +
+ setJsonField('headers', v)} + /> +
+ +