feat: OVOS-PIPELINE-1 §6.2 required_slots backstop + §7.3 reserved-name suppression

Reworked onto dev's IntentDispatcher (#788) architecture. The matched/unmatched/
handled emission this PR originally carried now lives in #788; the two deltas
that remain unique are re-applied here:

- §7.3: suppress the §7.1 `active_handlers` push for reserved-name dispatches
  (converse/response/stop/fallback/common_query) — a reserved name is a
  continuation/termination of an already-active skill, not a fresh activation.
  `activate_skill` is already a §7.1 shim over `add_active_handler`, so only the
  suppression gate was missing.
- §6.2: orchestrator backstop that drops a match missing any `required_slots`
  (surfaced by the plugin in `match_data['__required_slots__']`) and continues
  iteration — a second line of defense behind engine-side enforcement. No-op and
  fully backward compatible when a plugin does not surface the constraint.

Adds unit coverage for both the backstop and reserved-name suppression.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: JarbasAl

ovos_core/intent_services/service.py

@@ -65,6 +65,33 @@
 
 _PIPELINE_RE = re.compile(r'-(high|medium|low)$')
 
+# OVOS-PIPELINE-1 §7.3 reserved intent_names. A Match produced by one of the
+# reserving pipeline-plugin roles below is a reserved-name dispatch: §7.1
+# requires the ``session.active_handlers`` push to be SUPPRESSED for it, because
+# a reserved name represents a continuation/termination of an already-active
+# skill's participation, not a fresh activation. Keyed off the producing
+# pipeline_id (the role that holds the namespace lease), with the confidence
+# suffix (``-high``/``-medium``/``-low``) stripped before lookup.
+#
+#   converse       -> ovos-converse-pipeline-plugin     (CONVERSE-1 §4/§5: converse, response)
+#   stop           -> ovos-stop-pipeline-plugin          (STOP-1 §4: stop)
+#   fallback       -> ovos-fallback-pipeline-plugin      (FALLBACK-1 §6.3: fallback)
+#   common_query   -> ovos-common-query-pipeline-plugin  (COMMON-QUERY-1 §3: common_query)
+_RESERVED_NAME_PIPELINES = {
+    "ovos-converse-pipeline-plugin",
+    "ovos-stop-pipeline-plugin",
+    "ovos-fallback-pipeline-plugin",
+    "ovos-common-query-pipeline-plugin",
+}
+
+
+def _produces_reserved_name(pipeline_id: Optional[str]) -> bool:
+    """OVOS-PIPELINE-1 §7.3: True when ``pipeline_id`` is a reserved-name role
+    whose dispatches must NOT stamp ``session.active_handlers`` (§7.1)."""
+    if not pipeline_id:
+        return False
+    return _PIPELINE_RE.sub("", pipeline_id) in _RESERVED_NAME_PIPELINES
+
 
 def on_started():
     LOG.info('IntentService is starting up.')
@@ -293,6 +320,35 @@ def _emit_utterance_handled(self, dispatch_msg: Message):
         utterance."""
         self.bus.emit(dispatch_msg.forward(SpecMessage.UTTERANCE_HANDLED, {}))
 
+    @staticmethod
+    def _missing_required_slots(match: IntentHandlerMatch) -> List[str]:
+        """OVOS-PIPELINE-1 §6.2 orchestrator backstop for ``required_slots``.
+
+        After a plugin returns a Match, the orchestrator verifies the match's
+        slot map contains every slot the matched intent declares as required
+        (OVOS-INTENT-3 §5.3, OVOS-INTENT-4 §6.1). If any is absent, the
+        orchestrator treats the match as if the plugin had declined and
+        continues iteration — no bus event is emitted; the only observable
+        effect is a non-match (§6.2). The primary obligation to enforce
+        ``required_slots`` still lies with the engine during ``match()``; this
+        is a second line of defense against engine bugs.
+
+        The plugin surfaces the constraint in ``match.match_data``: the required
+        slot names under ``__required_slots__`` and the captured slot map as the
+        remaining keys (OVOS-INTENT-3 §7; ``Match.slots`` in PIPELINE-1 §4.3).
+        When a plugin does not surface it (the common case today) this returns
+        an empty list and the backstop is a no-op, leaving engine-side
+        enforcement authoritative — fully backward compatible.
+
+        Returns:
+            List[str]: required slot names absent from the match's slot map.
+        """
+        match_data = match.match_data or {}
+        required_slots = match_data.get("__required_slots__")
+        if not required_slots:
+            return []
+        return [slot for slot in required_slots if not match_data.get(slot)]
+
     def _dispatch_match(self, match: IntentHandlerMatch, message: Message, lang: str,
                         pipeline_id: str = None) -> None:
         """Orchestrate the OVOS-PIPELINE-1 §6.1 post-match steps, then dispatch.
@@ -349,7 +405,15 @@ def _dispatch_match(self, match: IntentHandlerMatch, message: Message, lang: str
 
                 was_deactivated = match.skill_id in self._deactivations[sess.session_id]
                 if not was_deactivated:
-                    sess.activate_skill(match.skill_id)
+                    # 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/
+                    # fallback/common_query): a reserved name is a continuation
+                    # or termination of an already-active skill's participation,
+                    # not a fresh activation. `activate_skill` is a back-compat
+                    # shim over `add_active_handler` (§7.1) in current bus-client.
+                    if not _produces_reserved_name(pipeline_id):
+                        sess.activate_skill(match.skill_id)
                     # emit event for skills callback -> self.handle_activate
                     self.bus.emit(reply.forward(f"{match.skill_id}.activate"))
 
@@ -520,6 +584,14 @@ def handle_utterance(self, message: Message):
                             LOG.debug(
                                 f"ignoring match, intent '{match.match_type}' blacklisted by Session '{sess.session_id}'")
                             continue
+                        # OVOS-PIPELINE-1 §6.2: if the matched intent is missing
+                        # any required slot, treat it as if the plugin had
+                        # declined and continue iteration; no bus event is emitted.
+                        missing = self._missing_required_slots(match)
+                        if missing:
+                            LOG.debug(f"ignoring match '{match.match_type}': "
+                                      f"missing required slots {missing} (§6.2)")
+                            continue
                         try:
                             self._dispatch_match(match, message, intent_lang,
                                                      pipeline_id=pipeline)

test/unittests/test_intent_service_extended.py

@@ -602,5 +602,79 @@ def test_shutdown_calls_pipeline_stop_and_shutdown(self):
         pipeline.shutdown.assert_called_once()
 
 
+# ---------------------------------------------------------------------------
+# OVOS-PIPELINE-1 §6.2 required_slots backstop
+# ---------------------------------------------------------------------------
+
+class TestRequiredSlotsBackstop(unittest.TestCase):
+
+    def test_no_required_slots_is_noop(self):
+        m = _make_match()
+        m.match_data = {"skill_id": "s"}
+        self.assertEqual(IntentService._missing_required_slots(m), [])
+
+    def test_all_required_slots_present(self):
+        m = _make_match()
+        m.match_data = {"__required_slots__": ["room"], "room": "kitchen"}
+        self.assertEqual(IntentService._missing_required_slots(m), [])
+
+    def test_missing_required_slot_reported(self):
+        m = _make_match()
+        m.match_data = {"__required_slots__": ["room", "device"], "room": "kitchen"}
+        self.assertEqual(IntentService._missing_required_slots(m), ["device"])
+
+    def test_falsy_slot_counts_as_missing(self):
+        m = _make_match()
+        m.match_data = {"__required_slots__": ["room"], "room": ""}
+        self.assertEqual(IntentService._missing_required_slots(m), ["room"])
+
+
+# ---------------------------------------------------------------------------
+# OVOS-PIPELINE-1 §7.1/§7.3 active-handler push + reserved-name suppression
+# ---------------------------------------------------------------------------
+
+class TestReservedNameActivation(unittest.TestCase):
+
+    def _dispatch(self, pipeline_id):
+        svc = _make_service()
+        sess = Session("s1")
+        msg = Message("recognizer_loop:utterance", {"utterances": ["hi"]},
+                      {"session": sess.serialize()})
+        match = _make_match(match_type="test.skill:intent",
+                            skill_id="test.skill", session=sess)
+        svc._dispatch_match(match, msg, "en-US", pipeline_id=pipeline_id)
+        return sess
+
+    def test_regular_pipeline_pushes_active_handler(self):
+        sess = self._dispatch("ovos-adapt-pipeline-plugin-high")
+        ids = [h.get("skill_id") if isinstance(h, dict) else getattr(h, "skill_id", h)
+               for h in sess.active_handlers]
+        self.assertIn("test.skill", ids)
+
+    def test_reserved_name_pipeline_suppresses_push(self):
+        # §7.3: converse/stop/fallback/common_query dispatches must NOT push
+        for pid in ("ovos-converse-pipeline-plugin",
+                    "ovos-stop-pipeline-plugin-high",
+                    "ovos-fallback-pipeline-plugin-medium",
+                    "ovos-common-query-pipeline-plugin"):
+            sess = self._dispatch(pid)
+            ids = [h.get("skill_id") if isinstance(h, dict) else getattr(h, "skill_id", h)
+                   for h in sess.active_handlers]
+            self.assertNotIn("test.skill", ids, f"{pid} should suppress the push")
+
+
+class TestProducesReservedName(unittest.TestCase):
+
+    def test_reserved_roles_true_with_confidence_suffix(self):
+        from ovos_core.intent_services.service import _produces_reserved_name
+        self.assertTrue(_produces_reserved_name("ovos-stop-pipeline-plugin-high"))
+        self.assertTrue(_produces_reserved_name("ovos-converse-pipeline-plugin"))
+
+    def test_regular_role_false(self):
+        from ovos_core.intent_services.service import _produces_reserved_name
+        self.assertFalse(_produces_reserved_name("ovos-adapt-pipeline-plugin-high"))
+        self.assertFalse(_produces_reserved_name(None))
+
+
 if __name__ == "__main__":
     unittest.main()