fix: stop message.context source/destination (#706)

* refactor: reduce bus spam

* fix: stop message routing

Author: JarbasAl

ovos_core/intent_services/stop_service.py

@@ -7,7 +7,7 @@
 from langcodes import closest_match
 from ovos_bus_client.client import MessageBusClient
 from ovos_bus_client.message import Message
-from ovos_bus_client.session import SessionManager
+from ovos_bus_client.session import SessionManager, UtteranceState
 
 from ovos_config.config import Configuration
 from ovos_plugin_manager.templates.pipeline import ConfidenceMatcherPipeline, IntentHandlerMatch
@@ -29,12 +29,17 @@ def __init__(self, bus: Optional[Union[MessageBusClient, FakeBus]] = None,
         self._voc_cache = {}
         self.load_resource_files()
         self.bus.on("stop:global", self.handle_global_stop)
+        self.bus.on("stop:skill", self.handle_skill_stop)
 
     def handle_global_stop(self, message: Message):
         self.bus.emit(message.forward("mycroft.stop"))
         # TODO - this needs a confirmation dialog if nothing was stopped
         self.bus.emit(message.forward("ovos.utterance.handled"))
 
+    def handle_skill_stop(self, message: Message):
+        skill_id = message.data["skill_id"]
+        self.bus.emit(message.reply(f"{skill_id}.stop"))
+
     def load_resource_files(self):
         base = f"{dirname(__file__)}/locale"
         for lang in os.listdir(base):
@@ -148,11 +153,21 @@ def handle_stop_confirmation(self, message: Message):
             error_msg = message.data['error']
             LOG.error(f"{skill_id}: {error_msg}")
         elif message.data.get('result', False):
-            # force-kill any ongoing get_response/converse/TTS - see @killable_event decorator
-            self.bus.emit(message.forward("mycroft.skills.abort_question", {"skill_id": skill_id}))
-            self.bus.emit(message.forward("ovos.skills.converse.force_timeout", {"skill_id": skill_id}))
-            # TODO - track if speech is coming from this skill! not currently tracked
-            self.bus.emit(message.reply("mycroft.audio.speech.stop", {"skill_id": skill_id}))
+            sess = SessionManager.get(message)
+            utt_state = sess.utterance_states.get(skill_id, UtteranceState.INTENT)
+            if utt_state == UtteranceState.RESPONSE:
+                LOG.debug("Forcing get_response timeout")
+                # force-kill any ongoing get_response - see @killable_event decorator (ovos-workshop)
+                self.bus.emit(message.reply("mycroft.skills.abort_question", {"skill_id": skill_id}))
+            if sess.is_active(skill_id):
+                LOG.debug("Forcing converse timeout")
+                # force-kill any ongoing converse - see @killable_event decorator (ovos-workshop)
+                self.bus.emit(message.reply("ovos.skills.converse.force_timeout", {"skill_id": skill_id}))
+
+            # TODO - track if speech is coming from this skill! not currently tracked (ovos-audio)
+            if sess.is_speaking:
+                # force-kill any ongoing TTS
+                self.bus.emit(message.forward("mycroft.audio.speech.stop", {"skill_id": skill_id}))
 
     def match_high(self, utterances: List[str], lang: str, message: Message) -> Optional[IntentHandlerMatch]:
         """
@@ -211,8 +226,8 @@ def match_high(self, utterances: List[str], lang: str, message: Message) -> Opti
                 sess.disable_response_mode(skill_id)
                 self.bus.once(f"{skill_id}.stop.response", self.handle_stop_confirmation)
                 return IntentHandlerMatch(
-                    match_type=f"{skill_id}.stop",
-                    match_data={"conf": conf},
+                    match_type="stop:skill",
+                    match_data={"conf": conf, "skill_id": skill_id},
                     updated_session=sess,
                     utterance=utterance,
                     skill_id="stop.openvoiceos"
@@ -299,8 +314,8 @@ def match_low(self, utterances: List[str], lang: str, message: Message) -> Optio
             sess.disable_response_mode(skill_id)
             self.bus.once(f"{skill_id}.stop.response", self.handle_stop_confirmation)
             return IntentHandlerMatch(
-                match_type=f"{skill_id}.stop",
-                match_data={"conf": conf},
+                match_type="stop:skill",
+                match_data={"conf": conf, "skill_id": skill_id},
                 updated_session=sess,
                 utterance=utterance,
                 skill_id="stop.openvoiceos"

test/end2end/test_stop.py

@@ -166,7 +166,7 @@ def make_it_count():
             nonlocal session
             message = Message("recognizer_loop:utterance",
                               {"utterances": ["count to infinity"], "lang": "en-US"},
-                              {"session": session.serialize()})
+                              {"session": session.serialize(), "source": "A", "destination": "B"})
             session.activate_skill(self.skill_id)  # ensure in active skill list
             self.minicroft.bus.emit(message)
 
@@ -177,33 +177,41 @@ def make_it_count():
 
         message = Message("recognizer_loop:utterance",
                           {"utterances": ["stop"], "lang": "en-US"},
-                          {"session": session.serialize()})  # skill in active list now
+                          {"session": session.serialize(), "source": "A", "destination": "B"})
 
         stop_skill_active = [
             message,
-            Message("ovos-skill-count.openvoiceos.stop.ping",
+            Message(f"{self.skill_id}.stop.ping",
                     {"skill_id":self.skill_id}),
             Message("skill.stop.pong",
                     {"skill_id": self.skill_id, "can_handle": True},
                     {"skill_id": self.skill_id}),
 
             Message("stop.openvoiceos.activate",
                     context={"skill_id": "stop.openvoiceos"}),
+            Message("stop:skill",
+                    context={"skill_id": "stop.openvoiceos"}),
             Message(f"{self.skill_id}.stop",
                     context={"skill_id": "stop.openvoiceos"}),
             Message(f"{self.skill_id}.stop.response",
                     {"skill_id": self.skill_id, "result": True},
                     {"skill_id": self.skill_id}),
 
-            # skill callback to stop everything
-            # TODO - clean up! most arent needed/can check session if needed (ovos-workshop)
-            Message("mycroft.skills.abort_question", {"skill_id": self.skill_id},
-                    {"skill_id": self.skill_id}),
-            Message("ovos.skills.converse.force_timeout", {"skill_id": self.skill_id},
-                    {"skill_id": self.skill_id}),
-            Message("mycroft.audio.speech.stop", {"skill_id": self.skill_id},
+            # stop pipeline callback to stop everything
+
+            # if skill is in middle of get_response
+            #Message("mycroft.skills.abort_question", {"skill_id": self.skill_id},
+            #        {"skill_id": self.skill_id}),
+
+            # if skill is in active_list
+            Message("ovos.skills.converse.force_timeout",
+                    {"skill_id": self.skill_id},
                     {"skill_id": self.skill_id}),
 
+            # if skill is executing TTS
+            #Message("mycroft.audio.speech.stop", {"skill_id": self.skill_id},
+            #        {"skill_id": self.skill_id}),
+
             # the intent running in the daemon thread exits cleanly
             Message("mycroft.skill.handler.complete",
                     {"name": "CountSkill.handle_how_are_you_intent"},
@@ -214,10 +222,15 @@ def make_it_count():
         ]
         test = End2EndTest(
             minicroft=self.minicroft,
-            # inject_active=[self.skill_id],  # ensure this skill is in active skills list for the test
             skill_ids=[],
             eof_msgs=[],
             flip_points=["recognizer_loop:utterance"],
+            # messages in 'keep_original_src' would not be sent to hivemind clients
+            # i.e. they are directed towards ovos-core
+            keep_original_src=[f"{self.skill_id}.stop.ping",
+                               f"{self.skill_id}.stop",
+                               "mycroft.skills.abort_question",
+                               "ovos.skills.converse.force_timeout"],
             ignore_messages=self.ignore_messages,
             source_message=message,
             expected_messages=stop_skill_active