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

CHANGELOG.md

@@ -1,5 +1,14 @@
 # Changelog
 
+## [2.2.4a3](https://github.com/OpenVoiceOS/ovos-core/tree/2.2.4a3) (2026-06-28)
+
+[Full Changelog](https://github.com/OpenVoiceOS/ovos-core/compare/2.2.4a1...2.2.4a3)
+
+**Merged pull requests:**
+
+- test\(e2e\): fix transient test\_fallback\_match meta mismatch on dev [\#792](https://github.com/OpenVoiceOS/ovos-core/pull/792) ([JarbasAl](https://github.com/JarbasAl))
+- ci: wire shared opm-check workflow \(opm.pipeline entry-points\) [\#787](https://github.com/OpenVoiceOS/ovos-core/pull/787) ([JarbasAl](https://github.com/JarbasAl))
+
 ## [2.2.4a1](https://github.com/OpenVoiceOS/ovos-core/tree/2.2.4a1) (2026-06-28)
 
 [Full Changelog](https://github.com/OpenVoiceOS/ovos-core/compare/2.2.3a1...2.2.4a1)

ovos_core/intent_services/converse_service.py

@@ -1,8 +1,9 @@
 import time
-from threading import Event
+from threading import Event, Lock, Timer
 from typing import Optional, Dict, List, Union
 
 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, Session
 from ovos_config.config import Configuration
@@ -14,12 +15,20 @@
 from ovos_plugin_manager.templates.pipeline import PipelinePlugin, IntentHandlerMatch
 from ovos_workshop.permissions import ConverseMode, ConverseActivationMode
 
+#: upper bound, seconds, on how long core waits for a skill's
+#: ``skill.converse.response`` before the dispatch lifecycle is declared a
+#: timeout (``mycroft.skill.handler.error``). Generous: converse handlers may
+#: legitimately run a while, but this must eventually backstop a silent skill so
+#: an orchestrator observing the done-signal never hangs on the in-flight
+#: dispatch.
+CONVERSE_HANDLER_TIMEOUT = 5 * 60
+
 
 class ConverseService(PipelinePlugin):
     """Intent Service handling conversational skills."""
 
     def __init__(self, bus: Optional[Union[MessageBusClient, FakeBus]] = None,
-                 config: Optional[Dict] = None):
+                 config: Optional[Dict] = None) -> None:
         config = config or Configuration().get("skills", {}).get("converse", {})
         super().__init__(bus, config)
         self._consecutive_activations = {}
@@ -31,7 +40,54 @@ def __init__(self, bus: Optional[Union[MessageBusClient, FakeBus]] = None,
         self.bus.on("converse:skill", self.handle_converse)
 
     def handle_converse(self, message: Message):
+        """Priority-based skill activation and deactivation. Tracks active skills per session, handles converse requests, and manages lifecycle events."""
         skill_id = message.data["skill_id"]
+        # the dispatch belongs to this session; only an ack carrying the same
+        # session may resolve it (a concurrent converse dispatch to the same
+        # skill in another session must not cross-resolve).
+        session_id = SessionManager.get(message).session_id
+
+        lifecycle = HandlerLifecycle(self.bus, message, skill_id=skill_id,
+                                     handler_name=f"{skill_id}.converse")
+
+        resolved = Event()
+        resolve_lock = Lock()
+
+        def _claim() -> bool:
+            with resolve_lock:
+                if resolved.is_set():
+                    return False
+                resolved.set()
+                return True
+
+        def _resolve_complete(msg: Message) -> None:
+            if msg.data.get("skill_id") and msg.data.get("skill_id") != skill_id:
+                return  # ack from a different skill, ignore
+            if "session" in msg.context and \
+                    SessionManager.get(msg).session_id != session_id:
+                return  # ack from a different session, ignore
+            if not _claim():
+                return
+            timer.cancel()
+            self.bus.remove("skill.converse.response", _resolve_complete)
+            lifecycle.complete()
+
+        def _resolve_timeout() -> None:
+            if not _claim():
+                return
+            self.bus.remove("skill.converse.response", _resolve_complete)
+            LOG.warning(f"converse dispatch to {skill_id} timed out after "
+                        f"{CONVERSE_HANDLER_TIMEOUT}s; emitting handler error")
+            lifecycle.error(TimeoutError(
+                f"converse handler timed out after {CONVERSE_HANDLER_TIMEOUT} seconds"))
+
+        timer = Timer(CONVERSE_HANDLER_TIMEOUT, _resolve_timeout)
+        timer.daemon = True
+
+        self.bus.on("skill.converse.response", _resolve_complete)
+        timer.start()
+        # mycroft.skill.handler.start, then the dispatch itself
+        lifecycle.start()
         self.bus.emit(message.reply(f"{skill_id}.converse.request", message.data))
 
     @property
@@ -58,7 +114,7 @@ def get_active_skills(message: Optional[Message] = None) -> List[str]:
         return [skill[0] for skill in session.active_skills]
 
     def deactivate_skill(self, skill_id: str, source_skill: Optional[str] = None,
-                         message: Optional[Message] = None):
+                         message: Optional[Message] = None) -> None:
         """Remove a skill from being targetable by converse.
 
         Args:
@@ -68,13 +124,13 @@ def deactivate_skill(self, skill_id: str, source_skill: Optional[str] = None,
         """
         source_skill = source_skill or skill_id
         if self._deactivate_allowed(skill_id, source_skill):
+            message = message or Message("")
             session = SessionManager.get(message)
             if session.is_active(skill_id):
                 # update converse session
                 session.deactivate_skill(skill_id)
 
                 # keep message.context
-                message = message or Message("")
                 message.context["session"] = session.serialize()  # update session active skills
                 # send bus event
                 self.bus.emit(
@@ -97,12 +153,12 @@ def activate_skill(self, skill_id: str, source_skill: Optional[str] = None,
         """
         source_skill = source_skill or skill_id
         if self._activate_allowed(skill_id, source_skill):
+            message = message or Message("")
             # update converse session
             session = SessionManager.get(message)
             session.activate_skill(skill_id)
 
             # keep message.context
-            message = message or Message("")
             message.context["session"] = session.serialize()  # update session active skills
             message = message.forward("intent.service.skills.activated",
                                       {"skill_id": skill_id})
@@ -159,7 +215,7 @@ def _activate_allowed(self, skill_id: str, source_skill: Optional[str] = None) -
         default_max = self.config.get("max_activations", -1)
         # per skill override limit of consecutive activations
         skill_max = self.config.get("skill_activations", {}).get(skill_id)
-        max_activations = skill_max or default_max
+        max_activations = skill_max if skill_max is not None else default_max
         if skill_id not in self._consecutive_activations:
             self._consecutive_activations[skill_id] = 0
         if max_activations < 0:
@@ -386,7 +442,7 @@ def handle_get_active_skills(self, message: Message):
         self.bus.emit(message.reply("intent.service.active_skills.reply",
                                     {"skills": self.get_active_skills(message)}))
 
-    def shutdown(self):
+    def shutdown(self) -> None:
         self.bus.remove("converse:skill", self.handle_converse)
         self.bus.remove('intent.service.skills.deactivate', self.handle_deactivate_skill_request)
         self.bus.remove('intent.service.skills.activate', self.handle_activate_skill_request)

ovos_core/intent_services/dispatcher.py

@@ -0,0 +1,220 @@
+# Copyright 2017 Mycroft AI Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+"""§7/§8 handler-lifecycle trio — dispatcher.
+
+Emits ``ovos.intent.handler.start`` before each ``<skill_id>:<intent_name>``
+dispatch and exactly one terminal (``complete``/``error``/timeout) after. The
+framework done-signal (``mycroft.skill.handler.complete``/``.error``) is consumed
+as the completion hint. A §8.3 timeout backstops every dispatch. The §9.5
+``ovos.utterance.handled`` end-marker is NOT this class's concern — the
+orchestrator's ``on_terminal`` callback is invoked after each §8 terminal.
+"""
+import threading
+from typing import Callable, Dict, List, Optional
+
+from ovos_bus_client.message import Message
+from ovos_spec_tools import SpecMessage
+from ovos_utils.log import LOG
+
+#: default upper bound on handler execution before §8.3 timeout fires, seconds.
+#: handlers are long-running by design (§6.5) so this is generous; set to 0 or a
+#: negative value (config ``intents.handler_timeout``) to disable the timer.
+DEFAULT_HANDLER_TIMEOUT = 5 * 60
+
+
+class _InFlightDispatch:
+    """A dispatch awaiting its §8 terminal."""
+
+    __slots__ = ("skill_id", "intent_name", "dispatch_msg", "timer", "resolved")
+
+    def __init__(self, skill_id: str, intent_name: str, dispatch_msg: Message):
+        self.skill_id = skill_id
+        self.intent_name = intent_name
+        self.dispatch_msg = dispatch_msg
+        self.timer: Optional[threading.Timer] = None
+        self.resolved = False
+
+
+class IntentDispatcher:
+    """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,
+                 on_terminal: Optional[Callable[[Message], None]] = None):
+        self.bus = bus
+        self.timeout = timeout
+        # Called synchronously with the dispatch Message immediately AFTER each §8
+        # terminal (complete/error/timeout) is emitted, so the orchestrator can emit
+        # its §9.5 ovos.utterance.handled end-marker. Doing this in the same step
+        # (rather than via a separate bus subscription) guarantees the terminal is
+        # observed before the end-marker — otherwise a consumer subscribed to the
+        # terminal could emit the end-marker before the terminal is recorded.
+        self.on_terminal = on_terminal
+        # session_id -> stack of _InFlightDispatch (LIFO for nested lifecycles)
+        self._in_flight: Dict[str, List[_InFlightDispatch]] = {}
+        self._lock = threading.Lock()
+        # framework done-signals (legacy namespace; do NOT bridge to the spec trio)
+        self.bus.on("mycroft.skill.handler.complete", self._on_skill_complete)
+        self.bus.on("mycroft.skill.handler.error", self._on_skill_error)
+
+    def shutdown(self):
+        try:
+            self.bus.remove("mycroft.skill.handler.complete", self._on_skill_complete)
+            self.bus.remove("mycroft.skill.handler.error", self._on_skill_error)
+        except Exception:
+            LOG.exception("failed to remove done-signal handlers during shutdown")
+        with self._lock:
+            for stack in self._in_flight.values():
+                for entry in stack:
+                    entry.resolved = True
+                    if entry.timer is not None:
+                        entry.timer.cancel()
+            self._in_flight.clear()
+
+    # -- public API ------------------------------------------------------
+    def dispatch(self, dispatch_msg: Message,
+                 skill_id: Optional[str] = None,
+                 intent_name: Optional[str] = None):
+        """Dispatch a matched intent and own its §8 handler-lifecycle trio.
+
+        Emits ``ovos.intent.handler.start`` (§8.1), the dispatch on
+        ``<skill_id>:<intent_name>`` (§7), then exactly one terminal
+        (``complete``/``error``/timeout) once the handler reports. The dispatch goes
+        out asynchronously — this call does NOT block. The orchestrator reacts to the
+        §8 terminal to emit its §9.5 ``ovos.utterance.handled`` end-marker.
+
+        ``skill_id``/``intent_name`` default to the two halves of the dispatch
+        topic; the orchestrator passes them explicitly from its own ``Match`` so
+        they never come from the skill.
+        """
+        topic = dispatch_msg.msg_type
+        if skill_id is None:
+            skill_id = topic.split(":", 1)[0]
+        if intent_name is None:
+            intent_name = topic.split(":", 1)[-1]
+
+        entry = _InFlightDispatch(skill_id, intent_name, dispatch_msg)
+        sid = self._session_id(dispatch_msg)
+        with self._lock:
+            self._in_flight.setdefault(sid, []).append(entry)
+            if self.timeout and self.timeout > 0:
+                entry.timer = threading.Timer(self.timeout, self._on_timeout,
+                                              args=(sid, entry))
+                entry.timer.daemon = True
+                entry.timer.start()
+
+        # §8.1: start immediately before invoking (dispatching) the handler
+        self._emit(SpecMessage.INTENT_HANDLER_START, dispatch_msg,
+                   {"skill_id": skill_id, "intent_name": intent_name})
+        # §7: the dispatch itself
+        self.bus.emit(dispatch_msg)
+
+    # -- emission helpers ------------------------------------------------
+    @staticmethod
+    def _session_id(message: Message) -> str:
+        return (message.context.get("session") or {}).get("session_id", "default")
+
+    def _emit(self, topic, dispatch_msg: Message, data: dict):
+        """Emit a Message forwarded from the dispatch (§6.1 / §8 — context, incl.
+        session, preserved unchanged via MSG-1 §5.1 ``forward``)."""
+        self.bus.emit(dispatch_msg.forward(topic, data))
+
+    def _notify_terminal(self, dispatch_msg: Message):
+        """Tell the orchestrator a §8 terminal just fired so it can emit its §9.5
+        end-marker. Called after the terminal is on the bus, so the terminal is
+        always observed before the end-marker."""
+        if self.on_terminal is not None:
+            self.on_terminal(dispatch_msg)
+
+    # -- terminal resolution ---------------------------------------------
+    def _pop(self, sid: str, skill_id: Optional[str]) -> Optional[_InFlightDispatch]:
+        """Pop the most-recent unresolved in-flight dispatch for this session
+        whose ``skill_id`` matches (when known). LIFO so nested lifecycles
+        resolve innermost-first (§6.5)."""
+        with self._lock:
+            stack = self._in_flight.get(sid)
+            if not stack:
+                return None
+            for i in range(len(stack) - 1, -1, -1):
+                entry = stack[i]
+                if entry.resolved:
+                    continue
+                if skill_id and entry.skill_id != skill_id:
+                    continue
+                entry.resolved = True
+                stack.pop(i)
+                if not stack:
+                    self._in_flight.pop(sid, None)
+                if entry.timer is not None:
+                    entry.timer.cancel()
+                return entry
+            return None
+
+    def _on_skill_complete(self, message: Message):
+        """Framework done-signal -> ``complete`` (§8.1)."""
+        entry = self._pop(self._session_id(message), message.context.get("skill_id"))
+        if entry is None:
+            return
+        try:
+            self._emit(SpecMessage.INTENT_HANDLER_COMPLETE, entry.dispatch_msg,
+                       {"skill_id": entry.skill_id, "intent_name": entry.intent_name})
+        finally:
+            self._notify_terminal(entry.dispatch_msg)
+
+    def _on_skill_error(self, message: Message):
+        """Framework done-signal -> ``error`` with the exception (§8.2)."""
+        entry = self._pop(self._session_id(message), message.context.get("skill_id"))
+        if entry is None:
+            return
+        exception = (message.data.get("exception")
+                     or message.data.get("error")
+                     or "handler raised an exception")
+        try:
+            self._emit(SpecMessage.INTENT_HANDLER_ERROR, entry.dispatch_msg,
+                       {"skill_id": entry.skill_id,
+                        "intent_name": entry.intent_name,
+                        "exception": str(exception)})
+        finally:
+            self._notify_terminal(entry.dispatch_msg)
+
+    def _on_timeout(self, sid: str, entry: _InFlightDispatch):
+        """§8.3 — bound handler execution; on timeout emit ``error``."""
+        with self._lock:
+            if entry.resolved:
+                return
+            entry.resolved = True
+            entry.timer = None
+            stack = self._in_flight.get(sid)
+            if stack and entry in stack:
+                stack.remove(entry)
+                if not stack:
+                    self._in_flight.pop(sid, None)
+        LOG.warning(f"handler timeout for {entry.skill_id}:{entry.intent_name} "
+                    f"after {self.timeout}s; emitting ovos.intent.handler.error")
+        try:
+            self._emit(SpecMessage.INTENT_HANDLER_ERROR, entry.dispatch_msg,
+                       {"skill_id": entry.skill_id,
+                        "intent_name": entry.intent_name,
+                        "exception": f"handler timed out after {self.timeout} seconds"})
+        finally:
+            self._notify_terminal(entry.dispatch_msg)