feat: OVOS-STOP-1 conformance in stop_service (ovos.stop.ping/pong, global ovos.stop, session drain)

ovos_core/intent_services/service.py

@@ -407,7 +407,12 @@ def _dispatch_match(self, match: IntentHandlerMatch, message: Message, lang: str
                 reply.context["skill_id"] = match.skill_id
 
                 was_deactivated = match.skill_id in self._deactivations[sess.session_id]
-                if not was_deactivated:
+                # A Match may opt out of activation entirely (e.g. a targeted stop
+                # whose Match.skill_id is the skill being stopped — activating it
+                # would send it a bogus `.activate` callback and refresh its
+                # recency right as it is torn down).
+                suppress_activation = getattr(match, "suppress_activation", False)
+                if not was_deactivated and not suppress_activation:
                     # OVOS-PIPELINE-1 §7.1 pushes the skill onto the session's
                     # active-handler recency list. §7.3 SUPPRESSES that push for
                     # reserved intent_name dispatches (converse/response/stop/

ovos_core/intent_services/stop_service.py

@@ -5,11 +5,12 @@
 from ovos_bus_client.client import MessageBusClient
 from ovos_bus_client.handler import HandlerLifecycle
 from ovos_bus_client.message import Message
-from ovos_bus_client.session import SessionManager, UtteranceState
+from ovos_bus_client.session import Session, SessionManager, UtteranceState
 
 from ovos_config.config import Configuration
 from ovos_plugin_manager.templates.pipeline import ConfidenceMatcherPipeline, IntentHandlerMatch
 from ovos_spec_tools import LocaleResources
+from ovos_spec_tools.messages import SpecMessage
 from ovos_utils import flatten_list
 from ovos_utils.fakebus import FakeBus
 from ovos_utils.log import LOG
@@ -24,25 +25,136 @@ def __init__(self, bus: Optional[Union[MessageBusClient, FakeBus]] = None,
         config = config if config is not None else Configuration().get("skills", {}).get("stop") or {}
         bus = bus or FakeBus()
         ConfidenceMatcherPipeline.__init__(self, config=config, bus=bus)
+        # OVOS-STOP-1 §3.1: the pipeline_id under which this plugin's global-stop
+        # handler self-dispatches and which a global_stop Match carries as skill_id.
+        self.pipeline_id = "ovos-stop-pipeline-plugin"
         self._locale = LocaleResources(skill_locale=join(dirname(__file__), "locale"))
+        # OVOS-STOP-1 §2/§3.1: a targeted stop is dispatched on the reserved
+        # intent_name ``stop`` (the ``<target_skill_id>:stop`` topic) and a global
+        # stop on ``<pipeline_id>:global_stop``. The plugin owns the global
+        # self-dispatch; the targeted dispatch is caught per match (``bus.once`` on
+        # ``<target>:stop``, in ``match_high``). The ``<target>:stop`` dispatch is
+        # bridged to the legacy ``<target>.stop`` a skill honours by the
+        # ovos-spec-tools namespace translator, so no plugin-side re-emit is needed.
+        self.bus.on(f"{self.pipeline_id}:global_stop", self.handle_global_stop)
+        # Back-compat: retain the legacy internal dispatch topics.
         self.bus.on("stop:global", self.handle_global_stop)
         self.bus.on("stop:skill", self.handle_skill_stop)
 
+    @staticmethod
+    def _select_stop_target(candidates: List[str],
+                            message: Optional[Message] = None) -> Optional[str]:
+        """OVOS-STOP-1 §4.1 step 4 — single-target recency selection.
+
+        From the positive pong responders in ``candidates``, select the one
+        whose ``active_handlers`` entry has the highest ``activated_at``
+        (OVOS-PIPELINE-1 §7.1 recency record). Selection is driven by the
+        spec ``activated_at`` field, NOT the legacy ``active_skills`` list
+        order. On an ``activated_at`` tie, the entry appearing first in the
+        head-first ``active_handlers`` list (the most recently stamped one)
+        wins.
+
+        Returns the selected ``skill_id``, or ``None`` if no candidate is
+        present in ``active_handlers``.
+        """
+        if not candidates:
+            return None
+        session = SessionManager.get(message)
+        # active_handlers is head-first (index 0 == most recently stamped);
+        # enumerate so a lower index breaks an activated_at tie.
+        best = None  # (activated_at, -index, skill_id)
+        for idx, handler in enumerate(session.active_handlers):
+            skill_id = handler.get("skill_id")
+            if skill_id not in candidates:
+                continue
+            key = (handler.get("activated_at", 0.0), -idx)
+            if best is None or key > best[0]:
+                best = (key, skill_id)
+        if best is not None:
+            return best[1]
+        # Defensive: a candidate that is not represented in active_handlers
+        # (should not happen — §4.1 step 3 filters pongs to active_handlers).
+        return candidates[0]
+
+    @staticmethod
+    def _stop_match(**kwargs) -> IntentHandlerMatch:
+        """Build a stop Match that opts out of PIPELINE-1 §7.1 activation.
+
+        A stop is a termination, never a fresh activation: its Match.skill_id is
+        the target being stopped (or the pipeline_id for a global stop), so the
+        orchestrator must not send it a `.activate` callback or refresh its
+        recency. The orchestrator honours ``suppress_activation`` in _dispatch_match.
+        """
+        match = IntentHandlerMatch(**kwargs)
+        match.suppress_activation = True
+        return match
+
     def handle_global_stop(self, message: Message) -> None:
-        """Emit a global mycroft.stop; the §9.5 end-marker is the orchestrator's
-        responsibility (``IntentDispatcher._notify_terminal``)."""
+        """Emit the global stop broadcast.
+
+        OVOS-STOP-1 §5.3: the global-stop handler emits the spec broadcast
+        ``ovos.stop`` (``SpecMessage.STOP``).
+
+        Back-compat: ``mycroft.stop ↔ ovos.stop`` is a payload-compatible rename
+        in the spec-tools MIGRATION_MAP, and the bus bridge re-delivers a spec
+        emit on its legacy counterpart when ``emit_legacy`` is enabled. So
+        emitting ``ovos.stop`` ALSO reaches un-migrated skills still listening on
+        ``mycroft.stop`` without a hand-rolled second emit. A manual
+        ``mycroft.stop`` here would double-deliver to anyone subscribed on both
+        topics, so the bridge is relied on instead.
+        """
+        # §3.1: the global_stop dispatch carries skill_id == pipeline_id, so the
+        # §8 handler-lifecycle done-signal must correlate under that same id.
         with HandlerLifecycle(self.bus, message,
-                              skill_id="stop.openvoiceos",
+                              skill_id=self.pipeline_id,
                               data={"name": "StopService.handle_global_stop"}):
-            self.bus.emit(message.forward("mycroft.stop"))
+            self.bus.emit(message.forward(SpecMessage.STOP))
+
+    @staticmethod
+    def _drain_global_stop_session(session) -> "Session":
+        """OVOS-STOP-1 §5.2 — build the fully-cleaned ``updated_session`` for a
+        ``global_stop`` Match.
+
+        The global_stop Match MUST carry a session with:
+
+        - ``active_handlers``  → ``[]``  (§5.2 / §6.2)
+        - ``converse_handlers`` → ``[]`` (§5.2 / §6.2, OVOS-CONVERSE-1 §2.1)
+        - ``active_skills``    → ``[]``  (legacy recency list read by
+          ``get_active_skills`` → stop-target routing; left stale it could
+          route a later targeted stop to a skill this global_stop drained)
+        - ``response_mode``    → absent  (§5.2 / §6.1)
+
+        All four are cleared atomically at match time so the drained state is
+        committed before dispatch (PIPELINE-1 §4.2).
+        """
+        session.active_handlers = []
+        session.converse_handlers = []
+        session.active_skills = []
+        session.clear_response_mode()
+        return session
 
     def handle_skill_stop(self, message: Message) -> None:
-        """Forward a stop request to the specific skill."""
+        """Deliver a targeted stop to the skill and close the §8 lifecycle.
+
+        The ``<target_skill_id>:stop`` dispatch is the spec-named topic; the
+        skill honours the legacy ``<skill_id>.stop``. When the bus emits legacy
+        counterparts (``emit_legacy``), the ovos-spec-tools namespace translator
+        already bridges ``<skill_id>:stop`` → ``<skill_id>.stop``, so re-emitting
+        here would double-deliver. When it does not (a spec-only or legacy-only
+        bus), re-emit ``<skill_id>.stop`` so the target is stopped under every bus
+        configuration. Either way this runs under the target's skill_id so the §8
+        handler-lifecycle done-signal correlates; the target reports back on
+        ``<skill_id>.stop.response``.
+        """
         skill_id = message.data["skill_id"]
         with HandlerLifecycle(self.bus, message,
-                              skill_id="stop.openvoiceos",
+                              skill_id=skill_id,
                               data={"name": "StopService.handle_skill_stop"}):
-            self.bus.emit(message.reply(f"{skill_id}.stop"))
+            translator = getattr(self.bus, "_translator", None)
+            bridged = bool(translator and
+                           translator.counterpart_topics(f"{skill_id}:stop"))
+            if not bridged:
+                self.bus.emit(message.reply(f"{skill_id}.stop"))
 
     @staticmethod
     def get_active_skills(message: Optional[Message] = None) -> List[str]:
@@ -122,17 +234,30 @@ def handle_ack(msg: Message) -> None:
                 # all skills answered the ping!
                 event.set()
 
-        self.bus.on("skill.stop.pong", handle_ack)
+        # OVOS-STOP-1 §4.2: subscribe the spec pong topic ``ovos.stop.pong``
+        # (SpecMessage.STOP_PONG). It is a 1:1 rename in MIGRATION_MAP, so the bus
+        # bridge ALSO delivers un-migrated skills' legacy ``skill.stop.pong`` here
+        # — one subscription receives both namespaces.
+        self.bus.on(SpecMessage.STOP_PONG, handle_ack)
         try:
-            # ask skills if they can stop
+            # OVOS-STOP-1 §4.1/§4.2: emit the stoppability query as a single
+            # broadcast ``ovos.stop.ping`` (SpecMessage.STOP_PING). The skill_id is
+            # not part of a broadcast payload; responders self-identify in the pong.
+            self.bus.emit(message.forward(SpecMessage.STOP_PING))
+
+            # Back-compat: the per-skill ``{skill_id}.stop.ping`` is a structural
+            # placeholder the broadcast replaces and which CANNOT be statically
+            # bus-bridged (it has no fixed counterpart topic). Skills that listen
+            # only on their per-skill ping are reached by emitting it alongside
+            # the broadcast.
             for skill_id in active_skills:
                 self.bus.emit(message.forward(f"{skill_id}.stop.ping",
                                               {"skill_id": skill_id}))
 
             # wait for all skills to acknowledge they can stop
             event.wait(timeout=0.5)
         finally:
-            self.bus.remove("skill.stop.pong", handle_ack)
+            self.bus.remove(SpecMessage.STOP_PONG, handle_ack)
         return want_stop or active_skills
 
     def handle_stop_confirmation(self, message: Message) -> None:
@@ -193,27 +318,41 @@ def match_high(self, utterances: List[str], lang: str, message: Message) -> Opti
 
         if is_global_stop:
             LOG.info(f"Emitting global stop, {len(self.get_active_skills(message))} active skills")
-            # emit a global stop, full stop anything OVOS is doing
-            return IntentHandlerMatch(
-                match_type="stop:global",
+            # emit a global stop, full stop anything OVOS is doing.
+            # OVOS-STOP-1 §5.2: drain active_handlers + converse_handlers and
+            # clear response_mode in the committed updated_session.
+            return self._stop_match(
+                match_type=f"{self.pipeline_id}:global_stop",
                 match_data={"conf": conf},
-                updated_session=sess,
+                updated_session=self._drain_global_stop_session(sess),
                 utterance=utterance,
-                skill_id="stop.openvoiceos"
+                skill_id=self.pipeline_id
             )
 
         if is_stop:
-            # check if any skill can stop
-            for skill_id in self._collect_stop_skills(message):
+            # OVOS-STOP-1 §4.1 step 4: from the positive pong responders select
+            # the single most-recently-activated target (highest activated_at in
+            # session.active_handlers), not the legacy active_skills order.
+            skill_id = self._select_stop_target(self._collect_stop_skills(message),
+                                                 message)
+            if skill_id:
                 LOG.debug(f"Telling skill to stop: {skill_id}")
+                # OVOS-STOP-1 §6.1: clear the dispatch target's response_mode and
+                # §6.2: remove only that target from active_handlers, via the
+                # committed updated_session.
                 sess.disable_response_mode(skill_id)
+                sess.remove_active_handler(skill_id)
                 self.bus.once(f"{skill_id}.stop.response", self.handle_stop_confirmation)
-                return IntentHandlerMatch(
-                    match_type="stop:skill",
+                # §2/§3.1: dispatch the reserved `stop` on `<target_skill_id>:stop`
+                # (skill_id == the target); catch the own dispatch to close the §8
+                # lifecycle. The bus bridges it to the legacy `<skill_id>.stop`.
+                self.bus.once(f"{skill_id}:stop", self.handle_skill_stop)
+                return self._stop_match(
+                    match_type=f"{skill_id}:stop",
                     match_data={"conf": conf, "skill_id": skill_id},
                     updated_session=sess,
                     utterance=utterance,
-                    skill_id="stop.openvoiceos"
+                    skill_id=skill_id
                 )
 
         return None
@@ -288,27 +427,39 @@ def match_low(self, utterances: List[str], lang: str, message: Message) -> Optio
         if conf < self.config.get("min_conf", 0.5):
             return None
 
-        # check if any skill can stop
-        for skill_id in self._collect_stop_skills(message):
+        # OVOS-STOP-1 §4.1 step 4: select the single most-recently-activated
+        # positive pong responder as the stop target.
+        skill_id = self._select_stop_target(self._collect_stop_skills(message),
+                                            message)
+        if skill_id:
             LOG.debug(f"Telling skill to stop: {skill_id}")
+            # OVOS-STOP-1 §6.1 (clear target response_mode) + §6.2 (remove only
+            # the target from active_handlers) via the committed updated_session.
             sess.disable_response_mode(skill_id)
+            sess.remove_active_handler(skill_id)
             self.bus.once(f"{skill_id}.stop.response", self.handle_stop_confirmation)
-            return IntentHandlerMatch(
-                match_type="stop:skill",
+            # §2/§3.1: dispatch the reserved `stop` on `<target_skill_id>:stop`;
+            # catch the own dispatch to close the §8 lifecycle. The bus bridges it
+            # to the legacy `<skill_id>.stop`.
+            self.bus.once(f"{skill_id}:stop", self.handle_skill_stop)
+            return self._stop_match(
+                match_type=f"{skill_id}:stop",
                 match_data={"conf": conf, "skill_id": skill_id},
                 updated_session=sess,
                 utterance=utterance,
-                skill_id="stop.openvoiceos"
+                skill_id=skill_id
             )
 
-        # emit a global stop, full stop anything OVOS is doing
+        # emit a global stop, full stop anything OVOS is doing.
+        # OVOS-STOP-1 §5.2: drain active_handlers + converse_handlers and clear
+        # response_mode in the committed updated_session.
         LOG.debug(f"Emitting global stop signal, {len(self.get_active_skills(message))} active skills")
-        return IntentHandlerMatch(
-            match_type="stop:global",
+        return self._stop_match(
+            match_type=f"{self.pipeline_id}:global_stop",
             match_data={"conf": conf},
-            updated_session=sess,
+            updated_session=self._drain_global_stop_session(sess),
             utterance=utterance,
-            skill_id="stop.openvoiceos"
+            skill_id=self.pipeline_id
         )
 
     def shutdown(self) -> None:

pyproject.toml

@@ -21,7 +21,7 @@ dependencies = [
     "ovos-config>=0.0.13,<3.0.0",
     "ovos-workshop>=9.0.2a1,<10.0.0",
     "rapidfuzz>=3.6,<4.0",
-    "ovos-spec-tools[langcodes]>=1.1.0a1,<2.0.0",
+    "ovos-spec-tools[langcodes]>=1.3.0a1,<2.0.0",  # <skill_id>:stop -> .stop namespace bridge (STOP-1 §2)
 ]
 
 [project.urls]
@@ -37,7 +37,7 @@ test = [
     "pytest-testmon>=2.1.3",
     "pytest-randomly>=3.16.0",
     "cov-core>=1.15.0",
-    "ovoscope>=1.4.0a1,<2.0.0",
+    "ovoscope>=1.5.0a1,<2.0.0",  # pipeline_id filter (STOP-1 §3.1 concurrent isolation)
     "ovos-m2v-pipeline>=0.3.1a1,<1.0.0",
     "ovos-adapt-parser>=1.4.2a1, <2.0.0",
     "ovos_padatious>=1.4.5a2,<2.0.0",