diff --git a/app/routes/workflows.py b/app/routes/workflows.py
index a7d4a2c..091b17a 100644
--- a/app/routes/workflows.py
+++ b/app/routes/workflows.py
@@ -4,7 +4,7 @@ from __future__ import annotations
 
 import uuid
 
-from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response, status
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.db import get_db
@@ -291,3 +291,290 @@ async def cancel_instance(
             detail={"detail": result["error"], "code": "invalid_state"},
         )
     return result
+
+
+# ─── G-RUN: Resume waiting workflow instance ──────────────────────────────────
+
+
+@router.post("/instances/{instance_id}/resume")
+async def resume_instance(
+    instance_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(require_permission("workflows:write")),
+):
+    """Resume a waiting workflow instance (e.g. after wait timer expired)."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+
+    from app.workflows.engine import WorkflowEngine
+    from app.models.workflow import WorkflowInstance
+    from sqlalchemy import select
+
+    result = await db.execute(
+        select(WorkflowInstance).where(
+            WorkflowInstance.id == uuid.UUID(instance_id),
+            WorkflowInstance.tenant_id == tenant_id,
+            WorkflowInstance.deleted_at.is_(None),
+        )
+    )
+    instance = result.scalar_one_or_none()
+    if instance is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail={"detail": "Instance not found", "code": "not_found"},
+        )
+    if instance.status != "waiting":
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail={"detail": "Instance is not waiting", "code": "invalid_state"},
+        )
+
+    engine = WorkflowEngine(db, tenant_id)
+    return await engine.resume(instance)
+
+
+# ─── G-MAN: Manual trigger — start workflow from UI button ────────────────────
+
+
+@router.post("/{workflow_id}/trigger", status_code=status.HTTP_201_CREATED)
+async def manual_trigger(
+    workflow_id: str,
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(require_permission("workflows:write")),
+):
+    """Manually trigger a workflow — starts a new instance with optional context."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+
+    try:
+        body = await request.json()
+    except Exception:
+        body = {}
+
+    result = await workflow_service.create_instance(
+        db,
+        tenant_id,
+        user_id,
+        workflow_id=workflow_id,
+        context=body,
+    )
+    if result is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail={"detail": "Workflow not found", "code": "not_found"},
+        )
+    return result
+
+
+# ─── G-WEB: Incoming webhook trigger ──────────────────────────────────────────
+
+
+@router.post("/webhook/{token}", status_code=status.HTTP_200_OK)
+async def webhook_trigger(
+    token: str,
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+):
+    """Incoming webhook trigger — starts a workflow via secure token."""
+    from app.models.webhook import Webhook
+    from sqlalchemy import select
+
+    result = await db.execute(
+        select(Webhook).where(
+            Webhook.token == token,
+            Webhook.is_active.is_(True),
+        )
+    )
+    webhook = result.scalar_one_or_none()
+    if webhook is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail={"detail": "Webhook not found", "code": "not_found"},
+        )
+
+    try:
+        payload = await request.json()
+    except Exception:
+        payload = {}
+
+    result = await workflow_service.create_instance(
+        db,
+        webhook.tenant_id,
+        None,
+        workflow_id=str(webhook.workflow_id) if hasattr(webhook, "workflow_id") else str(webhook.entity_id),
+        context={"webhook_payload": payload, "webhook_token": token},
+    )
+    if result is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail={"detail": "Workflow not found", "code": "not_found"},
+        )
+    return {"status": "triggered", "instance": result}
+
+
+# ─── G-LOG: Step history for a workflow instance ─────────────────────────────
+
+
+@router.get("/instances/{instance_id}/history")
+async def get_instance_history(
+    instance_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(require_permission("workflows:read")),
+):
+    """Get step history for a workflow instance (execution log)."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+
+    from app.models.workflow import WorkflowStepHistory
+    from sqlalchemy import select
+
+    result = await db.execute(
+        select(WorkflowStepHistory)
+        .where(
+            WorkflowStepHistory.tenant_id == tenant_id,
+            WorkflowStepHistory.instance_id == uuid.UUID(instance_id),
+        )
+        .order_by(WorkflowStepHistory.created_at.asc())
+    )
+    history = result.scalars().all()
+    return {
+        "items": [
+            {
+                "id": str(h.id),
+                "step_index": h.step_index,
+                "step_type": h.step_type,
+                "action": h.action,
+                "actor_id": str(h.actor_id) if h.actor_id else None,
+                "details": h.details,
+                "created_at": h.created_at.isoformat() if h.created_at else None,
+            }
+            for h in history
+        ],
+        "total": len(history),
+    }
+
+
+# ─── G-APPROVAL: Approval step using central ApprovalRequest ─────────────────
+
+
+@router.post("/instances/{instance_id}/approve")
+async def approve_workflow_step(
+    instance_id: str,
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(require_permission("workflows:write")),
+):
+    """Approve the current approval step of a workflow instance."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+
+    try:
+        body = await request.json()
+    except Exception:
+        body = {}
+    comment = body.get("comment", "")
+
+    from app.core.approval import create_approval_request, decide_approval
+    from app.models.workflow import WorkflowInstance
+    from sqlalchemy import select
+
+    result = await db.execute(
+        select(WorkflowInstance).where(
+            WorkflowInstance.id == uuid.UUID(instance_id),
+            WorkflowInstance.tenant_id == tenant_id,
+            WorkflowInstance.deleted_at.is_(None),
+        )
+    )
+    instance = result.scalar_one_or_none()
+    if instance is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail={"detail": "Instance not found", "code": "not_found"},
+        )
+
+    approval = await create_approval_request(
+        db=db,
+        tenant_id=tenant_id,
+        entity_type="workflow_instance",
+        entity_id=uuid.UUID(instance_id),
+        action="workflow_step_approval",
+        requested_by=user_id,
+        requested_by_type="user",
+    )
+    await decide_approval(
+        db=db,
+        approval_id=approval["id"],
+        decision="approved",
+        decided_by=user_id,
+        comment=comment,
+    )
+
+    return await workflow_service.advance_instance(
+        db,
+        tenant_id,
+        user_id,
+        instance_id=instance_id,
+        decision="approved",
+        comment=comment,
+        is_system_admin=current_user.get("is_system_admin", False),
+    )
+
+
+@router.post("/instances/{instance_id}/reject")
+async def reject_workflow_step(
+    instance_id: str,
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(require_permission("workflows:write")),
+):
+    """Reject the current approval step of a workflow instance."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+
+    try:
+        body = await request.json()
+    except Exception:
+        body = {}
+    comment = body.get("comment", "")
+
+    from app.core.approval import create_approval_request, decide_approval
+    from app.models.workflow import WorkflowInstance
+    from sqlalchemy import select
+
+    result = await db.execute(
+        select(WorkflowInstance).where(
+            WorkflowInstance.id == uuid.UUID(instance_id),
+            WorkflowInstance.tenant_id == tenant_id,
+            WorkflowInstance.deleted_at.is_(None),
+        )
+    )
+    instance = result.scalar_one_or_none()
+    if instance is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail={"detail": "Instance not found", "code": "not_found"},
+        )
+
+    approval = await create_approval_request(
+        db=db,
+        tenant_id=tenant_id,
+        entity_type="workflow_instance",
+        entity_id=uuid.UUID(instance_id),
+        action="workflow_step_approval",
+        requested_by=user_id,
+        requested_by_type="user",
+    )
+    await decide_approval(
+        db=db,
+        approval_id=approval["id"],
+        decision="rejected",
+        decided_by=user_id,
+        comment=comment,
+    )
+
+    return await workflow_service.cancel_instance(
+        db,
+        tenant_id,
+        user_id,
+        instance_id=instance_id,
+        is_system_admin=current_user.get("is_system_admin", False),
+    )
diff --git a/tests/test_phase_g_workflows.py b/tests/test_phase_g_workflows.py
new file mode 100644
index 0000000..0350dab
--- /dev/null
+++ b/tests/test_phase_g_workflows.py
@@ -0,0 +1,372 @@
+"""Tests for Phase G — Workflow MVP: step handlers, wait/resume, retry, SSRF, triggers.
+
+All tests use mocks — no real DB/LLM/Redis/HTTP needed.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import uuid
+from datetime import UTC, datetime, timedelta
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from app.workflows.step_handlers import (
+    StepResult,
+    get_step_handler,
+    get_available_step_types,
+    _is_url_safe,
+)
+
+
+# ─── Step Handler Registry ────────────────────────────────────────────────────
+
+
+class TestStepHandlerRegistry:
+    """Test the step handler registry."""
+
+    def test_all_step_types_registered(self):
+        """All 10 built-in step types are registered."""
+        types = get_available_step_types()
+        expected = {"agent", "calendar", "crm", "dms", "event", "http", "mail", "search", "wait", "webhook"}
+        assert set(types) == expected
+
+    def test_get_step_handler_returns_callable(self):
+        """get_step_handler returns a callable for each registered type."""
+        for step_type in get_available_step_types():
+            handler = get_step_handler(step_type)
+            assert handler is not None
+            assert callable(handler)
+
+    def test_get_step_handler_unknown_returns_none(self):
+        """get_step_handler returns None for unknown step type."""
+        assert get_step_handler("nonexistent") is None
+
+
+# ─── Wait Step (G-WAIT) ───────────────────────────────────────────────────────
+
+
+class TestWaitStep:
+    """Test the wait/delay step handler."""
+
+    @pytest.mark.asyncio
+    async def test_wait_with_duration(self):
+        """Wait step with duration_seconds sets resume_at correctly."""
+        instance = MagicMock()
+        instance.context = {}
+        step = {"type": "wait", "config": {"duration_seconds": 60}}
+        result = await get_step_handler("wait")(
+            MagicMock(), uuid.uuid4(), instance, step
+        )
+        assert result.advance is False
+        assert result.wait_until is not None
+        assert result.wait_reason == "wait"
+        # resume_at should be ~60s in the future
+        now = datetime.now(UTC)
+        delta = result.wait_until - now
+        assert 50 < delta.total_seconds() < 70
+
+    @pytest.mark.asyncio
+    async def test_wait_with_absolute_time(self):
+        """Wait step with resume_at sets exact resume time."""
+        instance = MagicMock()
+        instance.context = {}
+        future = (datetime.now(UTC) + timedelta(hours=2)).isoformat()
+        step = {"type": "wait", "config": {"resume_at": future}}
+        result = await get_step_handler("wait")(
+            MagicMock(), uuid.uuid4(), instance, step
+        )
+        assert result.advance is False
+        assert result.wait_until is not None
+        assert result.wait_reason == "wait"
+
+    @pytest.mark.asyncio
+    async def test_wait_without_config_aborts(self):
+        """Wait step without duration_seconds or resume_at aborts."""
+        instance = MagicMock()
+        instance.context = {}
+        step = {"type": "wait", "config": {}}
+        result = await get_step_handler("wait")(
+            MagicMock(), uuid.uuid4(), instance, step
+        )
+        assert result.abort is True
+        assert "requires" in result.error
+
+
+# ─── SSRF Protection (G-HTTP) ────────────────────────────────────────────────
+
+
+class TestSSRFProtection:
+    """Test the SSRF protection for HTTP and webhook steps."""
+
+    def test_blocks_localhost(self):
+        """SSRF blocks localhost."""
+        assert _is_url_safe("http://localhost:8080/api") is False
+        assert _is_url_safe("http://127.0.0.1:8080/api") is False
+
+    def test_blocks_private_ips(self):
+        """SSRF blocks private IP ranges."""
+        assert _is_url_safe("http://192.168.1.1/api") is False
+        assert _is_url_safe("http://10.0.0.1/api") is False
+        assert _is_url_safe("http://172.16.0.1/api") is False
+
+    def test_blocks_non_http_schemes(self):
+        """SSRF blocks non-http/https schemes."""
+        assert _is_url_safe("ftp://example.com/file") is False
+        assert _is_url_safe("file:///etc/passwd") is False
+        assert _is_url_safe("gopher://example.com") is False
+
+    def test_allows_public_urls(self):
+        """SSRF allows public HTTP/HTTPS URLs."""
+        assert _is_url_safe("https://api.example.com/webhook") is True
+        assert _is_url_safe("http://example.com/api") is True
+
+    def test_blocks_metadata_endpoint(self):
+        """SSRF blocks cloud metadata endpoints."""
+        assert _is_url_safe("http://metadata.google.internal/computeMetadata/") is False
+
+    def test_blocks_ipv6_loopback(self):
+        """SSRF blocks IPv6 loopback."""
+        assert _is_url_safe("http://[::1]:8080/api") is False
+
+    def test_handles_invalid_url(self):
+        """SSRF handles invalid URLs gracefully."""
+        assert _is_url_safe("") is False
+        assert _is_url_safe("not-a-url") is False
+
+
+# ─── HTTP Step (G-HTTP) ──────────────────────────────────────────────────────
+
+
+class TestHttpStep:
+    """Test the HTTP request step handler."""
+
+    @pytest.mark.asyncio
+    async def test_http_without_url_aborts(self):
+        """HTTP step without URL aborts."""
+        instance = MagicMock()
+        instance.context = {}
+        step = {"type": "http", "config": {"method": "GET"}}
+        result = await get_step_handler("http")(
+            MagicMock(), uuid.uuid4(), instance, step
+        )
+        assert result.abort is True
+        assert "url" in result.error.lower()
+
+    @pytest.mark.asyncio
+    async def test_http_with_ssrf_url_aborts(self):
+        """HTTP step with SSRF-blocked URL aborts."""
+        instance = MagicMock()
+        instance.context = {}
+        step = {"type": "http", "config": {"url": "http://127.0.0.1:8080/secret"}}
+        result = await get_step_handler("http")(
+            MagicMock(), uuid.uuid4(), instance, step
+        )
+        assert result.abort is True
+        assert "ssrf" in result.error.lower()
+
+
+# ─── Event Step (G-EVT) ──────────────────────────────────────────────────────
+
+
+class TestEventStep:
+    """Test the event publishing step handler."""
+
+    @pytest.mark.asyncio
+    async def test_event_without_name_aborts(self):
+        """Event step without event_name aborts."""
+        instance = MagicMock()
+        instance.id = uuid.uuid4()
+        instance.context = {}
+        step = {"type": "event", "config": {"payload": {"key": "value"}}}
+        result = await get_step_handler("event")(
+            MagicMock(), uuid.uuid4(), instance, step
+        )
+        assert result.abort is True
+        assert "event_name" in result.error
+
+    @pytest.mark.asyncio
+    async def test_event_publishes_successfully(self):
+        """Event step publishes event to event bus."""
+        instance = MagicMock()
+        instance.id = uuid.uuid4()
+        instance.context = {}
+        step = {"type": "event", "config": {"event_name": "test.event", "payload": {"key": "value"}}}
+
+        with patch("app.core.event_bus.get_event_bus") as mock_get_bus:
+            mock_bus = MagicMock()
+            mock_bus.publish = AsyncMock()
+            mock_get_bus.return_value = mock_bus
+
+            result = await get_step_handler("event")(
+                MagicMock(), uuid.uuid4(), instance, step
+            )
+            assert result.advance is True
+            assert result.output["event_published"] == "test.event"
+            mock_bus.publish.assert_called_once()
+
+
+# ─── CRM Step (G-CRM) ────────────────────────────────────────────────────────
+
+
+class TestCrmStep:
+    """Test the CRM action step handler."""
+
+    @pytest.mark.asyncio
+    async def test_crm_without_action_aborts(self):
+        """CRM step without action aborts."""
+        instance = MagicMock()
+        instance.context = {}
+        step = {"type": "crm", "config": {"data": {"name": "Test"}}}
+        result = await get_step_handler("crm")(
+            MagicMock(), uuid.uuid4(), instance, step
+        )
+        assert result.abort is True
+        assert "action" in result.error
+
+    @pytest.mark.asyncio
+    async def test_crm_unknown_action_aborts(self):
+        """CRM step with unknown action aborts."""
+        instance = MagicMock()
+        instance.context = {}
+        step = {"type": "crm", "config": {"action": "invalid_action"}}
+        result = await get_step_handler("crm")(
+            MagicMock(), uuid.uuid4(), instance, step
+        )
+        assert result.abort is True
+        assert "unknown" in result.error.lower()
+
+
+# ─── StepResult ──────────────────────────────────────────────────────────────
+
+
+class TestStepResult:
+    """Test the StepResult class."""
+
+    def test_default_step_result_advances(self):
+        """Default StepResult advances to next step."""
+        result = StepResult()
+        assert result.advance is True
+        assert result.next_index is None
+        assert result.wait_until is None
+        assert result.error is None
+        assert result.abort is False
+        assert result.output == {}
+
+    def test_step_result_with_wait(self):
+        """StepResult with wait_until does not advance."""
+        wait_time = datetime.now(UTC) + timedelta(seconds=30)
+        result = StepResult(advance=False, wait_until=wait_time, wait_reason="wait")
+        assert result.advance is False
+        assert result.wait_until == wait_time
+        assert result.wait_reason == "wait"
+
+    def test_step_result_with_error(self):
+        """StepResult with error but no abort is retryable."""
+        result = StepResult(error="Something failed")
+        assert result.error == "Something failed"
+        assert result.abort is False
+
+    def test_step_result_with_abort(self):
+        """StepResult with abort stops the workflow."""
+        result = StepResult(error="Fatal error", abort=True)
+        assert result.abort is True
+        assert result.error == "Fatal error"
+
+    def test_step_result_with_branch(self):
+        """StepResult with next_index branches to specific step."""
+        result = StepResult(next_index=5)
+        assert result.next_index == 5
+        assert result.advance is True  # Still advances, just to specific index
+
+
+# ─── Workflow Engine Resume (G-RUN) ──────────────────────────────────────────
+
+
+class TestWorkflowEngineResume:
+    """Test the WorkflowEngine resume functionality."""
+
+    @pytest.mark.asyncio
+    async def test_resume_non_waiting_instance_returns_unchanged(self):
+        """Resume on a non-waiting instance returns the instance unchanged."""
+        from app.workflows.engine import WorkflowEngine
+
+        instance = MagicMock()
+        instance.status = "completed"
+        instance.id = uuid.uuid4()
+        instance.workflow_id = uuid.uuid4()
+        instance.current_step_index = 0
+
+        db = MagicMock()
+        engine = WorkflowEngine(db, uuid.uuid4())
+        result = await engine.resume(instance)
+        # Should return _instance_to_dict result, not process
+        assert result is not None
+
+    @pytest.mark.asyncio
+    async def test_find_resumable_workflows_query(self):
+        """find_resumable_workflows queries for waiting instances with passed resume_at."""
+        from app.workflows.engine import find_resumable_workflows
+        from app.models.workflow import WorkflowInstance
+
+        db = MagicMock()
+        mock_result = MagicMock()
+        mock_result.scalars.return_value.all.return_value = []
+        db.execute = AsyncMock(return_value=mock_result)
+
+        await find_resumable_workflows(db, uuid.uuid4())
+        db.execute.assert_called_once()
+
+
+# ─── Workflow Schema (G-COND) ───────────────────────────────────────────────
+
+
+class TestWorkflowSchema:
+    """Test the extended WorkflowStep schema."""
+
+    def test_step_schema_accepts_new_types(self):
+        """WorkflowStep schema accepts all new step types."""
+        from app.schemas.workflow import WorkflowStep
+
+        for step_type in ["wait", "http", "mail", "calendar", "dms", "search", "agent", "crm", "event", "webhook"]:
+            step = WorkflowStep(name=f"Test {step_type}", type=step_type, config={})
+            assert step.type == step_type
+
+    def test_step_schema_rejects_unknown_type(self):
+        """WorkflowStep schema rejects unknown step types."""
+        from app.schemas.workflow import WorkflowStep
+        from pydantic import ValidationError
+
+        with pytest.raises(ValidationError):
+            WorkflowStep(name="Bad", type="unknown_type", config={})
+
+    def test_step_schema_still_accepts_legacy_types(self):
+        """WorkflowStep schema still accepts legacy step types."""
+        from app.schemas.workflow import WorkflowStep
+
+        for step_type in ["action", "approval", "notification", "condition"]:
+            step = WorkflowStep(name=f"Legacy {step_type}", type=step_type, config={})
+            assert step.type == step_type
+
+
+# ─── Workflow Model (G-RUN) ─────────────────────────────────────────────────
+
+
+class TestWorkflowModelDurableFields:
+    """Test the new durable/resumable fields on WorkflowInstance."""
+
+    def test_workflow_instance_has_resume_fields(self):
+        """WorkflowInstance model has all G-RUN fields."""
+        from app.models.workflow import WorkflowInstance
+
+        # Check that the model has the new columns
+        assert hasattr(WorkflowInstance, "resume_at")
+        assert hasattr(WorkflowInstance, "resume_reason")
+        assert hasattr(WorkflowInstance, "step_state")
+        assert hasattr(WorkflowInstance, "idempotency_key")
+        assert hasattr(WorkflowInstance, "lock_owner")
+        assert hasattr(WorkflowInstance, "lock_expires_at")
+        assert hasattr(WorkflowInstance, "error_message")
+        assert hasattr(WorkflowInstance, "retry_count")
+        assert hasattr(WorkflowInstance, "max_retries")