Reset slot value for set_slots once flow ends

rasa/core/policies/flow_policy.py

@@ -559,16 +559,40 @@ def _reset_scoped_slots(
         self, current_flow: Flow, tracker: DialogueStateTracker
     ) -> List[Event]:
         """Reset all scoped slots."""
+
+        def _reset_slot(
+            slot_name: Text, dialogue_tracker: DialogueStateTracker
+        ) -> None:
+            slot = dialogue_tracker.slots.get(slot_name, None)
+            initial_value = slot.initial_value if slot else None
+            events.append(SlotSet(slot_name, initial_value))
+
         events: List[Event] = []
+
+        not_resettable_slot_names = set()
+
         for step in current_flow.steps:
-            # reset all slots scoped to the flow
-            if (
-                isinstance(step, CollectInformationFlowStep)
-                and step.reset_after_flow_ends
-            ):
-                slot = tracker.slots.get(step.collect, None)
-                initial_value = slot.initial_value if slot else None
-                events.append(SlotSet(step.collect, initial_value))
+            if isinstance(step, CollectInformationFlowStep):
+                # reset all slots scoped to the flow
+                if step.reset_after_flow_ends:
+                    _reset_slot(step.collect, tracker)
+                else:
+                    not_resettable_slot_names.add(step.collect)
+
+        # slots set by the set slots step should be reset after the flow ends
+        # unless they are also used in a collect step where `reset_after_flow_ends`
+        # is set to `False`
+        resettable_set_slots = [
+            slot["key"]
+            for step in current_flow.steps
+            if isinstance(step, SetSlotsFlowStep)
+            for slot in step.slots
+            if slot["key"] not in not_resettable_slot_names
+        ]
+
+        for name in resettable_set_slots:
+            _reset_slot(name, tracker)
+
         return events
 
     def run_step(

tests/core/policies/test_flow_policy.py

@@ -16,6 +16,7 @@
 from rasa.shared.core.events import ActionExecuted, Event, SlotSet
 from rasa.shared.core.flows.flow import FlowsList
 from rasa.shared.core.flows.yaml_flows_io import YAMLFlowsReader
+from rasa.shared.core.slots import TextSlot
 from rasa.shared.core.trackers import DialogueStateTracker
 from rasa.dialogue_understanding.stack.frames import (
     UserFlowStackFrame,
@@ -328,3 +329,88 @@ def test_executor_does_not_get_tripped_if_an_action_is_predicted_in_loop():
 
     selection = executor.select_next_action(tracker)
     assert selection.action_name == "action_listen"
+
+
+def test_flow_policy_resets_all_slots_after_flow_ends() -> None:
+    flows = flows_from_str(
+        """
+        flows:
+          foo_flow:
+            steps:
+            - id: "1"
+              collect: my_slot
+            - id: "2"
+              set_slots:
+                - foo: bar
+                - other_slot: other_value
+            - id: "3"
+              action: action_listen
+        """
+    )
+    tracker = DialogueStateTracker.from_events(
+        "test",
+        [
+            SlotSet("my_slot", "my_value"),
+            SlotSet("foo", "bar"),
+            SlotSet("other_slot", "other_value"),
+            ActionExecuted("action_listen"),
+        ],
+        slots=[
+            TextSlot("my_slot", mappings=[], initial_value="initial_value"),
+            TextSlot("foo", mappings=[]),
+            TextSlot("other_slot", mappings=[]),
+        ],
+    )
+
+    domain = Domain.empty()
+    executor = FlowExecutor.from_tracker(tracker, flows, domain)
+
+    current_flow = flows.flow_by_id("foo_flow")
+    events = executor._reset_scoped_slots(current_flow, tracker)
+    assert events == [
+        SlotSet("my_slot", "initial_value"),
+        SlotSet("foo", None),
+        SlotSet("other_slot", None),
+    ]
+
+
+def test_flow_policy_set_slots_inherit_reset_from_collect_step() -> None:
+    """Test that `reset_after_flow_ends` is inherited from the collect step."""
+    slot_name = "my_slot"
+    flows = flows_from_str(
+        f"""
+        flows:
+          foo_flow:
+            steps:
+            - id: "1"
+              collect: {slot_name}
+              reset_after_flow_ends: false
+            - id: "2"
+              set_slots:
+                - foo: bar
+                - {slot_name}: my_value
+            - id: "3"
+              action: action_listen
+        """
+    )
+    tracker = DialogueStateTracker.from_events(
+        "test123",
+        [
+            SlotSet("my_slot", "my_value"),
+            SlotSet("foo", "bar"),
+            ActionExecuted("action_listen"),
+        ],
+        slots=[
+            TextSlot("my_slot", mappings=[], initial_value="initial_value"),
+            TextSlot("foo", mappings=[]),
+        ],
+    )
+
+    domain = Domain.empty()
+    executor = FlowExecutor.from_tracker(tracker, flows, domain)
+
+    current_flow = flows.flow_by_id("foo_flow")
+    events = executor._reset_scoped_slots(current_flow, tracker)
+    assert events == [
+        SlotSet("foo", None),
+    ]