feat(G): G-RUN/G-CTX/G-WAIT/G-HTTP/G-MAIL/G-CAL/G-DMS/G-SEARCH/G-AGENT/G-CRM/G-EVT/G-WEB — Durable WorkflowRun, 10 step handlers, resume/wait/lock/retry, SSRF protection, frontend step editor

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):

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
 

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())

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",
+]