diff --git a/mistral/engine/workflows.py b/mistral/engine/workflows.py
index 2681cd5ab..f988434fe 100644
--- a/mistral/engine/workflows.py
+++ b/mistral/engine/workflows.py
@@ -430,10 +430,21 @@ class Workflow(object):
 
         # When we set an ERROR state we should safely set output value getting
         # w/o exceptions due to field size limitations.
-        msg = utils.cut_by_kb(
-            msg,
-            cfg.CONF.engine.execution_field_size_limit_kb
-        )
+
+        length_output_on_error = len(str(output_on_error).encode("utf-8"))
+        total_output_length = utils.get_number_of_chars_from_kilobytes(
+            cfg.CONF.engine.execution_field_size_limit_kb)
+
+        if length_output_on_error < total_output_length:
+            msg = utils.cut_by_char(
+                msg,
+                total_output_length - length_output_on_error
+            )
+        else:
+            msg = utils.cut_by_kb(
+                msg,
+                cfg.CONF.engine.execution_field_size_limit_kb
+            )
 
         self.wf_ex.output = merge_dicts({'result': msg}, output_on_error)
 
diff --git a/mistral/tests/unit/engine/test_dataflow.py b/mistral/tests/unit/engine/test_dataflow.py
index f419011bf..04204f9ab 100644
--- a/mistral/tests/unit/engine/test_dataflow.py
+++ b/mistral/tests/unit/engine/test_dataflow.py
@@ -26,6 +26,8 @@ from mistral.tests.unit.engine import base as engine_test_base
 from mistral.workflow import data_flow
 from mistral.workflow import states
 
+import sys
+
 
 # Use the set_default method to set value otherwise in certain test cases
 # the change in value is not permanent.
@@ -666,6 +668,55 @@ class DataFlowEngineTest(engine_test_base.EngineTestCase):
 
         self.assertIn('task(task1).result.message', task1.state_info)
 
+    def test_size_of_output_by_execution_field_size_limit_kb(self):
+        wf_text = """
+        version: '2.0'
+
+        wf:
+          type: direct
+
+          output-on-error:
+            custom_error: The action in the task does not exists
+
+          tasks:
+            task1:
+              action: wrong.task
+        """
+        # Note: The number 1121 below added as value for field size
+        # limit is because the output of workflow error comes as
+        # workflow error string + custom error message and total length
+        # might be greater than 1121. It varies depending on the length
+        # of the custom message. This is a random number value used for
+        # test case only.
+        cfg.CONF.set_default(
+            'execution_field_size_limit_kb',
+            1121,
+            group='engine'
+        )
+
+        kilobytes = cfg.CONF.engine.execution_field_size_limit_kb
+
+        bytes_per_char = sys.getsizeof('s') - sys.getsizeof('')
+
+        total_output_length = int(kilobytes * 1024 / bytes_per_char)
+
+        wf_service.create_workflows(wf_text)
+
+        wf_ex = self.engine.start_workflow('wf', '', None)
+
+        self.await_workflow_error(wf_ex.id)
+
+        with db_api.transaction():
+            # Note: We need to reread execution to access related tasks.
+            wf_ex = db_api.get_workflow_execution(wf_ex.id)
+
+            wf_output = wf_ex.output
+
+        self.assertLess(
+            len(str(wf_output.get('custom_error'))),
+            total_output_length
+        )
+
     def test_override_json_input(self):
         wf_text = """---
         version: 2.0
diff --git a/mistral/utils/__init__.py b/mistral/utils/__init__.py
index 02d09fbfb..8a3f33373 100644
--- a/mistral/utils/__init__.py
+++ b/mistral/utils/__init__.py
@@ -311,9 +311,11 @@ def cut_by_kb(data, kilobytes):
     if kilobytes <= 0:
         return cut(data)
 
-    bytes_per_char = sys.getsizeof('s') - sys.getsizeof('')
-    length = int(kilobytes * 1024 / bytes_per_char)
+    length = get_number_of_chars_from_kilobytes(kilobytes)
+    return cut(data, length)
 
+
+def cut_by_char(data, length):
     return cut(data, length)
 
 
@@ -365,6 +367,12 @@ class NotDefined(object):
     pass
 
 
+def get_number_of_chars_from_kilobytes(kilobytes):
+    bytes_per_char = sys.getsizeof('s') - sys.getsizeof('')
+    total_number_of_chars = int(kilobytes * 1024 / bytes_per_char)
+    return total_number_of_chars
+
+
 def get_dict_from_string(string, delimiter=','):
     if not string:
         return {}