feat: orchestrator owns the PIPELINE-1 §9 utterance-terminal events

Complete the orchestrator's ownership of the OVOS-PIPELINE-1 §6.1
per-utterance terminal sequence, on top of the §8 handler-lifecycle
trio:

- §9.2 ovos.intent.matched — emitted by _dispatch_match on every
  accepted match, before the dispatch goes out (notification, not a
  dispatch). Carries skill_id, intent_name (the full
  <skill_id>:<intent_name> match_type), lang, utterance, slots,
  pipeline_id.
- §9.3 ovos.intent.unmatched — the no-match / all-filtered terminal,
  replacing the legacy complete_intent_failure (the two are bridged by
  ovos-spec-tools' MIGRATION_MAP, so emitting the spec topic re-delivers
  the legacy one to consumers still on it).
- §6.4 cancellation now emits the spec ovos.utterance.cancelled.

Each utterance terminates with exactly one ovos.utterance.handled
(§9.5): core owns it on the no-match, cancel and §8.3-timeout paths;
on the ordinary matched path the skill framework still emits it (moving
that fully into core is gated on the ovos-workshop reduction).

Rename _emit_match_message -> _dispatch_match (it orchestrates the §6.1
post-match steps then dispatches) and correct the IntentDispatcher
docstring to scope it to §7 dispatch + §8 trio (the §9.2 notification
lives in the service).

Verified on a real minicroft: matched path emits matched/start/
complete/handled exactly once each; no-match path emits
ovos.intent.unmatched + ovos.utterance.handled (no complete_intent_
failure). test_no_skills / test_lang_detect conformance suites green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: JarbasAl

ovos_core/intent_services/dispatcher.py

@@ -47,12 +47,16 @@
 - ``mycroft.skill.handler.error`` (carrying a human-readable error) → the
   orchestrator emits ``error`` with the reported ``exception`` (§8.2).
 
-These are **legacy-namespace** topics and, with the OVOS-MSG-1 trio bridge
-removed from ovos-spec-tools, they do **not** bridge to the spec trio — so the
-orchestrator's own spec emissions never echo back as a done-signal and no
-echo-guard is needed. The framework done-signal and the spec trio live in
-separate namespaces: workshop owns the legacy one, the orchestrator owns the
-spec one.
+These are **legacy-namespace** topics. **Hard dependency:** the ovos-spec-tools
+MIGRATION_MAP trio bridge (``mycroft.skill.handler.* ↔ ovos.intent.handler.*``)
+MUST be removed so the orchestrator's own spec emissions do not bridge back to a
+legacy done-signal. Until that lands the bridge is still active, but the
+resolved-guard in :meth:`_pop` keeps the terminal count at exactly one even if a
+bridged echo arrives (it claims an already-resolved entry and returns ``None``);
+the ``"message"``-aggregate consumers (the ovoscope harness) also never see the
+bridged counterpart. Once the bridge is removed, the framework done-signal and the
+spec trio live cleanly in separate namespaces: workshop owns the legacy one, the
+orchestrator owns the spec one.
 
 The §8.3 timeout backstops every dispatch so exactly one terminal is guaranteed
 even if no done-signal ever arrives.
@@ -95,11 +99,15 @@ def __init__(self, skill_id: str, intent_name: str, dispatch_msg: Message):
 
 
 class IntentDispatcher:
-    """Owns the PIPELINE-1 §6.1 matched-path sequence (§9.2 notification, §7
-    dispatch, §8 handler-lifecycle trio).
-
-    Owned by ``IntentService``; wires its own bus observers for the framework
-    done-signals. The orchestrator hands it a dispatch Message via :meth:`dispatch`.
+    """Owns the PIPELINE-1 §7 dispatch + §8 handler-lifecycle trio.
+
+    Emits ``ovos.intent.handler.start`` before the ``<skill_id>:<intent_name>``
+    dispatch and exactly one terminal (``complete``/``error``/timeout) after. The
+    surrounding §6.1 orchestration — the §9.2 ``ovos.intent.matched`` notification,
+    skill activation, session update — lives in
+    ``IntentService._dispatch_match``, which hands a built dispatch Message to
+    :meth:`dispatch`. This class wires its own bus observers for the framework
+    done-signals (``mycroft.skill.handler.complete``/``.error``).
     """
 
     def __init__(self, bus, timeout: Optional[float] = DEFAULT_HANDLER_TIMEOUT):

ovos_core/intent_services/service.py

@@ -122,11 +122,12 @@ def __init__(self, bus, config=None, preload_pipelines=True,
         self.metadata_plugins = MetadataTransformersService(bus)
         self.intent_plugins = IntentTransformersService(bus)
 
-        # OVOS-PIPELINE-1 §6.1: the orchestrator owns the post-match terminal
-        # sequence — ovos.intent.matched (§9.2), dispatch (§7) and the §8
-        # handler-lifecycle trio. ``handler_timeout`` (seconds, §8.3) bounds
-        # handler execution so exactly one terminal is guaranteed even if a
-        # handler never reports; 0/None disables the timer.
+        # OVOS-PIPELINE-1 §7/§8: the dispatcher owns the dispatch on
+        # <skill_id>:<intent_name> (§7) and the handler-lifecycle trio (§8). The
+        # surrounding §6.1 orchestration (§9.2 ovos.intent.matched, skill
+        # activation, session update) lives in _dispatch_match. ``handler_timeout``
+        # (seconds, §8.3) bounds handler execution so exactly one terminal is
+        # guaranteed even if a handler never reports; 0/None disables the timer.
         handler_timeout = self.config.get("handler_timeout", DEFAULT_HANDLER_TIMEOUT)
         self.intent_dispatcher = IntentDispatcher(bus, timeout=handler_timeout)
 
@@ -281,31 +282,24 @@ def _handle_deactivate(self, message):
         skill_id = message.data.get("skill_id")
         self._deactivations[sess.session_id].append(skill_id)
 
-    def _emit_match_message(self, match: IntentHandlerMatch, message: Message, lang: str,
-                            pipeline_id: str = None):
-        """
-        Emit a reply message for a matched intent, updating session and skill activation.
+    def _dispatch_match(self, match: IntentHandlerMatch, message: Message, lang: str,
+                        pipeline_id: str = None):
+        """Orchestrate the OVOS-PIPELINE-1 §6.1 post-match steps, then dispatch.
 
-        This method processes matched intents from either a pipeline matcher or an intent handler,
-        creating a reply message with matched intent details and managing skill activation.
+        Runs the service-state-dependent post-match orchestration — the
+        intent-transformer chain (TRANSFORM-1 §3.4), skill activation +
+        ``{skill_id}.activate``, session update, and ``context['pipeline_id']``
+        stamping (§7.1) — builds the dispatch Message, emits the §9.2
+        ``ovos.intent.matched`` notification, and hands the dispatch Message to
+        the IntentDispatcher, which owns the §7 dispatch + §8 handler-lifecycle
+        trio.
 
         Args:
-            match (IntentHandlerMatch): The matched intent object containing
-                utterance and matching information.
-            message (Message): The original messagebus message that triggered the intent match.
-            lang (str): The language of the pipeline plugin match
-
-        Details:
-            - Handles two types of matches: PipelineMatch and IntentHandlerMatch
-            - Creates a reply message with matched intent data
-            - Activates the corresponding skill if not previously deactivated
-            - Updates session information
-            - Emits the reply message on the messagebus
-
-        Side Effects:
-            - Modifies session state
-            - Emits a messagebus event
-            - Can trigger skill activation events
+            match (IntentHandlerMatch): The matched intent (utterance, match_type,
+                skill_id, match_data, optional updated_session).
+            message (Message): The originating utterance Message to derive from.
+            lang (str): The content language of the match.
+            pipeline_id (str): The pipeline plugin that produced the match (§3.1).
 
         Returns:
             None
@@ -358,6 +352,18 @@ def _emit_match_message(self, match: IntentHandlerMatch, message: Message, lang:
             if pipeline_id:
                 reply.context["pipeline_id"] = pipeline_id
 
+            # OVOS-PIPELINE-1 §9.2: broadcast ovos.intent.matched BEFORE the
+            # dispatch goes out. A notification, not a dispatch: consumers MUST NOT
+            # treat receipt as permission to run a handler.
+            self.bus.emit(reply.forward(SpecMessage.INTENT_MATCHED, {
+                "skill_id": match.skill_id,
+                "intent_name": match.match_type,
+                "lang": lang,
+                "utterance": match.utterance,
+                "slots": dict(match.match_data or {}),
+                "pipeline_id": reply.context.get("pipeline_id"),
+            }))
+
             # OVOS-PIPELINE-1 §7 dispatch + §8 handler-lifecycle trio: hand the
             # dispatch Message to the IntentDispatcher, which emits
             # ovos.intent.handler.start (§8.1) before the dispatch and the matching
@@ -429,8 +435,9 @@ def send_cancel_event(self, message):
         sound = Configuration().get('sounds', {}).get('cancel', "snd/cancel.mp3")
         # NOTE: message.reply to ensure correct message destination
         self.bus.emit(message.reply('mycroft.audio.play_sound', {"uri": sound}))
-        self.bus.emit(message.reply("ovos.utterance.cancelled"))
-        self.bus.emit(message.reply("ovos.utterance.handled"))
+        # OVOS-PIPELINE-1 §6.4 cancellation terminal path: cancelled -> handled
+        self.bus.emit(message.reply(SpecMessage.UTTERANCE_CANCELLED))
+        self.bus.emit(message.reply(SpecMessage.UTTERANCE_HANDLED))
 
     def handle_utterance(self, message: Message):
         """Main entrypoint for handling user utterances
@@ -501,7 +508,7 @@ def handle_utterance(self, message: Message):
                                 f"ignoring match, intent '{match.match_type}' blacklisted by Session '{sess.session_id}'")
                             continue
                         try:
-                            self._emit_match_message(match, message, intent_lang,
+                            self._dispatch_match(match, message, intent_lang,
                                                      pipeline_id=pipeline)
                             break
                         except Exception:
@@ -526,16 +533,28 @@ def handle_utterance(self, message: Message):
         return match, message.context, stopwatch
 
     def send_complete_intent_failure(self, message):
-        """Send a message that no skill could handle the utterance.
+        """Emit the OVOS-PIPELINE-1 §9.3 no-match terminal.
+
+        The orchestrator owns the no-match branch of the §6.1 lifecycle: it plays
+        the error sound, emits ``ovos.intent.unmatched`` (§9.3 — the intent-layer
+        failure signal) and then the universal end-marker ``ovos.utterance.handled``
+        (§9.5). Exactly one ``ovos.utterance.handled`` terminates the utterance.
+
+        ``ovos.intent.unmatched`` is the spec replacement for the legacy
+        ``complete_intent_failure``; the two are bridged by ovos-spec-tools'
+        MIGRATION_MAP, so emitting the spec topic re-delivers the legacy one to
+        any consumer still subscribed to it.
 
         Args:
             message (Message): original message to forward from
         """
         sound = Configuration().get('sounds', {}).get('error', "snd/error.mp3")
         # NOTE: message.reply to ensure correct message destination
         self.bus.emit(message.reply('mycroft.audio.play_sound', {"uri": sound}))
-        self.bus.emit(message.reply('complete_intent_failure', message.data))
-        self.bus.emit(message.reply("ovos.utterance.handled"))
+        # §9.3: intent-layer failure signal (carries lang from message.data)
+        self.bus.emit(message.reply(SpecMessage.INTENT_UNMATCHED, message.data))
+        # §9.5: universal end-marker
+        self.bus.emit(message.reply(SpecMessage.UTTERANCE_HANDLED))
 
     @staticmethod
     def handle_add_context(message: Message):

test/end2end/test_adapt.py

@@ -27,11 +27,16 @@
 LEGACY_UTTERANCE = migration_counterpart(SPEC_UTTERANCE)
 SPEC_SPEAK = SpecMessage.SPEAK.value
 UTTERANCE_HANDLED = SpecMessage.UTTERANCE_HANDLED.value
-# PIPELINE-1 §8 handler-lifecycle trio, emitted by the orchestrator (core)
-# wrapping the dispatch: start before, complete on the framework done-signal. The
-# skill's own ovos.utterance.handled (§9.5) is untouched by this change.
+# PIPELINE-1 orchestrator-emitted matched-path messages: §9.2 ovos.intent.matched
+# (before dispatch) and the §8 handler-lifecycle trio (start before dispatch,
+# complete on the framework done-signal). The skill's own ovos.utterance.handled
+# (§9.5) is left to ovos-workshop on this matched path.
+INTENT_MATCHED = SpecMessage.INTENT_MATCHED.value
 HANDLER_START = SpecMessage.INTENT_HANDLER_START.value
 HANDLER_COMPLETE = SpecMessage.INTENT_HANDLER_COMPLETE.value
+# PIPELINE-1 §9.3: the no-match / all-filtered terminal is ovos.intent.unmatched
+# (the spec replacement for the legacy complete_intent_failure).
+INTENT_UNMATCHED = SpecMessage.INTENT_UNMATCHED.value
 
 NAMESPACE_PATHS = {
     "spec": (False, False, SPEC_UTTERANCE),
@@ -76,6 +81,14 @@ def _run_adapt_match(self, namespace):
                 Message(f"{self.skill_id}.activate",
                         data={},
                         context={"skill_id": self.skill_id}),
+                # PIPELINE-1 §9.2: matched notification, before the dispatch.
+                # intent_name carries the full <skill_id>:<intent_name> match_type.
+                Message(INTENT_MATCHED,
+                        data={"skill_id": self.skill_id,
+                              "intent_name": f"{self.skill_id}:HelloWorldIntent",
+                              "utterance": "hello world",
+                              "lang": session.lang},
+                        context={"skill_id": self.skill_id}),
                 # PIPELINE-1 §8.1: orchestrator emits start immediately before dispatch
                 Message(HANDLER_START,
                         data={"skill_id": self.skill_id,
@@ -142,7 +155,7 @@ def _run_skill_blacklist(self, namespace):
             expected_messages=[
                 message,
                 Message("mycroft.audio.play_sound", {"uri": "snd/error.mp3"}),
-                Message("complete_intent_failure", {}),
+                Message(INTENT_UNMATCHED, {}),
                 Message(UTTERANCE_HANDLED, {})
             ]
         )
@@ -178,7 +191,7 @@ def _run_intent_blacklist(self, namespace):
             expected_messages=[
                 message,
                 Message("mycroft.audio.play_sound", {"uri": "snd/error.mp3"}),
-                Message("complete_intent_failure", {}),
+                Message(INTENT_UNMATCHED, {}),
                 Message(UTTERANCE_HANDLED, {})
             ]
         )
@@ -213,7 +226,7 @@ def _run_padatious_no_match(self, namespace):
             expected_messages=[
                 message,
                 Message("mycroft.audio.play_sound", {"uri": "snd/error.mp3"}),
-                Message("complete_intent_failure", {}),
+                Message(INTENT_UNMATCHED, {}),
                 Message(UTTERANCE_HANDLED, {})
             ]
         )

test/end2end/test_converse.py

@@ -25,9 +25,15 @@
 LEGACY_UTTERANCE = migration_counterpart(SPEC_UTTERANCE)
 SPEC_SPEAK = SpecMessage.SPEAK.value
 UTTERANCE_HANDLED = SpecMessage.UTTERANCE_HANDLED.value
-# PIPELINE-1 §8 handler-lifecycle trio, emitted by the orchestrator (core).
-HANDLER_START = SpecMessage.INTENT_HANDLER_START.value
-HANDLER_COMPLETE = SpecMessage.INTENT_HANDLER_COMPLETE.value
+INTENT_MATCHED = SpecMessage.INTENT_MATCHED.value      # ovos.intent.matched (§9.2)
+# §8 handler-lifecycle trio wraps every dispatch; this suite asserts converse
+# routing, not the trio (covered by the adapt/padatious suites), so it is
+# filtered via ignore_messages below.
+HANDLER_TRIO = [SpecMessage.INTENT_HANDLER_START.value,
+                SpecMessage.INTENT_HANDLER_COMPLETE.value,
+                SpecMessage.INTENT_HANDLER_ERROR.value,
+                "ovos.skills.settings_changed"]  # keep ovoscope's default ignore
+INTENT_UNMATCHED = SpecMessage.INTENT_UNMATCHED.value  # ovos.intent.unmatched (§9.3)
 
 # key -> (modernize, emit_legacy, utterance_topic)
 NAMESPACE_PATHS = {
@@ -75,10 +81,9 @@ def _run_parrot_mode(self, namespace: str) -> None:
             Message(f"{self.skill_id}.activate",
                     data={},
                     context={"skill_id": self.skill_id}),
-            # PIPELINE-1 §8.1: orchestrator start before dispatch
-            Message(HANDLER_START,
+            Message(INTENT_MATCHED,
                     data={"skill_id": self.skill_id,
-                          "intent_name": "start_parrot.intent"},
+                          "intent_name": f"{self.skill_id}:start_parrot.intent"},
                     context={"skill_id": self.skill_id}),
             Message(f"{self.skill_id}:start_parrot.intent",
                     data={"utterance": "start parrot mode", "lang": session.lang},
@@ -97,11 +102,6 @@ def _run_parrot_mode(self, namespace: str) -> None:
             Message("mycroft.skill.handler.complete",
                     data={"name": "ParrotSkill.handle_start_parrot_intent"},
                     context={"skill_id": self.skill_id}),
-            # PIPELINE-1 §8.1: orchestrator complete before the end-marker
-            Message(HANDLER_COMPLETE,
-                    data={"skill_id": self.skill_id,
-                          "intent_name": "start_parrot.intent"},
-                    context={"skill_id": self.skill_id}),
             Message(UTTERANCE_HANDLED,
                     data={},
                     context={"skill_id": self.skill_id}),
@@ -117,10 +117,8 @@ def _run_parrot_mode(self, namespace: str) -> None:
             Message(f"{self.skill_id}.activate",
                     data={},
                     context={"skill_id": self.skill_id}),
-            # PIPELINE-1 §7.0/§8.1: a converse dispatch is a dispatch -> orchestrator
-            # emits start before it (intent_name is the reserved name "skill").
-            Message(HANDLER_START,
-                    data={"skill_id": self.skill_id, "intent_name": "skill"},
+            Message(INTENT_MATCHED,
+                    data={"skill_id": self.skill_id, "intent_name": "converse:skill"},
                     context={"skill_id": self.skill_id}),
             Message("converse:skill",
                     data={"utterances": ["echo test"], "lang": session.lang, "skill_id": self.skill_id},
@@ -155,9 +153,8 @@ def _run_parrot_mode(self, namespace: str) -> None:
                     data={},
                     context={"skill_id": self.skill_id}),
 
-            # PIPELINE-1 §7.0/§8.1: orchestrator start before the converse dispatch
-            Message(HANDLER_START,
-                    data={"skill_id": self.skill_id, "intent_name": "skill"},
+            Message(INTENT_MATCHED,
+                    data={"skill_id": self.skill_id, "intent_name": "converse:skill"},
                     context={"skill_id": self.skill_id}),
             Message("converse:skill",
                     data={"utterances": ["stop parrot"], "lang": session.lang, "skill_id": self.skill_id},
@@ -191,7 +188,7 @@ def _run_parrot_mode(self, namespace: str) -> None:
                     data={"can_handle": False, "skill_id": self.skill_id},
                     context={"skill_id": self.skill_id}),
             Message("mycroft.audio.play_sound", data={"uri": "snd/error.mp3"}),
-            Message("complete_intent_failure"),
+            Message(INTENT_UNMATCHED),
             Message(UTTERANCE_HANDLED)
         ]
 
@@ -207,6 +204,7 @@ def _run_parrot_mode(self, namespace: str) -> None:
             final_session=final_session,
             source_message=[message1, message2, message3, message4],
             expected_messages=expected1 + expected2 + expected3 + expected4,
+            ignore_messages=HANDLER_TRIO,
             activation_points=[f"{self.skill_id}:start_parrot.intent"],
             # messages internal to ovos-core, i.e. would not be sent to clients such as hivemind
             keep_original_src=[f"{self.skill_id}.converse.ping",