Repository: RasaHQ/rasa
Pull request: Fix e2e training edge cases
**Proposed changes**:
- Fixes are variety of training scenarios outlined in #7129 
- fixes https://github.com/RasaHQ/rasa/issues/7360

**Status (please check what you already did)**:
- [x] added some tests for the functionality
- [ ] updated the documentation
- [ ] updated the changelog (please check [changelog](https://github.com/RasaHQ/rasa/tree/master/changelog) for instructions)
- [x] reformat files using `black` (please check [Readme](https://github.com/RasaHQ/rasa#code-style) for instructions)


Problem statement:
make sure that rasa train commands run gracefully with e2e
With e2e we linked training `rasa.nlu` and `rasa.core`. Therefore, we need to make sure that `rasa train`, `rasa train nlu` and `rasa train core` run perfectly

Potential pitfalls (just smth came to my mind, there might. be more edge cases):
- [ ] `rasa train` was called, nlu data didn't change but e2e stories did and added e2e examples -> need to retrain nlu model
- [ ] `rasa train core` was called, but core stories contain e2e examples -> the command should raise an error that `rasa train` with appropriate nlu pipeline should be called
- [ ] when nlu config changes, `rasa train` should retrain core as well
- [ ] do not allow both `intent:` and `user:` in training data
- [ ] during loading for prediction `core` should be able to check that loaded `nlu` model is the same as during training
- [ ] `rasa train` was called, but no nlu data are present. However, your stories contain e2e examples. -> We need to train an nlu model.

Hints from issue discussion:
@Ghostvv Somebody mentioned that training on `e2e` is currently broken - does that block implementing a solution or can we move forward with it?
the training on e2e should work now
Is the `e2ebot` example bot meant to be in a working state?
I get this error when using it:
```

2020-11-19 14:44:03 DEBUG    rasa.core.processor  - Action 'action_session_start' ended with events '[<rasa.shared.core.events.SessionStarted object at 0x153e4b450>, <rasa.shared.core.events.ActionExecuted object at 0x153d87250>]'.
2020-11-19 14:44:04 DEBUG    rasa.core.processor  - Received user message 'hello' with intent '{'id': -4740810205347421666, 'name': 'greet', 'confidence': 0.9999948740005493}' and entities '[]'
2020-11-19 14:44:04 DEBUG    rasa.core.processor  - Logged UserUtterance - tracker now has 4 events.
2020-11-19 14:44:04 DEBUG    rasa.core.policies.ensemble  - Made e2e prediction using user text.
2020-11-19 14:44:04 DEBUG    rasa.core.policies.ensemble  - Added `DefinePrevUserUtteredFeaturization(True)` event.
2020-11-19 14:44:04 DEBUG    rasa.core.policies.ensemble  - Predicted next action using policy_0_TEDPolicy.
2020-11-19 14:44:04 DEBUG    rasa.core.processor  - Predicted next action 'Welcome to moodbot. How are you feeling today?' with confidence 1.00.
2020-11-19 14:44:04 ERROR    rasa.core.processor  - Encountered an exception while running action 'Welcome to moodbot. How are you feeling today?'.Bot will continue, but the actions events are lost. Please check the logs of your action server for more information.
Traceback (most recent call last):
  File "/Users/josephjuzl/code/rasa/rasa/core/processor.py", line 750, in _run_action
    output_channel, nlg, temporary_tracker, self.domain
  File "/Users/josephjuzl/code/rasa/rasa/core/actions/action.py", line 599, in run
    f"Failed to execute custom action '{self.name()}' "
rasa.shared.exceptions.RasaException: Failed to execute custom action 'Welcome to moodbot. How are you feeling today?' because no endpoint is configured to run this custom action. Please take a look at the docs and set an endpoint configuration via the --endpoints flag. https://rasa.com/docs/rasa/custom-actions
2020-11-19 14:44:04 DEBUG    rasa.core.processor  - Policy prediction ended with events '[<rasa.shared.core.events.DefinePrevUserUtteredFeaturization object at 0x153f0c510>]'.
2020-11-19 14:44:04 DEBUG    rasa.core.processor  - Action 'Welcome to moodbot. How are you feeling today?' ended with events '[]'.
2020-11-19 14:44:04 DEBUG    rasa.core.policies.ensemble  - Predicted next action using policy_0_TEDPolicy.
2020-11-19 14:44:04 DEBUG    rasa.core.processor  - Predicted next action 'action_listen' with confidence 0.82.
2020-11-19 14:44:04 DEBUG    rasa.core.processor  - Policy prediction ended with events '[]'.
2020-11-19 14:44:04 DEBUG    rasa.core.processor  - Action 'action_listen' ended with events '[]'.
2020-11-19 14:44:04 DEBUG    rasa.core.lock_store  - Deleted lock for conversation 'test_user4'.
```

Do bot responses in e2e stories have to still be defined in the domain?
predicting e2e bot utterances in `rasa shell` doesn't work yet. There is a separate issue for it: https://github.com/RasaHQ/rasa/issues/6410
This was from the rest api - but I guess it has the same problem. 
Ok so the scope of this issue is purely to check for bugs in the training process?
the scope of this issue to check that `rasa train core` fails gracefully and that `rasa train` correctly retrains various models depending on the changed data (nlu data now influence core model and core data influence nlu model)
also, during loading for prediction core should be able to check that loaded nlu model is the same as during training
## Findings so far

1. > `rasa train` was called, nlu data didn't change but e2e stories did and added e2e examples -> need to retrain nlu model

Seems to work.

2. > `rasa train core` was called, but core stories contain e2e examples -> the command should raise an error that rasa train with appropriate nlu pipeline should be called

Throws this error:
```
> rasa train core                                                                                   (/Users/josephjuzl/.kube/config doesn't exist)
Training Core model...
2020-11-20 12:46:44 INFO     rasa.core.policies.ensemble  - MappingPolicy not included in policy ensemble. Default intents 'restart and back will not trigger actions 'action_restart' and 'action_back'.
Processed story blocks: 100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████| 3/3 [00:00<00:00, 2392.64it/s, # trackers=1]
Processed story blocks: 100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████| 3/3 [00:00<00:00, 1111.96it/s, # trackers=3]
Processed story blocks: 100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████| 3/3 [00:00<00:00, 307.63it/s, # trackers=12]
Processed story blocks: 100%|██████████████████████████████████████████████████████████████████████████████████████████████████████████| 3/3 [00:00<00:00, 74.00it/s, # trackers=39]
Processed trackers: 100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████| 120/120 [00:00<00:00, 255.81it/s, # actions=601]
Traceback (most recent call last):
  File "/Users/josephjuzl/.pyenv/versions/rasa-py3.7/bin/rasa", line 5, in <module>
    main()
  File "/Users/josephjuzl/code/rasa/rasa/__main__.py", line 116, in main
    cmdline_arguments.func(cmdline_arguments)
  File "/Users/josephjuzl/code/rasa/rasa/cli/train.py", line 124, in train_core
    additional_arguments=additional_arguments,
  File "/Users/josephjuzl/code/rasa/rasa/train.py", line 317, in train_core
    additional_arguments=additional_arguments,
  File "/Users/josephjuzl/code/rasa/rasa/utils/common.py", line 308, in run_in_loop
    result = loop.run_until_complete(f)
  File "uvloop/loop.pyx", line 1456, in uvloop.loop.Loop.run_until_complete
  File "/Users/josephjuzl/code/rasa/rasa/train.py", line 373, in train_core_async
    additional_arguments=additional_arguments,
  File "/Users/josephjuzl/code/rasa/rasa/train.py", line 409, in _train_core_with_validated_data
    interpreter=interpreter,
  File "/Users/josephjuzl/code/rasa/rasa/core/train.py", line 67, in train
    agent.train(training_data, **additional_arguments)
  File "/Users/josephjuzl/code/rasa/rasa/core/agent.py", line 721, in train
    training_trackers, self.domain, interpreter=self.interpreter, **kwargs
  File "/Users/josephjuzl/code/rasa/rasa/core/policies/ensemble.py", line 194, in train
    trackers_to_train, domain, interpreter=interpreter, **kwargs
  File "/Users/josephjuzl/code/rasa/rasa/core/policies/ted_policy.py", line 461, in train
    training_trackers, domain, interpreter, **kwargs
  File "/Users/josephjuzl/code/rasa/rasa/core/policies/policy.py", line 172, in featurize_for_training
    training_trackers, domain, interpreter
  File "/Users/josephjuzl/code/rasa/rasa/core/featurizers/tracker_featurizers.py", line 192, in featurize_trackers
    entity_tags = self._create_entity_tags(trackers_as_entities, interpreter)
  File "/Users/josephjuzl/code/rasa/rasa/core/featurizers/tracker_featurizers.py", line 104, in _create_entity_tags
    for trackers_entities in trackers_as_entities
  File "/Users/josephjuzl/code/rasa/rasa/core/featurizers/tracker_featurizers.py", line 104, in <listcomp>
    for trackers_entities in trackers_as_entities
  File "/Users/josephjuzl/code/rasa/rasa/core/featurizers/tracker_featurizers.py", line 102, in <listcomp>
    for entity_data in trackers_entities
  File "/Users/josephjuzl/code/rasa/rasa/core/featurizers/single_state_featurizer.py", line 274, in encode_entities
    for token in parsed_text.get(TOKENS_NAMES[TEXT]):
AttributeError: 'NoneType' object has no attribute 'get'
```

3. > when nlu config changes, `rasa train` should retrain core as well

Does not retrain core:
```
Core stories/configuration did not change. No need to retrain Core model.
```


4. > do not allow both intent: and user: in training data

Currently does allow it

5. > `rasa train` was called, but no nlu data are present. However, your stories contain e2e examples. -> We need to train an nlu model.

Throws this error:
```
Training Core model...
2020-11-20 14:34:15 INFO     rasa.core.policies.ensemble  - MappingPolicy not included in policy ensemble. Default intents 'restart and back will not trigger actions 'action_restart' and 'action_back'.
Processed story blocks: 100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████| 1/1 [00:00<00:00, 1837.99it/s, # trackers=1]
Processed story blocks: 100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████| 1/1 [00:00<00:00, 2242.94it/s, # trackers=1]
Processed story blocks: 100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████| 1/1 [00:00<00:00, 1276.80it/s, # trackers=2]
Processed story blocks: 100%|██████████████████████████████████████████████████████████████████████████████████████████████████████████| 1/1 [00:00<00:00, 854.41it/s, # trackers=3]
Processed trackers: 100%|██████████████████████████████████████████████████████████████████████████████████████████████████████████████| 4/4 [00:00<00:00, 393.59it/s, # actions=25]
Traceback (most recent call last):
  File "/Users/josephjuzl/.pyenv/versions/rasa-py3.7/bin/rasa", line 5, in <module>
    main()
  File "/Users/josephjuzl/code/rasa/rasa/__main__.py", line 116, in main
    cmdline_arguments.func(cmdline_arguments)
  File "/Users/josephjuzl/code/rasa/rasa/cli/train.py", line 90, in train
    nlu_additional_arguments=extract_nlu_additional_arguments(args),
  File "/Users/josephjuzl/code/rasa/rasa/train.py", line 55, in train
    loop,
  File "/Users/josephjuzl/code/rasa/rasa/utils/common.py", line 308, in run_in_loop
    result = loop.run_until_complete(f)
  File "uvloop/loop.pyx", line 1456, in uvloop.loop.Loop.run_until_complete
  File "/Users/josephjuzl/code/rasa/rasa/train.py", line 110, in train_async
    nlu_additional_arguments=nlu_additional_arguments,
  File "/Users/josephjuzl/code/rasa/rasa/train.py", line 183, in _train_async_internal
    additional_arguments=core_additional_arguments,
  File "/Users/josephjuzl/code/rasa/rasa/train.py", line 409, in _train_core_with_validated_data
    interpreter=interpreter,
  File "/Users/josephjuzl/code/rasa/rasa/core/train.py", line 67, in train
    agent.train(training_data, **additional_arguments)
  File "/Users/josephjuzl/code/rasa/rasa/core/agent.py", line 721, in train
    training_trackers, self.domain, interpreter=self.interpreter, **kwargs
  File "/Users/josephjuzl/code/rasa/rasa/core/policies/ensemble.py", line 194, in train
    trackers_to_train, domain, interpreter=interpreter, **kwargs
  File "/Users/josephjuzl/code/rasa/rasa/core/policies/ted_policy.py", line 461, in train
    training_trackers, domain, interpreter, **kwargs
  File "/Users/josephjuzl/code/rasa/rasa/core/policies/policy.py", line 172, in featurize_for_training
    training_trackers, domain, interpreter
  File "/Users/josephjuzl/code/rasa/rasa/core/featurizers/tracker_featurizers.py", line 192, in featurize_trackers
    entity_tags = self._create_entity_tags(trackers_as_entities, interpreter)
  File "/Users/josephjuzl/code/rasa/rasa/core/featurizers/tracker_featurizers.py", line 104, in _create_entity_tags
    for trackers_entities in trackers_as_entities
  File "/Users/josephjuzl/code/rasa/rasa/core/featurizers/tracker_featurizers.py", line 104, in <listcomp>
    for trackers_entities in trackers_as_entities
  File "/Users/josephjuzl/code/rasa/rasa/core/featurizers/tracker_featurizers.py", line 102, in <listcomp>
    for entity_data in trackers_entities
  File "/Users/josephjuzl/code/rasa/rasa/core/featurizers/single_state_featurizer.py", line 274, in encode_entities
    for token in parsed_text.get(TOKENS_NAMES[TEXT]):
AttributeError: 'NoneType' object has no attribute 'get'

```

6. > during loading for prediction core should be able to check that loaded nlu model is the same as during training

I'm not sure what this means exactly. Is this a scenario when the nlu model is manually replaced? Or via a training command?

in point 2. it should fail gracefully, by giving a user helpful error message
in point 6.
> Is this a scenario when the nlu model is manually replaced?

yes

> Or via a training command?

if we fix training commands, theoretically, it shouldn't happen

Commit message: fix e2e training edge cases

Patch to review:
diff --git a/data/test_yaml_stories/stories_e2e.yml b/data/test_yaml_stories/stories_e2e.yml
new file mode 100644
index 000000000000..b20be88ed78d
--- /dev/null
+++ b/data/test_yaml_stories/stories_e2e.yml
@@ -0,0 +1,30 @@
+version: "2.0"
+
+stories:
+- story: happy path (intent to action)
+  steps:
+  - intent: greet
+  - action: utter_greet
+  - intent: mood_great
+  - action: utter_happy
+
+- story: sad path (text to text)
+  steps:
+  - user: "[Hello](bla)"
+  - bot: "Welcome to moodbot. How are you feeling today?"
+  - user: "Horrible"
+  - bot: "Oh no! Here is a kitten photo. Did it help?"
+  - user: "Yes"
+  - bot: "Perfect"
+
+- story: sad path 2 (mixed)
+  steps:
+  - intent: greet
+  - action: utter_greet
+  - user: "I'm not happy"
+  - action: utter_cheer_up
+  - action: utter_did_that_help
+  - user: "Not at all"
+  - bot: "Goodbye!"
+
+
diff --git a/data/test_yaml_stories/stories_simple.yml b/data/test_yaml_stories/stories_simple.yml
new file mode 100644
index 000000000000..4b5fe55fd88c
--- /dev/null
+++ b/data/test_yaml_stories/stories_simple.yml
@@ -0,0 +1,8 @@
+- story: simple_story
+  steps:
+  - intent: greet
+  - action: utter_greet
+  - intent: default
+  - action: utter_default
+  - intent: goodbye
+  - action: utter_goodbye
diff --git a/rasa/model.py b/rasa/model.py
index 5f09ad29cb2f..d3f710e382d2 100644
--- a/rasa/model.py
+++ b/rasa/model.py
@@ -416,7 +416,10 @@ def move_model(source: Text, target: Text) -> bool:
 
 
 def should_retrain(
-    new_fingerprint: Fingerprint, old_model: Text, train_path: Text
+    new_fingerprint: Fingerprint,
+    old_model: Text,
+    train_path: Text,
+    has_e2e_examples: bool,
 ) -> FingerprintComparisonResult:
     """Check which components of a model should be retrained.
 
@@ -451,6 +454,9 @@ def should_retrain(
             ),
         )
 
+        if has_e2e_examples and fingerprint_comparison.should_retrain_nlu():
+            fingerprint_comparison.core = True
+
         core_merge_failed = False
         if not fingerprint_comparison.should_retrain_core():
             target_path = os.path.join(train_path, DEFAULT_CORE_SUBDIRECTORY_NAME)
diff --git a/rasa/shared/core/training_data/story_reader/yaml_story_reader.py b/rasa/shared/core/training_data/story_reader/yaml_story_reader.py
index 1bb6c7500d81..5a42848f2b48 100644
--- a/rasa/shared/core/training_data/story_reader/yaml_story_reader.py
+++ b/rasa/shared/core/training_data/story_reader/yaml_story_reader.py
@@ -266,7 +266,7 @@ def _parse_step(self, step: Union[Text, Dict[Text, Any]]) -> None:
             )
         elif KEY_USER_MESSAGE in step.keys():
             self._parse_user_message(step)
-        elif KEY_USER_INTENT in step.keys() or KEY_USER_MESSAGE in step.keys():
+        elif KEY_USER_INTENT in step.keys():
             self._parse_labeled_user_utterance(step)
         elif KEY_OR in step.keys():
             self._parse_or_statement(step)
@@ -317,9 +317,16 @@ def _parse_raw_user_utterance(self, step: Dict[Text, Any]) -> Optional[UserUtter
         return UserUttered(None, intent, entities)
 
     def _parse_user_message(self, step: Dict[Text, Any]) -> None:
-        is_end_to_end_utterance = KEY_USER_INTENT not in step
+        has_user_text_and_intent = KEY_USER_INTENT in step
 
-        if is_end_to_end_utterance:
+        if has_user_text_and_intent and not self.use_e2e:
+            rasa.shared.utils.io.raise_warning(
+                f'Story step contains both "{KEY_USER_MESSAGE}" and "{KEY_USER_INTENT}"'
+                f" keys. This can only be used in test stories, not training."
+            )
+            return
+
+        if not has_user_text_and_intent:
             intent = {"name": None}
         else:
             intent_name = self._user_intent_from_step(step)
diff --git a/rasa/shared/importers/importer.py b/rasa/shared/importers/importer.py
index b88b7d2d57a2..73aeb24fcd0d 100644
--- a/rasa/shared/importers/importer.py
+++ b/rasa/shared/importers/importer.py
@@ -100,8 +100,7 @@ def load_core_importer_from_config(
         importer = TrainingDataImporter.load_from_config(
             config_path, domain_path, training_data_paths, TrainingType.CORE
         )
-
-        return CoreDataImporter(importer)
+        return importer
 
     @staticmethod
     def load_nlu_importer_from_config(
diff --git a/rasa/shared/nlu/training_data/message.py b/rasa/shared/nlu/training_data/message.py
index b80e9a44ccac..3f62ec43bd92 100644
--- a/rasa/shared/nlu/training_data/message.py
+++ b/rasa/shared/nlu/training_data/message.py
@@ -324,13 +324,14 @@ def _combine_features(
 
         return combined_features
 
-    def is_core_message(self) -> bool:
-        """Checks whether the message is a core message or not.
+    def is_core_or_domain_message(self) -> bool:
+        """Checks whether the message is a core message or from the domain.
 
-        E.g. a core message is created from a story, not from the NLU data.
+        E.g. a core message is created from a story or a domain action,
+        not from the NLU data.
 
         Returns:
-            True, if message is a core message, false otherwise.
+            True, if message is a core or domain message, false otherwise.
         """
         return bool(
             self.data.get(ACTION_NAME)
@@ -344,3 +345,14 @@ def is_core_message(self) -> bool:
                 and not (self.data.get(INTENT) or self.data.get(RESPONSE))
             )
         )
+
+    def is_e2e_message(self) -> bool:
+        """Checks whether the message came from an e2e story.
+
+        Returns:
+            True, if message is a from an e2e story.
+        """
+        return bool(
+            (self.get(ACTION_TEXT) and not self.get(ACTION_NAME))
+            or (self.get(TEXT) and not self.get(INTENT))
+        )
diff --git a/rasa/shared/nlu/training_data/training_data.py b/rasa/shared/nlu/training_data/training_data.py
index 920fd023f15a..14f2d0df9ea0 100644
--- a/rasa/shared/nlu/training_data/training_data.py
+++ b/rasa/shared/nlu/training_data/training_data.py
@@ -162,7 +162,16 @@ def sanitize_examples(examples: List[Message]) -> List[Message]:
 
     @lazy_property
     def nlu_examples(self) -> List[Message]:
-        return [ex for ex in self.training_examples if not ex.is_core_message()]
+        """Returns examples which have come from NLU training data.
+
+        E.g. If the example came from a story or domain it is not included.
+
+        Returns:
+            List of NLU training examples.
+        """
+        return [
+            ex for ex in self.training_examples if not ex.is_core_or_domain_message()
+        ]
 
     @lazy_property
     def intent_examples(self) -> List[Message]:
@@ -659,7 +668,6 @@ def print_stats(self) -> None:
 
     def is_empty(self) -> bool:
         """Checks if any training data was loaded."""
-
         lists_to_check = [
             self.training_examples,
             self.entity_synonyms,
@@ -668,7 +676,7 @@ def is_empty(self) -> bool:
         ]
         return not any([len(lst) > 0 for lst in lists_to_check])
 
-    def can_train_nlu_model(self) -> bool:
+    def contains_no_pure_nlu_data(self) -> bool:
         """Checks if any NLU training data was loaded."""
         lists_to_check = [
             self.nlu_examples,
@@ -678,6 +686,10 @@ def can_train_nlu_model(self) -> bool:
         ]
         return not any([len(lst) > 0 for lst in lists_to_check])
 
+    def has_e2e_examples(self):
+        """Checks if there are any training examples from e2e stories."""
+        return any(message.is_e2e_message() for message in self.training_examples)
+
 
 def list_to_str(lst: List[Text], delim: Text = ", ", quote: Text = "'") -> Text:
     return delim.join([quote + e + quote for e in lst])
diff --git a/rasa/train.py b/rasa/train.py
index 802ba6dc99e6..8011b2e430ad 100644
--- a/rasa/train.py
+++ b/rasa/train.py
@@ -157,7 +157,7 @@ async def _train_async_internal(
         file_importer.get_stories(), file_importer.get_nlu_data()
     )
 
-    if stories.is_empty() and nlu_data.can_train_nlu_model():
+    if stories.is_empty() and nlu_data.contains_no_pure_nlu_data():
         print_error(
             "No training data given. Please provide stories and NLU data in "
             "order to train a Rasa model using the '--data' argument."
@@ -174,7 +174,7 @@ async def _train_async_internal(
             additional_arguments=nlu_additional_arguments,
         )
 
-    if nlu_data.can_train_nlu_model():
+    if nlu_data.is_empty():
         print_warning("No NLU data present. Just a Rasa Core model will be trained.")
         return await _train_core_with_validated_data(
             file_importer,
@@ -188,7 +188,7 @@ async def _train_async_internal(
 
     if not force_training:
         fingerprint_comparison = model.should_retrain(
-            new_fingerprint, old_model, train_path
+            new_fingerprint, old_model, train_path, nlu_data.has_e2e_examples()
         )
     else:
         fingerprint_comparison = FingerprintComparisonResult(force_training=True)
@@ -349,7 +349,19 @@ async def train_core_async(
     file_importer = TrainingDataImporter.load_core_importer_from_config(
         config, domain, [stories]
     )
-    domain = await file_importer.get_domain()
+    stories, nlu_data, domain = await asyncio.gather(
+        file_importer.get_stories(),
+        file_importer.get_nlu_data(),
+        file_importer.get_domain(),
+    )
+
+    if nlu_data.has_e2e_examples():
+        print_error(
+            "Stories file contains e2e stories. Please train using `rasa train` so that"
+            " the NLU model is also trained."
+        )
+        return None
+
     if domain.is_empty():
         print_error(
             "Core training was skipped because no valid domain file was found. "
@@ -495,7 +507,7 @@ async def _train_nlu_async(
     )
 
     training_data = await file_importer.get_nlu_data()
-    if training_data.can_train_nlu_model():
+    if training_data.contains_no_pure_nlu_data():
         print_error(
             f"Path '{nlu_data}' doesn't contain valid NLU data in it. "
             f"Please verify the data format. "
diff --git a/tests/conftest.py b/tests/conftest.py
index 9a49dd4bc3d6..aac2b10dc7ff 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,6 +1,8 @@
 import asyncio
 import os
 import random
+import shutil
+
 import pytest
 import sys
 import uuid
@@ -35,10 +37,12 @@
 from rasa.utils.common import TempDirectoryPath
 from tests.core.conftest import (
     DEFAULT_DOMAIN_PATH_WITH_SLOTS,
+    DEFAULT_E2E_STORIES_FILE,
     DEFAULT_STACK_CONFIG,
     DEFAULT_STORIES_FILE,
     END_TO_END_STORY_FILE,
     INCORRECT_NLU_DATA,
+    SIMPLE_STORIES_FILE,
 )
 
 DEFAULT_CONFIG_PATH = "rasa/cli/default_config.yml"
@@ -137,6 +141,16 @@ def default_stories_file() -> Text:
     return DEFAULT_STORIES_FILE
 
 
+@pytest.fixture(scope="session")
+def default_e2e_stories_file() -> Text:
+    return DEFAULT_E2E_STORIES_FILE
+
+
+@pytest.fixture(scope="session")
+def simple_stories_file() -> Text:
+    return SIMPLE_STORIES_FILE
+
+
 @pytest.fixture(scope="session")
 def default_stack_config() -> Text:
     return DEFAULT_STACK_CONFIG
@@ -153,7 +167,7 @@ def incorrect_nlu_data() -> Text:
 
 
 @pytest.fixture(scope="session")
-def end_to_end_story_file() -> Text:
+def end_to_end_test_story_file() -> Text:
     return END_TO_END_STORY_FILE
 
 
@@ -191,6 +205,22 @@ async def trained_rasa_model(
     return trained_stack_model_path
 
 
+@pytest.fixture(scope="session")
+async def trained_simple_rasa_model(
+    trained_async: Callable,
+    default_domain_path: Text,
+    default_nlu_data: Text,
+    simple_stories_file: Text,
+) -> Text:
+    trained_stack_model_path = await trained_async(
+        domain=default_domain_path,
+        config=DEFAULT_STACK_CONFIG,
+        training_files=[default_nlu_data, simple_stories_file],
+    )
+
+    return trained_stack_model_path
+
+
 @pytest.fixture(scope="session")
 async def trained_core_model(
     trained_async: Callable,
@@ -224,6 +254,21 @@ async def trained_nlu_model(
     return trained_nlu_model_path
 
 
+@pytest.fixture(scope="session")
+async def trained_e2e_model(
+    trained_async,
+    default_domain_path,
+    default_stack_config,
+    default_nlu_data,
+    default_e2e_stories_file,
+) -> Text:
+    return await trained_async(
+        domain=default_domain_path,
+        config=default_stack_config,
+        training_files=[default_nlu_data, default_e2e_stories_file],
+    )
+
+
 @pytest.fixture(scope="session")
 def moodbot_domain() -> Domain:
     domain_path = os.path.join("examples", "moodbot", "domain.yml")
diff --git a/tests/core/conftest.py b/tests/core/conftest.py
index 45e104816ba2..d32b81c6c116 100644
--- a/tests/core/conftest.py
+++ b/tests/core/conftest.py
@@ -30,7 +30,11 @@
 
 DEFAULT_DOMAIN_PATH_WITH_MAPPING = "data/test_domains/default_with_mapping.yml"
 
-DEFAULT_STORIES_FILE = "data/test_stories/stories_defaultdomain.md"
+DEFAULT_STORIES_FILE = "data/test_yaml_stories/stories_defaultdomain.yml"
+
+DEFAULT_E2E_STORIES_FILE = "data/test_yaml_stories/stories_e2e.yml"
+
+SIMPLE_STORIES_FILE = "data/test_yaml_stories/stories_simple.yml"
 
 DEFAULT_STACK_CONFIG = "data/test_config/stack_config.yml"
 
diff --git a/tests/core/test_training.py b/tests/core/test_training.py
index 3757a0cfce02..d0d25b477f5a 100644
--- a/tests/core/test_training.py
+++ b/tests/core/test_training.py
@@ -14,7 +14,12 @@
 from rasa.core.policies.form_policy import FormPolicy
 
 from rasa.shared.core.training_data.visualization import visualize_stories
-from tests.core.conftest import DEFAULT_DOMAIN_PATH_WITH_SLOTS, DEFAULT_STORIES_FILE
+from tests.core.conftest import (
+    DEFAULT_DOMAIN_PATH_WITH_SLOTS,
+    DEFAULT_E2E_STORIES_FILE,
+    DEFAULT_STACK_CONFIG,
+    DEFAULT_STORIES_FILE,
+)
 
 
 @pytest.mark.parametrize(
diff --git a/tests/shared/core/training_data/story_reader/test_yaml_story_reader.py b/tests/shared/core/training_data/story_reader/test_yaml_story_reader.py
index b795136cfc81..6906c3fee350 100644
--- a/tests/shared/core/training_data/story_reader/test_yaml_story_reader.py
+++ b/tests/shared/core/training_data/story_reader/test_yaml_story_reader.py
@@ -377,8 +377,7 @@ def test_end_to_end_story_with_shortcut_intent():
     """
 
     story_as_yaml = rasa.shared.utils.io.read_yaml(story)
-
-    steps = YAMLStoryReader().read_from_parsed_yaml(story_as_yaml)
+    steps = YAMLStoryReader(use_e2e=True).read_from_parsed_yaml(story_as_yaml)
     user_uttered = steps[0].events[0]
 
     assert user_uttered == UserUttered(
@@ -452,3 +451,25 @@ def test_or_statement_if_not_training_mode():
 
     assert or_statement[0].intent["name"] == "intent1"
     assert or_statement[1].intent["name"] == "intent2"
+
+
+@pytest.mark.parametrize("is_conversation_test, no_events", [(True, 3), (False, 2)])
+def test_gracefully_handles_mixed_steps_in_non_test_e2e_stories(
+    is_conversation_test, no_events
+):
+    stories = """
+    stories:
+    - story: hello world
+      steps:
+      - user: Hi
+      - bot: Hello?
+      - user: Well...
+        intent: suspicion
+    """
+
+    reader = YAMLStoryReader(use_e2e=is_conversation_test)
+    yaml_content = rasa.shared.utils.io.read_yaml(stories)
+
+    steps = reader.read_from_parsed_yaml(yaml_content)
+
+    assert len(steps[0].events) == no_events
diff --git a/tests/shared/nlu/training_data/test_message.py b/tests/shared/nlu/training_data/test_message.py
index 802f6026bf28..25ad054668f9 100644
--- a/tests/shared/nlu/training_data/test_message.py
+++ b/tests/shared/nlu/training_data/test_message.py
@@ -252,7 +252,7 @@ def test_features_present(
 
 
 @pytest.mark.parametrize(
-    "message, core_message",
+    "message, result",
     [
         (Message({INTENT: "intent", TEXT: "text"}), False),
         (Message({RESPONSE: "response", TEXT: "text"}), False),
@@ -262,7 +262,7 @@ def test_features_present(
         (Message({TEXT: "text"}), True),
     ],
 )
-def test_is_core_message(
-    message: Message, core_message: bool,
+def test_is_core_or_domain_message(
+    message: Message, result: bool,
 ):
-    assert core_message == message.is_core_message()
+    assert result == message.is_core_or_domain_message()
diff --git a/tests/test_server.py b/tests/test_server.py
index 5f683fbcf081..c18d6735bbe4 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -731,9 +731,9 @@ async def test_evaluate_stories_not_ready_agent(
 
 
 async def test_evaluate_stories_end_to_end(
-    rasa_app: SanicASGITestClient, end_to_end_story_file: Text
+    rasa_app: SanicASGITestClient, end_to_end_test_story_file
 ):
-    stories = rasa.shared.utils.io.read_file(end_to_end_story_file)
+    stories = rasa.shared.utils.io.read_file(end_to_end_test_story_file)
 
     _, response = await rasa_app.post("/model/test/stories?e2e=true", data=stories)
 
diff --git a/tests/test_train.py b/tests/test_train.py
index 14d855691b14..663ec8863188 100644
--- a/tests/test_train.py
+++ b/tests/test_train.py
@@ -1,9 +1,10 @@
 import asyncio
+import secrets
 import sys
 import tempfile
 import os
-from pathlib import Path
-from typing import Text, Dict
+from pathlib import Path, PosixPath
+from typing import Any, Text, Dict
 from unittest.mock import Mock
 
 import pytest
@@ -12,10 +13,10 @@
 
 import rasa.model
 import rasa.core
+import rasa.nlu
 import rasa.shared.importers.autoconfig as autoconfig
+import rasa.shared.utils.io
 from rasa.core.interpreter import RasaNLUInterpreter
-from rasa.shared.core.domain import Domain
-from rasa.shared.importers.importer import TrainingDataImporter
 
 from rasa.train import train_core, train_nlu, train
 from tests.conftest import DEFAULT_CONFIG_PATH, DEFAULT_NLU_DATA
@@ -358,3 +359,227 @@ def test_train_nlu_autoconfig(
     mocked_get_configuration.assert_called_once()
     _, args, _ = mocked_get_configuration.mock_calls[0]
     assert args[1] == autoconfig.TrainingType.NLU
+
+
+def mock_async(monkeypatch, target, name):
+    mock = Mock()
+
+    async def mock_async_func(*args: Any, **kwargs: Any) -> None:
+        mock(*args, **kwargs)
+
+    monkeypatch.setattr(target, name, mock_async_func)
+    return mock
+
+
+def mock_core_training(monkeypatch):
+    return mock_async(monkeypatch, rasa.core, rasa.core.train.__name__)
+
+
+def mock_nlu_training(monkeypatch):
+    return mock_async(monkeypatch, rasa.nlu, rasa.nlu.train.__name__)
+
+
+def new_model_path_in_same_dir(old_model_path) -> str:
+    return str(Path(old_model_path).parent / (secrets.token_hex(8) + ".tar.gz"))
+
+
+class TestE2e:
+    def test_models_not_retrained_if_no_new_data(
+        self,
+        monkeypatch: MonkeyPatch,
+        trained_e2e_model: Text,
+        default_domain_path,
+        default_stack_config,
+        default_e2e_stories_file,
+        default_nlu_data,
+    ):
+        mocked_nlu_training = mock_nlu_training(monkeypatch)
+        mocked_core_training = mock_core_training(monkeypatch)
+
+        train(
+            default_domain_path,
+            default_stack_config,
+            [default_e2e_stories_file, default_nlu_data],
+            output=new_model_path_in_same_dir(trained_e2e_model),
+        )
+
+        mocked_core_training.assert_not_called()
+        mocked_nlu_training.assert_not_called()
+
+    def test_retrains_nlu_and_core_if_new_e2e_example(
+        self,
+        monkeypatch: MonkeyPatch,
+        trained_e2e_model: Text,
+        default_domain_path,
+        default_stack_config,
+        default_e2e_stories_file,
+        default_nlu_data,
+    ):
+        stories_yaml = rasa.shared.utils.io.read_yaml_file(default_e2e_stories_file)
+        stories_yaml["stories"][1]["steps"].append({"user": "new message!"})
+
+        new_stories_file = tempfile.NamedTemporaryFile(mode="w", suffix=".yml").name
+
+        rasa.shared.utils.io.write_yaml(stories_yaml, new_stories_file)
+
+        mocked_nlu_training = mock_nlu_training(monkeypatch)
+        mocked_core_training = mock_core_training(monkeypatch)
+
+        new_model_path = train(
+            default_domain_path,
+            default_stack_config,
+            [new_stories_file, default_nlu_data],
+            output=new_model_path_in_same_dir(trained_e2e_model),
+        )
+        os.remove(new_model_path)
+
+        mocked_core_training.assert_called_once()
+        mocked_nlu_training.assert_called_once()
+
+    def test_retrains_only_core_if_new_e2e_example_seen_before(
+        self,
+        monkeypatch: MonkeyPatch,
+        trained_e2e_model: Text,
+        default_domain_path,
+        default_stack_config,
+        default_e2e_stories_file,
+        default_nlu_data,
+    ):
+        stories_yaml = rasa.shared.utils.io.read_yaml_file(default_e2e_stories_file)
+        stories_yaml["stories"][1]["steps"].append({"user": "Yes"})
+
+        new_stories_file = tempfile.NamedTemporaryFile(mode="w", suffix=".yml").name
+
+        rasa.shared.utils.io.write_yaml(stories_yaml, new_stories_file)
+
+        mocked_nlu_training = mock_nlu_training(monkeypatch)
+        mocked_core_training = mock_core_training(monkeypatch)
+
+        new_model_path = train(
+            default_domain_path,
+            default_stack_config,
+            [new_stories_file, default_nlu_data],
+            output=new_model_path_in_same_dir(trained_e2e_model),
+        )
+        os.remove(new_model_path)
+
+        mocked_core_training.assert_called_once()
+        mocked_nlu_training.assert_not_called()
+
+    def test_nlu_and_core_trained_if_no_nlu_data_but_e2e_stories(
+        self,
+        monkeypatch: MonkeyPatch,
+        default_domain_path,
+        default_stack_config,
+        default_e2e_stories_file,
+        tmp_path,
+    ):
+
+        mocked_nlu_training = mock_nlu_training(monkeypatch)
+        mocked_core_training = mock_core_training(monkeypatch)
+
+        output = self.make_tmp_model_dir(tmp_path)
+        train(
+            default_domain_path,
+            default_stack_config,
+            [default_e2e_stories_file],
+            output=output,
+        )
+
+        mocked_core_training.assert_called_once()
+        mocked_nlu_training.assert_called_once()
+
+    def make_tmp_model_dir(self, tmp_path):
+        (tmp_path / "models").mkdir()
+        output = str(tmp_path / "models")
+        return output
+
+    def test_new_nlu_data_retrains_core_if_there_are_e2e_stories(
+        self,
+        monkeypatch: MonkeyPatch,
+        trained_e2e_model: Text,
+        default_domain_path,
+        default_stack_config,
+        default_e2e_stories_file,
+        default_nlu_data,
+    ):
+        nlu_yaml = rasa.shared.utils.io.read_yaml_file(default_nlu_data)
+        nlu_yaml["nlu"][0]["examples"] += "- surprise!\n"
+
+        new_nlu_file = tempfile.NamedTemporaryFile(mode="w", suffix=".yml").name
+
+        rasa.shared.utils.io.write_yaml(nlu_yaml, new_nlu_file)
+
+        mocked_nlu_training = mock_nlu_training(monkeypatch)
+        mocked_core_training = mock_core_training(monkeypatch)
+
+        new_model_path = train(
+            default_domain_path,
+            default_stack_config,
+            [default_e2e_stories_file, new_nlu_file],
+            output=new_model_path_in_same_dir(trained_e2e_model),
+        )
+        os.remove(new_model_path)
+
+        mocked_core_training.assert_called_once()
+        mocked_nlu_training.assert_called_once()
+
+    def test_new_nlu_data_does_not_retrain_core_if_there_are_no_e2e_stories(
+        self,
+        monkeypatch: MonkeyPatch,
+        trained_simple_rasa_model: Text,
+        default_domain_path,
+        default_stack_config,
+        simple_stories_file,
+        default_nlu_data,
+    ):
+        nlu_yaml = rasa.shared.utils.io.read_yaml_file(default_nlu_data)
+        nlu_yaml["nlu"][0]["examples"] += "- surprise!\n"
+
+        new_nlu_file = tempfile.NamedTemporaryFile(mode="w", suffix=".yml").name
+
+        rasa.shared.utils.io.write_yaml(nlu_yaml, new_nlu_file)
+
+        mocked_nlu_training = mock_nlu_training(monkeypatch)
+        mocked_core_training = mock_core_training(monkeypatch)
+
+        new_model_path = train(
+            default_domain_path,
+            default_stack_config,
+            [simple_stories_file, new_nlu_file],
+            output=new_model_path_in_same_dir(trained_simple_rasa_model),
+        )
+        os.remove(new_model_path)
+
+        mocked_core_training.assert_not_called()
+        mocked_nlu_training.assert_called_once()
+
+    def test_training_core_with_e2e_fails_gracefully(
+        self,
+        capsys: CaptureFixture,
+        monkeypatch: MonkeyPatch,
+        tmp_path,
+        default_domain_path,
+        default_stack_config,
+        default_e2e_stories_file,
+    ):
+
+        mocked_nlu_training = mock_nlu_training(monkeypatch)
+        mocked_core_training = mock_core_training(monkeypatch)
+
+        output = self.make_tmp_model_dir(tmp_path)
+        train_core(
+            default_domain_path,
+            default_stack_config,
+            default_e2e_stories_file,
+            output=output,
+        )
+
+        mocked_core_training.assert_not_called()
+        mocked_nlu_training.assert_not_called()
+
+        captured = capsys.readouterr()
+        assert (
+            "Stories file contains e2e stories. "
+            + "Please train using `rasa train` so that the NLU model is also trained."
+        ) in captured.out


Code review comments:
diff_hunk: @@ -317,9 +317,16 @@ def _parse_raw_user_utterance(self, step: Dict[Text, Any]) -> Optional[UserUtter
         return UserUttered(None, intent, entities)
 
     def _parse_user_message(self, step: Dict[Text, Any]) -> None:
-        is_end_to_end_utterance = KEY_USER_INTENT not in step
+        has_user_text_and_intent = KEY_USER_INTENT in step
 
-        if is_end_to_end_utterance:
+        if has_user_text_and_intent and not self.use_e2e:
path: rasa/shared/core/training_data/story_reader/yaml_story_reader.py
text: @user2:
@user1 I don't think it is correct, since you mentioned that `self.use_e2e` will be removed

@user2:
actually we want to allow training on e2e test stories: https://github.com/RasaHQ/rasa/issues/7360

@author:
Just so we are clear `self.use_e2e` indicates an e2e conversation _test_ story right?

This change was to address the point:

> do not allow both intent: and user: in training data

And without it I think we get some tensor-flow input shape errors when training core. (If we have mixed `user` and `intent` steps in the training story)



@user2:
yes, but we should fix it inside core featurizers, I can quickly do it and push to your branch

@user2:
pushed a fix, let me know if it works, when you allow both intent and text go through

diff_hunk: @@ -174,7 +174,7 @@ async def _train_async_internal(
             additional_arguments=nlu_additional_arguments,
         )
 
-    if nlu_data.can_train_nlu_model():
+    if nlu_data.is_empty():
path: rasa/train.py
text: @author:
This means we will train NLU even if the only NLU data is from e2e stories.

@user1:
Shall we keep it as a comment?