Repository: RasaHQ/rasa
Pull request: Return all sessions for GET /story
**Proposed changes**:
- fixes #6860 

**Status (please check what you already did)**:
- [x] added some tests for the functionality
- [x] updated the documentation
- [x] 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:
/story endpoint does not fetch full tracker
**Issue**: The `GET /story` endpoint currently only fetches the conversation after the last session start. The endpoint should
1. fetch the story for the entire tracker
2. not cause a new session start sequence if the session expiration time has elapsed (this endpoint should not modify the tracker)

Hints from issue discussion:

Commit message: fix deepsource checks

Patch to review:
diff --git a/changelog/6860.improvement.md b/changelog/6860.improvement.md
new file mode 100644
index 000000000000..ac2af3b64a8c
--- /dev/null
+++ b/changelog/6860.improvement.md
@@ -0,0 +1,12 @@
+Fetching test stories from the HTTP API endpoint 
+`GET /conversations/<conversation_id>/story` no longer triggers an update 
+of the conversation session.
+
+Added a new boolean query parameter `all_sessions` (default: `false`) to the HTTP API 
+endpoint for fetching test stories (`GET /conversations/<conversation_id>/story`). 
+
+When setting `?all_sessions=true`, the endpoint returns test stories for all 
+conversation sessions for `conversation_id`. 
+When setting `?all_sessions=all_sessions`, or when omitting the `all_sessions` 
+parameter, a single test story is returned for `conversation_id`. In cases where 
+multiple conversation sessions exist, only the last story is returned.
diff --git a/docs/static/spec/rasa.yml b/docs/static/spec/rasa.yml
index f63fa1e50c2e..a1671e0ea16f 100644
--- a/docs/static/spec/rasa.yml
+++ b/docs/static/spec/rasa.yml
@@ -245,6 +245,7 @@ paths:
       parameters:
       - $ref: '#/components/parameters/conversation_id'
       - $ref: '#/components/parameters/until'
+      - $ref: '#/components/parameters/all_sessions'
       responses:
         200:
           $ref: '#/components/responses/200Story'
@@ -791,6 +792,15 @@ components:
         type: boolean
         default: false
       required: false
+    all_sessions:
+      in: query
+      name: all_sessions
+      description: Fetch test stories for all conversation sessions
+      example: false
+      schema:
+        type: boolean
+        default: false
+      required: false
     model:
       in: query
       name: model
diff --git a/rasa/core/processor.py b/rasa/core/processor.py
index f1c25746f818..44eedd59cc3a 100644
--- a/rasa/core/processor.py
+++ b/rasa/core/processor.py
@@ -2,7 +2,7 @@
 import os
 import time
 from types import LambdaType
-from typing import Any, Dict, List, Optional, Text, Tuple, Union
+from typing import Any, Dict, List, Optional, Text, Tuple, Union, TYPE_CHECKING
 
 import numpy as np
 
@@ -14,6 +14,7 @@
     OutputChannel,
     UserMessage,
 )
+import rasa.core.utils
 from rasa.shared.core.constants import (
     USER_INTENT_RESTART,
     ACTION_LISTEN_NAME,
@@ -42,11 +43,14 @@
 )
 from rasa.core.nlg import NaturalLanguageGenerator
 from rasa.core.policies.ensemble import PolicyEnsemble
-from rasa.core.tracker_store import TrackerStore
+import rasa.core.tracker_store
 from rasa.shared.core.trackers import DialogueStateTracker, EventVerbosity
 from rasa.shared.nlu.constants import INTENT_NAME_KEY
 from rasa.utils.endpoints import EndpointConfig
 
+if TYPE_CHECKING:
+    from rasa.core.tracker_store import TrackerStore
+
 logger = logging.getLogger(__name__)
 
 MAX_NUMBER_OF_PREDICTIONS = int(os.environ.get("MAX_NUMBER_OF_PREDICTIONS", "10"))
@@ -58,7 +62,7 @@ def __init__(
         interpreter: NaturalLanguageInterpreter,
         policy_ensemble: PolicyEnsemble,
         domain: Domain,
-        tracker_store: TrackerStore,
+        tracker_store: "TrackerStore",
         generator: NaturalLanguageGenerator,
         action_endpoint: Optional[EndpointConfig] = None,
         max_number_of_predictions: int = MAX_NUMBER_OF_PREDICTIONS,
@@ -175,7 +179,7 @@ async def get_tracker_with_session_start(
         sender_id: Text,
         output_channel: Optional[OutputChannel] = None,
         metadata: Optional[Dict] = None,
-    ) -> Optional[DialogueStateTracker]:
+    ) -> DialogueStateTracker:
         """Get tracker for `sender_id` or create a new tracker for `sender_id`.
 
         If a new tracker is created, `action_session_start` is run.
@@ -190,14 +194,12 @@ async def get_tracker_with_session_start(
         """
 
         tracker = self.get_tracker(sender_id)
-        if not tracker:
-            return None
 
         await self._update_tracker_session(tracker, output_channel, metadata)
 
         return tracker
 
-    def get_tracker(self, conversation_id: Text) -> Optional[DialogueStateTracker]:
+    def get_tracker(self, conversation_id: Text) -> DialogueStateTracker:
         """Get the tracker for a conversation.
 
         In contrast to `get_tracker_with_session_start` this does not add any
@@ -213,10 +215,39 @@ def get_tracker(self, conversation_id: Text) -> Optional[DialogueStateTracker]:
             conversation.
         """
         conversation_id = conversation_id or DEFAULT_SENDER_ID
+
         return self.tracker_store.get_or_create_tracker(
             conversation_id, append_action_listen=False
         )
 
+    def get_trackers_for_all_conversation_sessions(
+        self, conversation_id: Text,
+    ) -> List[DialogueStateTracker]:
+        """Get all trackers for a conversation.
+
+        Individual trackers are returned for each conversation session found
+        for `conversation_id`.
+
+        Args:
+            conversation_id: The ID of the conversation for which the trackers should
+                be retrieved.
+
+        Returns:
+            Trackers for the conversation.
+        """
+        import rasa.shared.core.trackers
+
+        conversation_id = conversation_id or DEFAULT_SENDER_ID
+
+        with rasa.core.tracker_store.tracker_store_with_full_conversation_retrieval(
+            self.tracker_store
+        ):
+            tracker = self.tracker_store.get_or_create_tracker(
+                conversation_id, append_action_listen=False
+            )
+
+        return rasa.shared.core.trackers.get_trackers_for_conversation_sessions(tracker)
+
     async def log_message(
         self, message: UserMessage, should_save_tracker: bool = True
     ) -> Optional[DialogueStateTracker]:
diff --git a/rasa/core/tracker_store.py b/rasa/core/tracker_store.py
index 2b2db3279f87..50d97f7697fd 100644
--- a/rasa/core/tracker_store.py
+++ b/rasa/core/tracker_store.py
@@ -18,6 +18,7 @@
     Text,
     Union,
     TYPE_CHECKING,
+    Generator,
 )
 
 from boto3.dynamodb.conditions import Key
@@ -40,9 +41,7 @@
     DialogueStateTracker,
     EventVerbosity,
 )
-import rasa.cli.utils as rasa_cli_utils
 from rasa.shared.nlu.constants import INTENT_NAME_KEY
-from rasa.utils import common as common_utils
 from rasa.utils.endpoints import EndpointConfig
 import sqlalchemy as sa
 
@@ -95,11 +94,10 @@ def create(
         event_broker: Optional[EventBroker] = None,
     ) -> "TrackerStore":
         """Factory to create a tracker store."""
-
         if isinstance(obj, TrackerStore):
             return obj
-        else:
-            return _create_from_endpoint_config(obj, domain, event_broker)
+
+        return _create_from_endpoint_config(obj, domain, event_broker)
 
     def get_or_create_tracker(
         self,
@@ -1115,3 +1113,22 @@ def _load_from_module_name_in_endpoint_config(
             f"Using `InMemoryTrackerStore` instead."
         )
         return InMemoryTrackerStore(domain)
+
+
+@contextlib.contextmanager
+def tracker_store_with_full_conversation_retrieval(
+    tracker_store: "TrackerStore",
+) -> Generator[None, None, None]:
+    """Context manager that ensures `tracker_store` fetches all tracker sessions.
+
+    Args:
+        tracker_store: Tracker store to modify.
+    """
+    previous_value = tracker_store.load_events_from_previous_conversation_sessions
+
+    tracker_store.load_events_from_previous_conversation_sessions = True
+
+    try:
+        yield
+    finally:
+        tracker_store.load_events_from_previous_conversation_sessions = previous_value
diff --git a/rasa/core/utils.py b/rasa/core/utils.py
index 289de4900934..06e94a91a873 100644
--- a/rasa/core/utils.py
+++ b/rasa/core/utils.py
@@ -38,6 +38,7 @@
 from sanic.views import CompositionView
 import rasa.cli.utils as cli_utils
 
+
 logger = logging.getLogger(__name__)
 
 
diff --git a/rasa/server.py b/rasa/server.py
index 7ad1c54bd1c7..46efea857384 100644
--- a/rasa/server.py
+++ b/rasa/server.py
@@ -227,27 +227,85 @@ def event_verbosity_parameter(
         )
 
 
-async def get_tracker(
+async def get_tracker_with_session_start(
     processor: "MessageProcessor", conversation_id: Text
 ) -> DialogueStateTracker:
-    """Get tracker object from `MessageProcessor`."""
-    tracker = await processor.get_tracker_with_session_start(conversation_id)
-    _validate_tracker(tracker, conversation_id)
+    """Get tracker object from `MessageProcessor` and update the conversation session.
 
-    # `_validate_tracker` ensures we can't return `None` so `Optional` is not needed
-    return tracker  # pytype: disable=bad-return-type
+    Args:
+        processor: An instance of `MessageProcessor`.
+        conversation_id: Conversation ID to fetch the tracker for.
 
+    Returns:
+        The tracker for `conversation_id` with an updated conversation session.
+    """
+    return await processor.get_tracker_with_session_start(conversation_id)
 
-def _validate_tracker(
-    tracker: Optional[DialogueStateTracker], conversation_id: Text
-) -> None:
-    if not tracker:
-        raise ErrorResponse(
-            409,
-            "Conflict",
-            f"Could not retrieve tracker with ID '{conversation_id}'. Most likely "
-            f"because there is no domain set on the agent.",
-        )
+
+def get_test_stories(
+    processor: "MessageProcessor",
+    conversation_id: Text,
+    until_time: Optional[float],
+    fetch_all_sessions: bool = False,
+) -> Text:
+    """Retrieves test stories from `processor` for all conversation sessions for
+    `conversation_id`.
+
+    Args:
+        processor: An instance of `MessageProcessor`.
+        conversation_id: Conversation ID to fetch stories for.
+        until_time: Timestamp up to which to include events.
+        fetch_all_sessions: Whether to fetch stories for all conversation sessions.
+            If `False`, only the last conversation session is retrieved.
+
+    Returns:
+        The stories for `conversation_id` in test format.
+    """
+    if fetch_all_sessions:
+        trackers: List[
+            DialogueStateTracker
+        ] = processor.get_trackers_for_all_conversation_sessions(conversation_id)
+    else:
+        trackers = [processor.get_tracker(conversation_id)]
+
+    if until_time is not None:
+        trackers = [tracker.travel_back_in_time(until_time) for tracker in trackers]
+        # keep only non-empty trackers
+        trackers = [tracker for tracker in trackers if len(tracker.events)]
+
+    logger.debug(
+        f"Fetched trackers for {len(trackers)} conversation sessions "
+        f"for conversation ID {conversation_id}."
+    )
+
+    story_steps = []
+
+    more_than_one_story = len(trackers) > 1
+
+    for i, tracker in enumerate(trackers, 1):
+        for story_step in tracker.as_story().story_steps:
+            story_step.block_name = conversation_id
+            if more_than_one_story:
+                story_step.block_name += f", story {i}"
+
+            story_steps.append(story_step)
+
+    return YAMLStoryWriter().dumps(story_steps, is_test_story=True)
+
+
+def get_tracker(
+    processor: "MessageProcessor", conversation_id: Text
+) -> DialogueStateTracker:
+    """Retrieves tracker from `processor` without updating the conversation session.
+
+    Args:
+        processor: An instance of `MessageProcessor`.
+        conversation_id: Conversation ID to fetch the tracker for.
+
+    Returns:
+        The tracker for `conversation_id`.
+    """
+    return processor.get_tracker(conversation_id)
 
 
 def validate_request_body(request: Request, error_message: Text):
@@ -474,7 +532,9 @@ async def retrieve_tracker(request: Request, conversation_id: Text):
         verbosity = event_verbosity_parameter(request, EventVerbosity.AFTER_RESTART)
         until_time = rasa.utils.endpoints.float_arg(request, "until")
 
-        tracker = await get_tracker(app.agent.create_processor(), conversation_id)
+        tracker = await get_tracker_with_session_start(
+            app.agent.create_processor(), conversation_id
+        )
 
         try:
             if until_time is not None:
@@ -506,7 +566,6 @@ async def append_events(request: Request, conversation_id: Text):
                 processor = app.agent.create_processor()
                 tracker = processor.get_tracker(conversation_id)
                 output_channel = _get_output_channel(request, tracker)
-                _validate_tracker(tracker, conversation_id)
 
                 events = _get_events_from_request_body(request)
 
@@ -584,19 +643,19 @@ async def replace_events(request: Request, conversation_id: Text):
     @ensure_loaded_agent(app)
     async def retrieve_story(request: Request, conversation_id: Text):
         """Get an end-to-end story corresponding to this conversation."""
-
-        # retrieve tracker and set to requested state
-        tracker = await get_tracker(app.agent.create_processor(), conversation_id)
-
         until_time = rasa.utils.endpoints.float_arg(request, "until")
+        fetch_all_sessions = rasa.utils.endpoints.bool_arg(
+            request, "all_sessions", default=False
+        )
 
         try:
-            if until_time is not None:
-                tracker = tracker.travel_back_in_time(until_time)
-
-            # dump and return tracker
-            state = YAMLStoryWriter().dumps(tracker.as_story().story_steps)
-            return response.text(state)
+            stories = get_test_stories(
+                app.agent.create_processor(),
+                conversation_id,
+                until_time,
+                fetch_all_sessions=fetch_all_sessions,
+            )
+            return response.text(stories)
         except Exception as e:
             logger.debug(traceback.format_exc())
             raise ErrorResponse(
@@ -625,7 +684,7 @@ async def execute_action(request: Request, conversation_id: Text):
 
         try:
             async with app.agent.lock_store.lock(conversation_id):
-                tracker = await get_tracker(
+                tracker = await get_tracker_with_session_start(
                     app.agent.create_processor(), conversation_id
                 )
                 output_channel = _get_output_channel(request, tracker)
@@ -643,7 +702,9 @@ async def execute_action(request: Request, conversation_id: Text):
                 500, "ConversationError", f"An unexpected error occurred. Error: {e}"
             )
 
-        tracker = await get_tracker(app.agent.create_processor(), conversation_id)
+        tracker = await get_tracker_with_session_start(
+            app.agent.create_processor(), conversation_id
+        )
         state = tracker.current_state(verbosity)
 
         response_body = {"tracker": state}
@@ -674,7 +735,7 @@ async def trigger_intent(request: Request, conversation_id: Text) -> HTTPRespons
 
         try:
             async with app.agent.lock_store.lock(conversation_id):
-                tracker = await get_tracker(
+                tracker = await get_tracker_with_session_start(
                     app.agent.create_processor(), conversation_id
                 )
                 output_channel = _get_output_channel(request, tracker)
diff --git a/rasa/shared/core/events.py b/rasa/shared/core/events.py
index 9e50238b5ba7..efb485cd61e4 100644
--- a/rasa/shared/core/events.py
+++ b/rasa/shared/core/events.py
@@ -4,11 +4,10 @@
 
 import jsonpickle
 import time
-import typing
 import uuid
 from dateutil import parser
 from datetime import datetime
-from typing import List, Dict, Text, Any, Type, Optional
+from typing import List, Dict, Text, Any, Type, Optional, TYPE_CHECKING
 
 import rasa.shared.utils.common
 from typing import Union
@@ -31,7 +30,7 @@
     INTENT_NAME_KEY,
 )
 
-if typing.TYPE_CHECKING:
+if TYPE_CHECKING:
     from rasa.shared.core.trackers import DialogueStateTracker
 
 logger = logging.getLogger(__name__)
diff --git a/rasa/shared/core/trackers.py b/rasa/shared/core/trackers.py
index ba6518f8582d..fab6414f9f0a 100644
--- a/rasa/shared/core/trackers.py
+++ b/rasa/shared/core/trackers.py
@@ -41,6 +41,7 @@
     LOOP_REJECTED,
     TRIGGER_MESSAGE,
     LOOP_INTERRUPTED,
+    ACTION_SESSION_START_NAME,
 )
 from rasa.shared.core.conversation import Dialogue  # pytype: disable=pyi-error
 from rasa.shared.core.events import (  # pytype: disable=pyi-error
@@ -130,7 +131,7 @@ def from_events(
         cls,
         sender_id: Text,
         evts: List[Event],
-        slots: Optional[List[Slot]] = None,
+        slots: Optional[Iterable[Slot]] = None,
         max_event_history: Optional[int] = None,
         sender_source: Optional[Text] = None,
     ):
@@ -407,9 +408,21 @@ def generate_all_prior_trackers(
 
         yield tracker
 
-    def applied_events(self) -> List[Event]:
-        """Returns all actions that should be applied - w/o reverted events."""
+    def applied_events(
+        self,
+        include_restarts: bool = False,
+        include_previous_conversation_sessions: bool = False,
+    ) -> List[Event]:
+        """Returns all actions that should be applied - w/o reverted events.
+
+        Args:
+            include_restarts: Whether events should be returned across restarts.
+            include_previous_conversation_sessions: Whether events should be returned
+                across conversation sessions.
 
+        Returns:
+            The events applied to the tracker.
+        """
         loop_names = [
             event.name
             for event in self.events
@@ -419,7 +432,12 @@ def applied_events(self) -> List[Event]:
         applied_events = []
 
         for event in self.events:
-            if isinstance(event, (Restarted, SessionStarted)):
+            if isinstance(event, Restarted) and not include_restarts:
+                applied_events = []
+            elif (
+                isinstance(event, SessionStarted)
+                and not include_previous_conversation_sessions
+            ):
                 applied_events = []
             elif isinstance(event, ActionReverted):
                 self._undo_till_previous(ActionExecuted, applied_events)
@@ -759,7 +777,6 @@ def _merge_slots(
         ]
         return new_slots
 
-    # pytype: disable=bad-return-type
     @property
     def active_loop_name(self) -> Optional[Text]:
         """Get the name of the currently active loop.
@@ -769,9 +786,7 @@ def active_loop_name(self) -> Optional[Text]:
         if not self.active_loop or self.active_loop.get(LOOP_NAME) == SHOULD_NOT_BE_SET:
             return None
 
-        return self.active_loop.get(LOOP_NAME)
-
-    # pytype: enable=bad-return-type
+        return self.active_loop.get(LOOP_NAME)  # pytype: disable=bad-return-type
 
     @property
     def latest_action_name(self) -> Optional[Text]:
@@ -813,3 +828,56 @@ def is_prev_action_listen_in_state(state: State) -> bool:
     """
     prev_action_name = state.get(PREVIOUS_ACTION, {}).get(ACTION_NAME)
     return prev_action_name == ACTION_LISTEN_NAME
+
+
+def get_trackers_for_conversation_sessions(
+    tracker: "DialogueStateTracker",
+) -> List["DialogueStateTracker"]:
+    """Generate trackers for `tracker` that are split by conversation sessions.
+
+    Args:
+        tracker: Instance of `DialogueStateTracker` to split.
+
+    Returns:
+        The trackers split by conversation sessions.
+    """
+    trackers: List["DialogueStateTracker"] = []
+    tracker_slots = tracker.slots
+
+    applied_events = tracker.applied_events(
+        include_restarts=True, include_previous_conversation_sessions=True
+    )
+
+    def _create_empty_tracker() -> DialogueStateTracker:
+        return DialogueStateTracker.from_events(
+            tracker.sender_id, [], tracker_slots, sender_source=tracker.sender_source,
+        )
+
+    processed_first_session_start = False
+    current_tracker = _create_empty_tracker()
+
+    for i, event in enumerate(applied_events):
+        is_session_started_event = (
+            isinstance(event, ActionExecuted)
+            and event.action_name == ACTION_SESSION_START_NAME
+        )
+
+        if is_session_started_event and not processed_first_session_start:
+            # first session start or restart
+            current_tracker.update(event)
+            processed_first_session_start = True
+        elif is_session_started_event:
+            # session start or restart, append to old one and start new tracker
+            trackers.append(current_tracker)
+
+            current_tracker = _create_empty_tracker()
+            current_tracker.update(event)
+        elif i == len(applied_events) - 1:
+            # last event, append the tracker
+            current_tracker.update(event)
+            trackers.append(current_tracker)
+        else:
+            # ordinary event
+            current_tracker.update(event)
+
+    return trackers
diff --git a/rasa/shared/core/training_data/story_writer/yaml_story_writer.py b/rasa/shared/core/training_data/story_writer/yaml_story_writer.py
index 1e0253a36a08..2bd163c324c3 100644
--- a/rasa/shared/core/training_data/story_writer/yaml_story_writer.py
+++ b/rasa/shared/core/training_data/story_writer/yaml_story_writer.py
@@ -124,6 +124,7 @@ def stories_to_yaml(
             result[KEY_STORIES] = stories
         if rules:
             result[KEY_RULES] = rules
+
         return result
 
     def process_story_step(self, story_step: StoryStep) -> OrderedDict:
diff --git a/tests/core/test_tracker_stores.py b/tests/core/test_tracker_stores.py
index cce97db7d747..db2e5530fd5d 100644
--- a/tests/core/test_tracker_stores.py
+++ b/tests/core/test_tracker_stores.py
@@ -830,3 +830,19 @@ def test_current_state_without_events(default_domain: Domain):
 
     # `events` key should not be in there
     assert state and "events" not in state
+
+
+@pytest.mark.parametrize("initial_value", [True, False])
+def test_tracker_store_with_full_conversation_retrieval(initial_value: bool):
+    tracker_store = InMemoryTrackerStore(Domain.empty())
+    tracker_store.load_events_from_previous_conversation_sessions = initial_value
+
+    with rasa.core.tracker_store.tracker_store_with_full_conversation_retrieval(
+        tracker_store
+    ):
+        assert tracker_store.load_events_from_previous_conversation_sessions
+
+    # the initial value is restored after leaving the context manager
+    assert (
+        tracker_store.load_events_from_previous_conversation_sessions == initial_value
+    )
diff --git a/tests/shared/core/test_trackers.py b/tests/shared/core/test_trackers.py
index 9d96599b0609..04649445ba24 100644
--- a/tests/shared/core/test_trackers.py
+++ b/tests/shared/core/test_trackers.py
@@ -1,6 +1,7 @@
 import json
 import logging
 import os
+import time
 from pathlib import Path
 import tempfile
 from typing import List, Text, Dict, Any, Type
@@ -463,8 +464,6 @@ def test_traveling_back_in_time(default_domain: Domain):
     tracker.update(ActionExecuted(ACTION_LISTEN_NAME))
     tracker.update(UserUttered("/greet", intent, []))
 
-    import time
-
     time.sleep(1)
     time_for_timemachine = time.time()
     time.sleep(1)
@@ -1177,3 +1176,31 @@ def test_set_form_validation_deprecation_warning(validate: bool):
         tracker.set_form_validation(validate)
 
     assert tracker.active_loop[LOOP_INTERRUPTED] == (not validate)
+
+
+def test_trackers_for_conversation_sessions():
+    import rasa.shared.core.trackers as trackers_module
+
+    # conversation contains multiple sessions
+    conversation_events = [
+        ActionExecuted(ACTION_SESSION_START_NAME),
+        SessionStarted(),
+        UserUttered("hi", {"name": "greet"}),
+        ActionExecuted("utter_greet"),
+        # second session begins here
+        ActionExecuted(ACTION_SESSION_START_NAME),
+        SessionStarted(),
+        UserUttered("goodbye", {"name": "goodbye"}),
+        ActionExecuted("utter_goodbye"),
+    ]
+
+    tracker = DialogueStateTracker.from_events(
+        "some-conversation-ID", conversation_events
+    )
+
+    subtrackers = trackers_module.get_trackers_for_conversation_sessions(tracker)
+
+    assert len(subtrackers) == 2
+
+    assert list(subtrackers[0].events) == conversation_events[:4]
+    assert list(subtrackers[1].events) == conversation_events[4:]
diff --git a/tests/test_server.py b/tests/test_server.py
index 0e14917dcb76..90ffbf3769ac 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -6,8 +6,9 @@
 import requests
 import time
 import uuid
+import urllib.parse
 
-from typing import List, Text, Type, Generator, NoReturn, Dict
+from typing import List, Text, Type, Generator, NoReturn, Dict, Optional
 from contextlib import ExitStack
 
 from _pytest import pathlib
@@ -25,6 +26,7 @@
 import rasa.utils.io
 import rasa.server
 from rasa.core import utils
+from rasa.core.tracker_store import InMemoryTrackerStore
 from rasa.shared.core import events
 from rasa.core.agent import Agent
 from rasa.core.channels import (
@@ -35,7 +37,16 @@
     CallbackInput,
 )
 from rasa.core.channels.slack import SlackBot
-from rasa.shared.core.events import Event, UserUttered, SlotSet, BotUttered
+from rasa.shared.core.constants import ACTION_SESSION_START_NAME
+from rasa.shared.core.domain import Domain, SessionConfig
+from rasa.shared.core.events import (
+    Event,
+    UserUttered,
+    SlotSet,
+    BotUttered,
+    ActionExecuted,
+    SessionStarted,
+)
 from rasa.shared.core.trackers import DialogueStateTracker
 from rasa.model import unpack_model
 from rasa.shared.nlu.constants import INTENT_NAME_KEY
@@ -180,8 +191,6 @@ def mocked_training_function(*_, **__) -> Text:
         return fake_model_path
 
     def run_server() -> NoReturn:
-        import rasa
-
         rasa.train = mocked_training_function
 
         from rasa import __main__
@@ -1320,3 +1329,213 @@ class NoInputChannels:
         request, DialogueStateTracker.from_events("default", [])
     )
     assert isinstance(actual, CollectingOutputChannel)
+
+
+@pytest.mark.parametrize(
+    "conversation_events,until_time,fetch_all_sessions,expected",
+    # conversation with one session
+    [
+        (
+            [
+                ActionExecuted(ACTION_SESSION_START_NAME),
+                SessionStarted(),
+                UserUttered("hi", {"name": "greet"}),
+                ActionExecuted("utter_greet"),
+            ],
+            None,
+            True,
+            """version: "2.0"
+stories:
+- story: some-conversation-ID
+  steps:
+  - intent: greet
+    user: |-
+      hi
+  - action: utter_greet""",
+        ),
+        # conversation with multiple sessions
+        (
+            [
+                ActionExecuted(ACTION_SESSION_START_NAME),
+                SessionStarted(),
+                UserUttered("hi", {"name": "greet"}),
+                ActionExecuted("utter_greet"),
+                ActionExecuted(ACTION_SESSION_START_NAME),
+                SessionStarted(),
+                UserUttered("bye bye", {"name": "goodbye"}),
+                ActionExecuted("utter_goodbye"),
+            ],
+            None,
+            True,
+            """version: "2.0"
+stories:
+- story: some-conversation-ID, story 1
+  steps:
+  - intent: greet
+    user: |-
+      hi
+  - action: utter_greet
+- story: some-conversation-ID, story 2
+  steps:
+  - intent: goodbye
+    user: |-
+      bye bye
+  - action: utter_goodbye""",
+        ),
+        # conversation with multiple sessions, but setting `all_sessions=false`
+        # means only the last one is returned
+        (
+            [
+                ActionExecuted(ACTION_SESSION_START_NAME),
+                SessionStarted(),
+                UserUttered("hi", {"name": "greet"}),
+                ActionExecuted("utter_greet"),
+                ActionExecuted(ACTION_SESSION_START_NAME),
+                SessionStarted(),
+                UserUttered("bye bye", {"name": "goodbye"}),
+                ActionExecuted("utter_goodbye"),
+            ],
+            None,
+            False,
+            """version: "2.0"
+stories:
+- story: some-conversation-ID
+  steps:
+  - intent: goodbye
+    user: |-
+      bye bye
+  - action: utter_goodbye""",
+        ),
+        # the default for `all_sessions` is `false` - this test checks that
+        # only the latest session is returned in that case
+        (
+            [
+                ActionExecuted(ACTION_SESSION_START_NAME),
+                SessionStarted(),
+                UserUttered("hi", {"name": "greet"}),
+                ActionExecuted("utter_greet"),
+                ActionExecuted(ACTION_SESSION_START_NAME),
+                SessionStarted(),
+                UserUttered("bye bye", {"name": "goodbye"}),
+                ActionExecuted("utter_goodbye"),
+            ],
+            None,
+            None,
+            """version: "2.0"
+stories:
+- story: some-conversation-ID
+  steps:
+  - intent: goodbye
+    user: |-
+      bye bye
+  - action: utter_goodbye""",
+        ),
+        # `until` parameter means only the first session is returned
+        (
+            [
+                ActionExecuted(ACTION_SESSION_START_NAME, timestamp=1),
+                SessionStarted(timestamp=2),
+                UserUttered("hi", {"name": "greet"}, timestamp=3),
+                ActionExecuted("utter_greet", timestamp=4),
+                ActionExecuted(ACTION_SESSION_START_NAME, timestamp=5),
+                SessionStarted(timestamp=6),
+                UserUttered("bye bye", {"name": "goodbye"}, timestamp=7),
+                ActionExecuted("utter_goodbye", timestamp=8),
+            ],
+            4,
+            True,
+            """version: "2.0"
+stories:
+- story: some-conversation-ID
+  steps:
+  - intent: greet
+    user: |-
+      hi
+  - action: utter_greet""",
+        ),
+        # empty conversation
+        ([], None, True, 'version: "2.0"'),
+    ],
+)
+async def test_get_story(
+    rasa_app: SanicASGITestClient,
+    conversation_events: List[Event],
+    until_time: Optional[float],
+    fetch_all_sessions: Optional[bool],
+    expected: Text,
+):
+    conversation_id = "some-conversation-ID"
+
+    tracker_store = InMemoryTrackerStore(Domain.empty())
+    tracker = DialogueStateTracker.from_events(conversation_id, conversation_events)
+
+    tracker_store.save(tracker)
+
+    rasa_app.app.agent.tracker_store = tracker_store
+
+    url = f"/conversations/{conversation_id}/story?"
+
+    query = {}
+
+    if fetch_all_sessions is not None:
+        query["all_sessions"] = fetch_all_sessions
+
+    if until_time is not None:
+        query["until"] = until_time
+
+    _, response = await rasa_app.get(url + urllib.parse.urlencode(query))
+
+    assert response.status == 200
+    assert response.content.decode().strip() == expected
+
+
+async def test_get_story_does_not_update_conversation_session(
+    rasa_app: SanicASGITestClient,
+):
+    conversation_id = "some-conversation-ID"
+
+    # domain with short session expiration time of one second
+    domain = Domain.empty()
+    domain.session_config = SessionConfig(
+        session_expiration_time=1 / 60, carry_over_slots=True
+    )
+
+    # conversation contains one session that has expired
+    now = time.time()
+    conversation_events = [
+        ActionExecuted(ACTION_SESSION_START_NAME, timestamp=now - 10),
+        SessionStarted(timestamp=now - 9),
+        UserUttered("hi", {"name": "greet"}, timestamp=now - 8),
+        ActionExecuted("utter_greet", timestamp=now - 7),
+    ]
+
+    tracker = DialogueStateTracker.from_events(conversation_id, conversation_events)
+
+    tracker_store = InMemoryTrackerStore(domain)
+
+    tracker_store.save(tracker)
+
+    rasa_app.app.agent.tracker_store = tracker_store
+
+    _, response = await rasa_app.get(f"/conversations/{conversation_id}/story")
+
+    assert response.status == 200
+
+    # expected story is returned
+    assert (
+        response.content.decode().strip()
+        == """version: "2.0"
+stories:
+- story: some-conversation-ID
+  steps:
+  - intent: greet
+    user: |-
+      hi
+  - action: utter_greet"""
+    )
+
+    # the tracker has the same number of events as were initially added
+    assert len(tracker.events) == len(conversation_events)
+
+    # the last event is still the same as before
+    assert tracker.events[-1].timestamp == conversation_events[-1].timestamp


Code review comments:
diff_hunk: @@ -213,10 +215,39 @@ def get_tracker(self, conversation_id: Text) -> Optional[DialogueStateTracker]:
             conversation.
         """
         conversation_id = conversation_id or DEFAULT_SENDER_ID
+
         return self.tracker_store.get_or_create_tracker(
             conversation_id, append_action_listen=False
         )
 
+    def get_trackers_for_all_conversation_sessions(
+        self, conversation_id: Text,
+    ) -> List[DialogueStateTracker]:
+        """Get all trackers for a conversation.
+
+        Individual trackers are returned for each conversation session found
+        for `conversation_id`.
+
+        Args:
+            conversation_id: The ID of the conversation for which the trackers should
+                be retrieved.
+
+        Returns:
+            Trackers for the conversation.
+        """
+        import rasa.shared.core.trackers
path: rasa/core/processor.py
text: @user1:
it's already imported as top level import. A local import doesn't add any benefit then.

diff_hunk: @@ -813,3 +828,56 @@ def is_prev_action_listen_in_state(state: State) -> bool:
     """
     prev_action_name = state.get(PREVIOUS_ACTION, {}).get(ACTION_NAME)
     return prev_action_name == ACTION_LISTEN_NAME
+
+
+def get_trackers_for_conversation_sessions(
+    tracker: "DialogueStateTracker",
+) -> List["DialogueStateTracker"]:
+    """Generate trackers for `tracker` that are split by conversation sessions.
+
+    Args:
+        tracker: Instance of `DialogueStateTracker` to split.
+
+    Returns:
+        The trackers split by conversation sessions.
+    """
+    trackers: List["DialogueStateTracker"] = []
+    tracker_slots = tracker.slots
+
+    applied_events = tracker.applied_events(
+        include_restarts=True, include_previous_conversation_sessions=True
path: rasa/shared/core/trackers.py
text: @user1:
I think these flags dilute the meaning of `applied_events` a bit. What about splitting at the restarts / session starts and calling `applied_events` on each tracker? 

@author:
great idea!

diff_hunk: @@ -227,27 +227,85 @@ def event_verbosity_parameter(
         )
 
 
-async def get_tracker(
+async def get_tracker_with_session_start(
     processor: "MessageProcessor", conversation_id: Text
 ) -> DialogueStateTracker:
-    """Get tracker object from `MessageProcessor`."""
-    tracker = await processor.get_tracker_with_session_start(conversation_id)
-    _validate_tracker(tracker, conversation_id)
+    """Get tracker object from `MessageProcessor` and update the conversation session.
 
-    # `_validate_tracker` ensures we can't return `None` so `Optional` is not needed
-    return tracker  # pytype: disable=bad-return-type
+    Args:
+        processor: An instance of `MessageProcessor`.
+        conversation_id: Conversation ID to fetch the tracker for.
 
+    Returns:
+        The tracker for `conversation_id` with an updated conversation session.
+    """
+    return await processor.get_tracker_with_session_start(conversation_id)
path: rasa/server.py
text: @user1:
I don't think it's intuitive that a `get_` function has side effects not only on the objects but even on the persistence layer. However, somewhat out of the scope of this PR 😬

diff_hunk: @@ -227,27 +227,85 @@ def event_verbosity_parameter(
         )
 
 
-async def get_tracker(
+async def get_tracker_with_session_start(
     processor: "MessageProcessor", conversation_id: Text
 ) -> DialogueStateTracker:
-    """Get tracker object from `MessageProcessor`."""
-    tracker = await processor.get_tracker_with_session_start(conversation_id)
-    _validate_tracker(tracker, conversation_id)
+    """Get tracker object from `MessageProcessor` and update the conversation session.
 
-    # `_validate_tracker` ensures we can't return `None` so `Optional` is not needed
-    return tracker  # pytype: disable=bad-return-type
+    Args:
+        processor: An instance of `MessageProcessor`.
+        conversation_id: Conversation ID to fetch the tracker for.
 
+    Returns:
+        The tracker for `conversation_id` with an updated conversation session.
+    """
+    return await processor.get_tracker_with_session_start(conversation_id)
 
-def _validate_tracker(
-    tracker: Optional[DialogueStateTracker], conversation_id: Text
-) -> None:
-    if not tracker:
-        raise ErrorResponse(
-            409,
-            "Conflict",
-            f"Could not retrieve tracker with ID '{conversation_id}'. Most likely "
-            f"because there is no domain set on the agent.",
-        )
+
+def get_test_stories(
+    processor: "MessageProcessor",
+    conversation_id: Text,
+    until_time: Optional[float],
+    fetch_all_sessions: bool = False,
+) -> Text:
+    """Retrieves test stories from `processor` for all conversation sessions for
+    `conversation_id`.
+
+    Args:
+        processor: An instance of `MessageProcessor`.
+        conversation_id: Conversation ID to fetch stories for.
+        until_time: Timestamp up to which to include events.
+        fetch_all_sessions: Whether to fetch stories for all conversation sessions.
+            If `False`, only the last conversation session is retrieved.
+
+    Returns:
+        The stories for `conversation_id` in test format.
+    """
+    if fetch_all_sessions:
+        trackers: List[
+            DialogueStateTracker
+        ] = processor.get_trackers_for_all_conversation_sessions(conversation_id)
+    else:
+        trackers = [processor.get_tracker(conversation_id)]
+
+    if until_time is not None:
+        trackers = [tracker.travel_back_in_time(until_time) for tracker in trackers]
+        # keep only non-empty trackers
+        trackers = [tracker for tracker in trackers if len(tracker.events)]
+
+    logger.debug(
+        f"Fetched trackers for {len(trackers)} conversation sessions "
+        f"for conversation ID {conversation_id}."
+    )
+
+    story_steps = []
+
+    more_than_one_story = len(trackers) > 1
+
+    for i, tracker in enumerate(trackers, 1):
+        for story_step in tracker.as_story().story_steps:
+            story_step.block_name = conversation_id
+            if more_than_one_story:
+                story_step.block_name += f", story {i}"
+
+            story_steps.append(story_step)
path: rasa/server.py
text: @user1:
?
```suggestion
    for i, tracker in enumerate(trackers, 1):
        if more_than_one_story:
            tracker.sender_id = f"{tracker.sender_id}, story {i}"
        story_steps += tracker.as_story().story_steps
```

diff_hunk: @@ -42,11 +43,14 @@
 )
 from rasa.core.nlg import NaturalLanguageGenerator
 from rasa.core.policies.ensemble import PolicyEnsemble
-from rasa.core.tracker_store import TrackerStore
+import rasa.core.tracker_store
 from rasa.shared.core.trackers import DialogueStateTracker, EventVerbosity
 from rasa.shared.nlu.constants import INTENT_NAME_KEY
 from rasa.utils.endpoints import EndpointConfig
 
+if TYPE_CHECKING:
+    from rasa.core.tracker_store import TrackerStore
path: rasa/core/processor.py
text: @user1:
doesn't really make sense imo as we already import the entire module in line 46

diff_hunk: @@ -1115,3 +1113,22 @@ def _load_from_module_name_in_endpoint_config(
             f"Using `InMemoryTrackerStore` instead."
         )
         return InMemoryTrackerStore(domain)
+
+
+@contextlib.contextmanager
+def tracker_store_with_full_conversation_retrieval(
+    tracker_store: "TrackerStore",
+) -> Generator[None, None, None]:
+    """Context manager that ensures `tracker_store` fetches all tracker sessions.
+
+    Args:
+        tracker_store: Tracker store to modify.
+    """
+    previous_value = tracker_store.load_events_from_previous_conversation_sessions
+
+    tracker_store.load_events_from_previous_conversation_sessions = True
+
+    try:
path: rasa/core/tracker_store.py
text: @user1:
I'm a bit worried about the implementation as generator. Somebody might see and use this function with an `await` expression inside which would have side effects on other requests which are using trackers.

* Why not making it atomic - means returning the `DialogueStateTracker` right away?
* An alternative would be to add a new function `retrieve_full_tracker` to `TrackerStore` and implement this for the `TrackerStore` class and the trackers which are supporting sessions. By default it could just point to `retrieve` 

@author:
I'm not sure I follow - why would someone use this in an `await` expression? It is a context manager - the decorator and the function description says so.

I could also add an explicit example of how to use it in the function docstring. what do you think?

@user1:
I meant the following use case:

```python
        with rasa.core.tracker_store.tracker_store_with_full_conversation_retrieval(
            self.tracker_store
        ):
            tracker = self.tracker_store.get_or_create_tracker(
                conversation_id, append_action_listen=False
            )
            # imagine any `async` call here
            # the await will allow other requests to be processed while this is blocking
            # These requests will then use a tracker store with disabled sessions.
            await asyncio.sleep(123)
```

@author:
I don't think so - other calls to `self.tracker_store` within that class will happen _outside_ the context manager, so should not be affected by this

@author:
I agree with the suggestion to make this a tracker store method 👍

diff_hunk: @@ -813,3 +828,56 @@ def is_prev_action_listen_in_state(state: State) -> bool:
     """
     prev_action_name = state.get(PREVIOUS_ACTION, {}).get(ACTION_NAME)
     return prev_action_name == ACTION_LISTEN_NAME
+
+
+def get_trackers_for_conversation_sessions(
path: rasa/shared/core/trackers.py
text: @user1:
Can we consolidate this one with [_split_conversation_at_restarts](https://github.com/RasaHQ/rasa/blob/cee30cbacf1655f4796d49d15a9ad56a6ec2dc5e/rasa/core/training/interactive.py#L747)?

@author:
I don't see the benefit in trying to combine the two - the one in `interactive.py` works on serialised events, this one doesn't

@user1:
I don't think we should duplicate the logic to split conversations. E.g. we could make the one in `interactive` work on serialized events, too. It's not ideal for performance, but it will be called just once, so it's fine.

@author:
but the split is a very different one. in interactive we split on the `restart` event - here we split on `ActionExecuted(action_session_start)`. All of this could be parametrised in the function but I don't see the benefit in doing that, especially since we'd also have to deserialise events in the interactive case

@author:
Okay I've just convinced myself 😄   I've moved the split