From 173f567b6a93049f14ab190a5e9789384d4d8c90 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Ga=C3=ABtan=20Trellu?= <gaetan.trellu@gmail.com>
Date: Mon, 3 Aug 2026 20:24:59 -0400
Subject: [PATCH 1/7] fix: scope fallback polling by skill and session

---
 ovos_core/intent_services/fallback_service.py | 180 +++++++++++++-----
 test/unittests/test_fallback_service.py       | 155 +++++++++++++--
 2 files changed, 275 insertions(+), 60 deletions(-)

diff --git a/ovos_core/intent_services/fallback_service.py b/ovos_core/intent_services/fallback_service.py
index dad6a4607ae..5d1ce802c83 100644
--- a/ovos_core/intent_services/fallback_service.py
+++ b/ovos_core/intent_services/fallback_service.py
@@ -15,6 +15,7 @@
 import operator
 import threading
 import time
+from _thread import LockType
 from collections import namedtuple
 from typing import Callable, Dict, List, Optional, Tuple, Union
 
@@ -41,10 +42,12 @@ def __init__(self, bus: Optional[Union[MessageBusClient, FakeBus]] = None,
         config = config if config is not None else Configuration().get("skills", {}).get("fallbacks", {})
         super().__init__(bus, config)
         self.registered_fallbacks: Dict[str, int] = {}  # skill_id: priority
+        self._registered_fallbacks_lock = threading.RLock()
+        self._fallback_session_locks: Dict[str, Tuple[LockType, int]] = {}
+        self._fallback_session_locks_lock = threading.Lock()
         # skill_id -> (start_handler, response_handler) wired for the
         # done-signal translation, so they can be removed on deregister
         self._lifecycle_handlers: Dict[str, Tuple[Callable, Callable]] = {}
-        self._fallback_response_event = threading.Event()
         self.bus.on("ovos.skills.fallback.register", self.handle_register_fallback)
         self.bus.on("ovos.skills.fallback.deregister", self.handle_deregister_fallback)
 
@@ -84,12 +87,13 @@ def handle_register_fallback(self, message: Message) -> None:
 
         # check if .conf is overriding the priority for this skill
         priority_overrides = self.config.get("fallback_priorities", {})
-        if skill_id in priority_overrides:
-            new_priority = priority_overrides.get(skill_id)
-            LOG.info(f"forcing {skill_id} fallback priority from {priority} to {new_priority}")
-            self.registered_fallbacks[skill_id] = new_priority
-        else:
-            self.registered_fallbacks[skill_id] = priority
+        with self._registered_fallbacks_lock:
+            if skill_id in priority_overrides:
+                new_priority = priority_overrides.get(skill_id)
+                LOG.info(f"forcing {skill_id} fallback priority from {priority} to {new_priority}")
+                self.registered_fallbacks[skill_id] = new_priority
+            else:
+                self.registered_fallbacks[skill_id] = priority
 
         # report this skill's fallback dispatch lifecycle as the framework
         # done-signal so an orchestrator can resolve it (no skill_id -> skip)
@@ -98,10 +102,36 @@ def handle_register_fallback(self, message: Message) -> None:
 
     def handle_deregister_fallback(self, message: Message) -> None:
         skill_id = message.data.get("skill_id")
-        if skill_id in self.registered_fallbacks:
-            self.registered_fallbacks.pop(skill_id)
+        with self._registered_fallbacks_lock:
+            if skill_id in self.registered_fallbacks:
+                self.registered_fallbacks.pop(skill_id)
         self._unwire_lifecycle(skill_id)
 
+    def _fallback_registry_snapshot(self) -> Dict[str, int]:
+        """Return a stable fallback registry view for one match operation."""
+        with self._registered_fallbacks_lock:
+            return dict(self.registered_fallbacks)
+
+    def _acquire_fallback_session_lock(self, session_id: str) -> LockType:
+        """Serialize overlapping fallback polls for the same bus session."""
+        with self._fallback_session_locks_lock:
+            lock, users = self._fallback_session_locks.get(
+                session_id, (threading.Lock(), 0))
+            self._fallback_session_locks[session_id] = (lock, users + 1)
+        lock.acquire()
+        return lock
+
+    def _release_fallback_session_lock(self, session_id: str,
+                                       lock: LockType) -> None:
+        lock.release()
+        with self._fallback_session_locks_lock:
+            current_lock, users = self._fallback_session_locks[session_id]
+            if users == 1:
+                self._fallback_session_locks.pop(session_id)
+            else:
+                self._fallback_session_locks[session_id] = (
+                    current_lock, users - 1)
+
     def _fallback_allowed(self, skill_id: str) -> bool:
         """Checks if a skill_id is allowed to fallback
 
@@ -131,47 +161,100 @@ def _collect_fallback_skills(self, message: Message,
         """
         if fb_range is None:
             fb_range = FallbackRange(0, 100)
-        skill_ids = []  # skill_ids that already answered to ping
-        fallback_skills = []  # skill_ids that want to handle fallback
-
         sess = SessionManager.get(message)
         if sess is None:
-            return fallback_skills
-        # filter skills outside the fallback_range
-        in_range = [s for s, p in self.registered_fallbacks.items()
-                    if fb_range.start < p <= fb_range.stop
-                    and s not in (sess.blacklisted_skills or [])]
-        skill_ids += [s for s in self.registered_fallbacks if s not in in_range]
-
-        def handle_ack(msg):
-            skill_id = msg.data["skill_id"]
-            if msg.data.get("can_handle", True):
-                if skill_id in in_range:
-                    fallback_skills.append(skill_id)
-                    LOG.info(f"{skill_id} will try to handle fallback")
-                else:
-                    LOG.debug(f"{skill_id} is out of range, skipping")
-            else:
-                LOG.debug(f"{skill_id} does NOT WANT to try to handle fallback")
-            skill_ids.append(skill_id)
-            self._fallback_response_event.set()
-
-        if in_range:  # no need to search if no skills available
-            self.bus.on("ovos.skills.fallback.pong", handle_ack)
-
+            return []
+
+        registered_fallbacks = self._fallback_registry_snapshot()
+        pool = [
+            skill_id for skill_id, priority in sorted(
+                registered_fallbacks.items(), key=operator.itemgetter(1))
+            if fb_range.start < priority <= fb_range.stop
+            and skill_id not in (sess.blacklisted_skills or [])
+            and self._fallback_allowed(skill_id)
+        ]
+        if not pool:
+            return []
+
+        session_id = sess.session_id
+        session_lock = self._acquire_fallback_session_lock(session_id)
+        responses: Dict[str, Optional[bool]] = {
+            skill_id: None for skill_id in pool
+        }
+        response_event = threading.Event()
+        response_lock = threading.Lock()
+        handlers: Dict[str, Callable] = {}
+
+        def make_handler(expected_skill_id: str) -> Callable:
+            def handle_ack(msg: Message) -> None:
+                response_session = SessionManager.get(msg)
+                if response_session is None or \
+                        response_session.session_id != session_id:
+                    return
+                skill_id = msg.data.get("skill_id")
+                can_handle = msg.data.get("can_handle")
+                valid = skill_id == expected_skill_id and \
+                    isinstance(can_handle, bool)
+                with response_lock:
+                    if responses[expected_skill_id] is not None:
+                        return
+                    responses[expected_skill_id] = can_handle if valid else False
+                response_event.set()
+
+            return handle_ack
+
+        try:
             LOG.info("checking for FallbackSkill candidates")
-            message.data["range"] = (fb_range.start, fb_range.stop)
-            # wait for all skills to acknowledge they want to answer fallback queries
-            self.bus.emit(message.forward("ovos.skills.fallback.ping",
-                                          message.data))
-            start = time.time()
-            while not all(s in skill_ids for s in self.registered_fallbacks) \
-                    and time.time() - start <= 0.5:
-                self._fallback_response_event.clear()
-                self._fallback_response_event.wait(0.02)
-
-            self.bus.remove("ovos.skills.fallback.pong", handle_ack)
-        return fallback_skills
+            for skill_id in pool:
+                pong_type = f"{skill_id}.fallback.pong"
+                handler = make_handler(skill_id)
+                handlers[pong_type] = handler
+                self.bus.on(pong_type, handler)
+
+            query_data = {
+                "utterances": list(message.data.get("utterances", [])),
+                "lang": message.data.get("lang")
+            }
+            for skill_id in pool:
+                self.bus.emit(message.reply(
+                    f"{skill_id}.fallback.ping", query_data))
+
+            try:
+                timeout = max(0.0, float(self.config.get(
+                    "fallback_query_timeout", 0.5)))
+            except (TypeError, ValueError):
+                LOG.warning("Invalid fallback_query_timeout; using 0.5 seconds")
+                timeout = 0.5
+            deadline = time.monotonic() + timeout
+            while True:
+                response_event.clear()
+                with response_lock:
+                    ordered_responses = [responses[skill_id]
+                                         for skill_id in pool]
+                for index, response in enumerate(ordered_responses):
+                    if response is None:
+                        break
+                    if response:
+                        selected = pool[index]
+                        LOG.info(f"{selected} will try to handle fallback")
+                        return [selected]
+                else:
+                    return []
+
+                remaining = deadline - time.monotonic()
+                if remaining <= 0:
+                    with response_lock:
+                        final_responses = [responses[skill_id]
+                                           for skill_id in pool]
+                    for index, response in enumerate(final_responses):
+                        if response:
+                            return [pool[index]]
+                    return []
+                response_event.wait(remaining)
+        finally:
+            for pong_type, handler in handlers.items():
+                self.bus.remove(pong_type, handler)
+            self._release_fallback_session_lock(session_id, session_lock)
 
     def _fallback_range(self, utterances: List[str], lang: str,
                         message: Message, fb_range: FallbackRange) -> Optional[IntentHandlerMatch]:
@@ -198,7 +281,8 @@ def _fallback_range(self, utterances: List[str], lang: str,
             return None
         # new style bus api
         available_skills = self._collect_fallback_skills(message, fb_range)
-        fallbacks = [(k, v) for k, v in self.registered_fallbacks.items()
+        registered_fallbacks = self._fallback_registry_snapshot()
+        fallbacks = [(k, v) for k, v in registered_fallbacks.items()
                      if k in available_skills]
         sorted_handlers = sorted(fallbacks, key=operator.itemgetter(1))
 
diff --git a/test/unittests/test_fallback_service.py b/test/unittests/test_fallback_service.py
index 4b56ecd7c58..cd38de719a7 100644
--- a/test/unittests/test_fallback_service.py
+++ b/test/unittests/test_fallback_service.py
@@ -18,7 +18,7 @@
 from unittest.mock import MagicMock, patch
 
 from ovos_bus_client.message import Message
-from ovos_bus_client.session import Session, SessionManager
+from ovos_bus_client.session import Session
 from ovos_utils.fakebus import FakeBus
 from ovos_workshop.permissions import FallbackMode
 
@@ -34,8 +34,10 @@ def _make_service(config=None) -> FallbackService:
         svc.bus = bus
         svc.config = config or {}
         svc.registered_fallbacks = {}
+        svc._registered_fallbacks_lock = threading.RLock()
+        svc._fallback_session_locks = {}
+        svc._fallback_session_locks_lock = threading.Lock()
         svc._lifecycle_handlers = {}
-        svc._fallback_response_event = threading.Event()
         svc.bus.on("ovos.skills.fallback.register", svc.handle_register_fallback)
         svc.bus.on("ovos.skills.fallback.deregister", svc.handle_deregister_fallback)
     return svc
@@ -202,7 +204,7 @@ def test_skill_in_range_receives_ping_and_responds(self):
 
         def capture_on(event, handler):
             nonlocal ack_handler
-            if event == "ovos.skills.fallback.pong":
+            if event == "skill_a.fallback.pong":
                 ack_handler = handler
 
         svc.bus.on = capture_on
@@ -225,7 +227,7 @@ def run():
             t.start()
             time.sleep(0.05)
             if ack_handler:
-                ack_handler(Message("ovos.skills.fallback.pong",
+                ack_handler(Message("skill_a.fallback.pong",
                                     {"skill_id": "skill_a", "can_handle": True}))
         finally:
             if t is not None:
@@ -233,6 +235,10 @@ def run():
             svc.shutdown()
 
         self.assertIn("skill_a", result_holder[0])
+        ping = svc.bus.emit.call_args[0][0]
+        self.assertEqual(ping.msg_type, "skill_a.fallback.ping")
+        self.assertEqual(ping.data, {"utterances": [], "lang": None})
+        self.assertNotIn("fallback_request_id", ping.context)
 
     def test_skill_responds_can_handle_false_excluded(self):
         """A skill that replies can_handle=False is not included."""
@@ -243,7 +249,7 @@ def test_skill_responds_can_handle_false_excluded(self):
 
         def capture_on(event, handler):
             nonlocal ack_handler
-            if event == "ovos.skills.fallback.pong":
+            if event == "skill_a.fallback.pong":
                 ack_handler = handler
 
         svc.bus.on = capture_on
@@ -266,7 +272,7 @@ def run():
             t.start()
             time.sleep(0.05)
             if ack_handler:
-                ack_handler(Message("ovos.skills.fallback.pong",
+                ack_handler(Message("skill_a.fallback.pong",
                                     {"skill_id": "skill_a", "can_handle": False}))
         finally:
             if t is not None:
@@ -275,9 +281,137 @@ def run():
 
         self.assertEqual(result_holder[0], [])
 
+    def test_first_willing_skill_is_selected_in_priority_order(self):
+        """Reply arrival does not override registered fallback priority."""
+        svc = _make_service()
+        svc.registered_fallbacks = {"skill_low": 80, "skill_high": 10}
+        handlers = {}
+
+        def capture_on(event, handler):
+            handlers[event] = handler
+
+        def emit(message):
+            skill_id = message.msg_type.removesuffix(".fallback.ping")
+            handlers[f"{skill_id}.fallback.pong"](message.reply(
+                f"{skill_id}.fallback.pong",
+                {"skill_id": skill_id,
+                 "can_handle": skill_id == "skill_low"}))
+
+        svc.bus.on = capture_on
+        svc.bus.remove = MagicMock()
+        svc.bus.emit = emit
+        sess = Session("s")
+        message = Message("test", context={"session": sess.serialize()})
+
+        with patch("ovos_core.intent_services.fallback_service.SessionManager.get",
+                   return_value=sess):
+            result = svc._collect_fallback_skills(
+                message, fb_range=FallbackRange(5, 90))
+
+        self.assertEqual(result, ["skill_low"])
+
+    def test_malformed_pong_is_treated_as_declined(self):
+        """A non-boolean can_handle value cannot claim an utterance."""
+        svc = _make_service()
+        svc.registered_fallbacks = {"skill_a": 50}
+        handlers = {}
+
+        svc.bus.on = lambda event, handler: handlers.update({event: handler})
+        svc.bus.remove = MagicMock()
+
+        def emit(message):
+            handlers["skill_a.fallback.pong"](message.reply(
+                "skill_a.fallback.pong",
+                {"skill_id": "skill_a", "can_handle": "yes"}))
+
+        svc.bus.emit = emit
+        sess = Session("s")
+        message = Message("test", context={"session": sess.serialize()})
+
+        with patch("ovos_core.intent_services.fallback_service.SessionManager.get",
+                   return_value=sess):
+            result = svc._collect_fallback_skills(
+                message, fb_range=FallbackRange(5, 90))
+
+        self.assertEqual(result, [])
+
+    def test_fallback_registry_snapshot_is_isolated_from_mutation(self):
+        """A match keeps a stable registry while skills register or leave."""
+        svc = _make_service()
+        svc.registered_fallbacks = {"skill_a": 50}
+
+        snapshot = svc._fallback_registry_snapshot()
+        svc.handle_register_fallback(Message(
+            "ovos.skills.fallback.register",
+            {"skill_id": "skill_b", "priority": 40},
+        ))
+        svc.handle_deregister_fallback(Message(
+            "ovos.skills.fallback.deregister", {"skill_id": "skill_a"}))
+
+        self.assertEqual(snapshot, {"skill_a": 50})
+        self.assertEqual(svc.registered_fallbacks, {"skill_b": 40})
+
+    def test_concurrent_sessions_do_not_consume_each_others_pongs(self):
+        """Same-topic pongs are correlated by their propagated session."""
+        svc = _make_service()
+        svc.registered_fallbacks = {"skill_a": 50}
+        handlers = []
+        results = {}
+
+        def capture_on(event, handler):
+            if event == "skill_a.fallback.pong":
+                handlers.append(handler)
+
+        svc.bus.on = capture_on
+        svc.bus.remove = MagicMock()
+        svc.bus.emit = MagicMock()
+
+        def run(session_id):
+            session = Session(session_id)
+            message = Message(
+                "test", context={"session": session.serialize()})
+            results[session_id] = svc._collect_fallback_skills(
+                message, fb_range=FallbackRange(5, 90))
+
+        threads = [threading.Thread(target=run, args=(session_id,))
+                   for session_id in ("a", "b")]
+        for thread in threads:
+            thread.start()
+        for _ in range(100):
+            if len(handlers) == 2:
+                break
+            time.sleep(0.01)
+        self.assertEqual(len(handlers), 2)
+
+        pong_a = Message(
+            "skill_a.fallback.pong",
+            {"skill_id": "skill_a", "can_handle": True},
+            {"session": Session("a").serialize()},
+        )
+        for handler in handlers:
+            handler(pong_a)
+        for _ in range(100):
+            if "a" in results:
+                break
+            time.sleep(0.01)
+        self.assertEqual(results.get("a"), ["skill_a"])
+        self.assertNotIn("b", results)
+
+        pong_b = Message(
+            "skill_a.fallback.pong",
+            {"skill_id": "skill_a", "can_handle": True},
+            {"session": Session("b").serialize()},
+        )
+        for handler in handlers:
+            handler(pong_b)
+        for thread in threads:
+            thread.join(timeout=1)
+
+        self.assertEqual(results.get("b"), ["skill_a"])
+
     def test_listener_removed_on_timeout(self):
         """bus.remove must be called even when no skill replies (timeout path)."""
-        svc = _make_service()
+        svc = _make_service(config={"fallback_query_timeout": 0})
         svc.registered_fallbacks = {"slow_skill": 50}
         svc.bus.on = MagicMock()
         svc.bus.remove = MagicMock()
@@ -285,15 +419,12 @@ def test_listener_removed_on_timeout(self):
 
         sess = Session("s")
         with patch("ovos_core.intent_services.fallback_service.SessionManager.get",
-                   return_value=sess), \
-             patch("ovos_core.intent_services.fallback_service.time") as mock_time:
-            # Simulate time jumping forward immediately so loop exits
-            mock_time.time.side_effect = [0, 1.0]
+                   return_value=sess):
             svc._collect_fallback_skills(Message("test"), fb_range=FallbackRange(5, 90))
 
         svc.bus.remove.assert_called_once()
         args = svc.bus.remove.call_args[0]
-        self.assertEqual(args[0], "ovos.skills.fallback.pong")
+        self.assertEqual(args[0], "slow_skill.fallback.pong")
 
     def test_blacklisted_skill_excluded(self):
         """Skills blacklisted by the session are not collected."""

From 52cd4cfd855d19768993c73bf83e3c680cf98a02 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Ga=C3=ABtan=20Trellu?= <gaetan.trellu@gmail.com>
Date: Mon, 3 Aug 2026 20:44:57 -0400
Subject: [PATCH 2/7] fix: harden fallback listener lifecycle

---
 ovos_core/intent_services/fallback_service.py | 29 +++++++-------
 test/unittests/test_fallback_service.py       | 38 ++++++++++++++++++-
 2 files changed, 51 insertions(+), 16 deletions(-)

diff --git a/ovos_core/intent_services/fallback_service.py b/ovos_core/intent_services/fallback_service.py
index 5d1ce802c83..481284c857e 100644
--- a/ovos_core/intent_services/fallback_service.py
+++ b/ovos_core/intent_services/fallback_service.py
@@ -53,9 +53,6 @@ def __init__(self, bus: Optional[Union[MessageBusClient, FakeBus]] = None,
 
     def _wire_lifecycle(self, skill_id: str) -> None:
         """Translate lifecycle done-signal for a fallback skill."""
-        if skill_id in self._lifecycle_handlers:
-            return
-
         def _on_start(message: Message) -> None:
             HandlerLifecycle(self.bus, message, skill_id=skill_id,
                              handler_name=f"{skill_id}.fallback").start()
@@ -67,17 +64,21 @@ def _on_response(message: Message) -> None:
             HandlerLifecycle(self.bus, message, skill_id=skill_id,
                              handler_name=f"{skill_id}.fallback").complete()
 
-        self.bus.on(f"ovos.skills.fallback.{skill_id}.start", _on_start)
-        self.bus.on(f"ovos.skills.fallback.{skill_id}.response", _on_response)
-        self._lifecycle_handlers[skill_id] = (_on_start, _on_response)
+        with self._registered_fallbacks_lock:
+            if skill_id in self._lifecycle_handlers:
+                return
+            self.bus.on(f"ovos.skills.fallback.{skill_id}.start", _on_start)
+            self.bus.on(f"ovos.skills.fallback.{skill_id}.response", _on_response)
+            self._lifecycle_handlers[skill_id] = (_on_start, _on_response)
 
     def _unwire_lifecycle(self, skill_id: str) -> None:
-        handlers = self._lifecycle_handlers.pop(skill_id, None)
-        if not handlers:
-            return
-        start_handler, response_handler = handlers
-        self.bus.remove(f"ovos.skills.fallback.{skill_id}.start", start_handler)
-        self.bus.remove(f"ovos.skills.fallback.{skill_id}.response", response_handler)
+        with self._registered_fallbacks_lock:
+            handlers = self._lifecycle_handlers.pop(skill_id, None)
+            if not handlers:
+                return
+            start_handler, response_handler = handlers
+            self.bus.remove(f"ovos.skills.fallback.{skill_id}.start", start_handler)
+            self.bus.remove(f"ovos.skills.fallback.{skill_id}.response", response_handler)
 
     def handle_register_fallback(self, message: Message) -> None:
         skill_id = message.data.get("skill_id")
@@ -216,7 +217,7 @@ def handle_ack(msg: Message) -> None:
                 "lang": message.data.get("lang")
             }
             for skill_id in pool:
-                self.bus.emit(message.reply(
+                self.bus.emit(message.forward(
                     f"{skill_id}.fallback.ping", query_data))
 
             try:
@@ -286,7 +287,7 @@ def _fallback_range(self, utterances: List[str], lang: str,
                      if k in available_skills]
         sorted_handlers = sorted(fallbacks, key=operator.itemgetter(1))
 
-        for skill_id, prio in sorted_handlers:
+        for skill_id, _priority in sorted_handlers:
             if skill_id in (sess.blacklisted_skills or []):
                 LOG.debug(f"ignoring match, skill_id '{skill_id}' blacklisted by Session '{sess.session_id}'")
                 continue
diff --git a/test/unittests/test_fallback_service.py b/test/unittests/test_fallback_service.py
index cd38de719a7..519da3ddea9 100644
--- a/test/unittests/test_fallback_service.py
+++ b/test/unittests/test_fallback_service.py
@@ -219,7 +219,10 @@ def run():
                        return_value=sess):
                 result_holder.append(
                     svc._collect_fallback_skills(
-                        Message("test"), fb_range=FallbackRange(5, 90)))
+                        Message("test", context={
+                            "source": "client",
+                            "destination": "skills",
+                        }), fb_range=FallbackRange(5, 90)))
 
         t = None
         try:
@@ -239,6 +242,8 @@ def run():
         self.assertEqual(ping.msg_type, "skill_a.fallback.ping")
         self.assertEqual(ping.data, {"utterances": [], "lang": None})
         self.assertNotIn("fallback_request_id", ping.context)
+        self.assertEqual(ping.context["source"], "client")
+        self.assertEqual(ping.context["destination"], "skills")
 
     def test_skill_responds_can_handle_false_excluded(self):
         """A skill that replies can_handle=False is not included."""
@@ -353,7 +358,7 @@ def test_fallback_registry_snapshot_is_isolated_from_mutation(self):
 
     def test_concurrent_sessions_do_not_consume_each_others_pongs(self):
         """Same-topic pongs are correlated by their propagated session."""
-        svc = _make_service()
+        svc = _make_service(config={"fallback_query_timeout": 2})
         svc.registered_fallbacks = {"skill_a": 50}
         handlers = []
         results = {}
@@ -586,6 +591,35 @@ def test_register_wires_lifecycle_listeners(self):
                     {"skill_id": "skill_a", "priority": 50}))
         self.assertIn("skill_a", svc._lifecycle_handlers)
 
+    def test_concurrent_registration_wires_one_listener_pair(self):
+        """Concurrent registration cannot leak duplicate lifecycle handlers."""
+        svc = _make_service()
+        topics = []
+        start = threading.Barrier(8)
+
+        def slow_on(topic, handler):
+            topics.append(topic)
+            time.sleep(0.01)
+
+        svc.bus.on = slow_on
+
+        def wire():
+            start.wait()
+            svc._wire_lifecycle("skill_a")
+
+        threads = [threading.Thread(target=wire) for _ in range(8)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join(timeout=1)
+
+        self.assertTrue(all(not thread.is_alive() for thread in threads))
+        self.assertEqual(topics, [
+            "ovos.skills.fallback.skill_a.start",
+            "ovos.skills.fallback.skill_a.response",
+        ])
+        self.assertIn("skill_a", svc._lifecycle_handlers)
+
     def test_skill_start_emits_handler_start(self):
         """The skill's fallback .start is re-emitted as handler.start with the
         skill_id stamped in context."""

From 4e5a30df35a8497e364e2030effdd2f1f63fa0b3 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Ga=C3=ABtan=20Trellu?= <gaetan.trellu@gmail.com>
Date: Mon, 3 Aug 2026 20:45:34 -0400
Subject: [PATCH 3/7] test: bound concurrent fallback polling

---
 test/unittests/test_fallback_service.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/test/unittests/test_fallback_service.py b/test/unittests/test_fallback_service.py
index 519da3ddea9..75da5025c01 100644
--- a/test/unittests/test_fallback_service.py
+++ b/test/unittests/test_fallback_service.py
@@ -358,7 +358,7 @@ def test_fallback_registry_snapshot_is_isolated_from_mutation(self):
 
     def test_concurrent_sessions_do_not_consume_each_others_pongs(self):
         """Same-topic pongs are correlated by their propagated session."""
-        svc = _make_service(config={"fallback_query_timeout": 2})
+        svc = _make_service(config={"fallback_query_timeout": 30})
         svc.registered_fallbacks = {"skill_a": 50}
         handlers = []
         results = {}

From bf3d14776f606b07a438aab597f12623cd400385 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Ga=C3=ABtan=20Trellu?= <gaetan.trellu@gmail.com>
Date: Mon, 3 Aug 2026 21:09:42 -0400
Subject: [PATCH 4/7] test: cover skill-addressed fallback probe

---
 test/end2end/test_fallback.py | 43 ++++++++++++++++++++++++++++++-----
 1 file changed, 37 insertions(+), 6 deletions(-)

diff --git a/test/end2end/test_fallback.py b/test/end2end/test_fallback.py
index 51fb4d4f489..455c17c9179 100644
--- a/test/end2end/test_fallback.py
+++ b/test/end2end/test_fallback.py
@@ -44,6 +44,31 @@ class TestFallback(TestCase):
 
     skill_id = "ovos-skill-fallback-unknown.openvoiceos"
 
+    @classmethod
+    def _wire_skill_addressed_probe(cls, minicroft):
+        """Expose the companion Workshop FALLBACK-1 probe in this test.
+
+        The Ovoscope workflow intentionally installs released sibling packages
+        while testing this Core checkout.  Until the coordinated Workshop
+        change is released, adapt only the loaded test skill's capability probe
+        to the skill-addressed FALLBACK-1 topics.  The fallback request itself
+        still runs through the real skill and its normal lifecycle handlers.
+        """
+        skill = minicroft.plugin_skills[cls.skill_id].instance
+        ping_type = f"{cls.skill_id}.fallback.ping"
+        pong_type = f"{cls.skill_id}.fallback.pong"
+
+        def handle_ping(message: Message) -> None:
+            minicroft.bus.emit(message.reply(
+                pong_type,
+                data={"skill_id": cls.skill_id,
+                      "can_handle": skill.can_answer(message)},
+                context={"skill_id": cls.skill_id}
+            ))
+
+        minicroft.bus.on(ping_type, handle_ping)
+        return handle_ping
+
     def setUp(self):
         LOG.set_level("DEBUG")
 
@@ -54,6 +79,7 @@ def _run_fallback_match(self, namespace: str) -> None:
         modernize, emit_legacy, utt_topic = NAMESPACE_PATHS[namespace]
         minicroft = get_minicroft([self.skill_id], modernize=modernize,
                                   emit_legacy=emit_legacy)
+        probe_handler = self._wire_skill_addressed_probe(minicroft)
         try:
 
             session = Session("123")
@@ -73,8 +99,7 @@ def _run_fallback_match(self, namespace: str) -> None:
                 entry_points=[utt_topic],
                 final_session=final_session,
                 keep_original_src=[
-                    "ovos.skills.fallback.ping",
-                # "ovos.skills.fallback.pong", # TODO
+                    f"{self.skill_id}.fallback.ping",
                 ],
                 ignore_messages=["recognizer_loop:audio_output_start",
                                   "recognizer_loop:audio_output_end"],
@@ -82,9 +107,11 @@ def _run_fallback_match(self, namespace: str) -> None:
                 source_message=message,
                 expected_messages=[
                     message,
-                    Message("ovos.skills.fallback.ping",
-                            {"utterances": ["hello world"], "lang": session.lang, "range": [90, 101]}),
-                    Message("ovos.skills.fallback.pong", {"skill_id": self.skill_id, "can_handle": True}),
+                    Message(f"{self.skill_id}.fallback.ping",
+                            {"utterances": ["hello world"],
+                             "lang": session.lang}),
+                    Message(f"{self.skill_id}.fallback.pong",
+                            {"skill_id": self.skill_id, "can_handle": True}),
                 # PIPELINE-1 §9.2: matched notification precedes the dispatch. The
                 # fallback match_type is the .request topic; it bears no ':' so
                 # skill_id/intent_name resolve to that topic.
@@ -95,7 +122,9 @@ def _run_fallback_match(self, namespace: str) -> None:
                     Message(HANDLER_START,
                             data={"intent_name": f"ovos.skills.fallback.{self.skill_id}.request"}),
                     Message(f"ovos.skills.fallback.{self.skill_id}.request",
-                            {"utterances": ["hello world"], "lang": session.lang, "range": [90, 101], "skill_id": self.skill_id}),
+                            {"utterances": ["hello world"],
+                             "lang": session.lang,
+                             "skill_id": self.skill_id}),
                     Message(f"ovos.skills.fallback.{self.skill_id}.start", {}),
                 # core reports the fallback dispatch lifecycle as the framework
                 # done-signal by translating the skill's own .start/.response
@@ -128,6 +157,8 @@ def _run_fallback_match(self, namespace: str) -> None:
 
             test.execute(timeout=10)
         finally:
+            minicroft.bus.remove(f"{self.skill_id}.fallback.ping",
+                                 probe_handler)
             minicroft.stop()
 
     def test_fallback_match(self):

From c61423213660995e80dcd632dbb097e03deab9b3 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Ga=C3=ABtan=20Trellu?= <gaetan.trellu@gmail.com>
Date: Mon, 3 Aug 2026 23:58:59 -0400
Subject: [PATCH 5/7] fix: derive fallback probes as spec replies

---
 ovos_core/intent_services/fallback_service.py |  4 +++-
 test/end2end/test_fallback.py                 | 11 ++++++-----
 test/unittests/test_fallback_service.py       |  4 ++--
 3 files changed, 11 insertions(+), 8 deletions(-)

diff --git a/ovos_core/intent_services/fallback_service.py b/ovos_core/intent_services/fallback_service.py
index 481284c857e..a5c3b67a117 100644
--- a/ovos_core/intent_services/fallback_service.py
+++ b/ovos_core/intent_services/fallback_service.py
@@ -217,7 +217,9 @@ def handle_ack(msg: Message) -> None:
                 "lang": message.data.get("lang")
             }
             for skill_id in pool:
-                self.bus.emit(message.forward(
+                # FALLBACK-1 section 6.1 defines this as a dotted-addressed
+                # reply derived from the inbound utterance envelope.
+                self.bus.emit(message.reply(
                     f"{skill_id}.fallback.ping", query_data))
 
             try:
diff --git a/test/end2end/test_fallback.py b/test/end2end/test_fallback.py
index 455c17c9179..5a2ae0a5f55 100644
--- a/test/end2end/test_fallback.py
+++ b/test/end2end/test_fallback.py
@@ -95,12 +95,12 @@ def _run_fallback_match(self, namespace: str) -> None:
                 minicroft=minicroft,
                 skill_ids=[self.skill_id],
                 eof_msgs=[UTTERANCE_HANDLED],
-                flip_points=[utt_topic],
+                flip_points=[
+                    utt_topic,
+                    f"{self.skill_id}.fallback.pong",
+                ],
                 entry_points=[utt_topic],
                 final_session=final_session,
-                keep_original_src=[
-                    f"{self.skill_id}.fallback.ping",
-                ],
                 ignore_messages=["recognizer_loop:audio_output_start",
                                   "recognizer_loop:audio_output_end"],
                 activation_points=[f"ovos.skills.fallback.{self.skill_id}.request"],
@@ -111,7 +111,8 @@ def _run_fallback_match(self, namespace: str) -> None:
                             {"utterances": ["hello world"],
                              "lang": session.lang}),
                     Message(f"{self.skill_id}.fallback.pong",
-                            {"skill_id": self.skill_id, "can_handle": True}),
+                            {"skill_id": self.skill_id, "can_handle": True},
+                            {"source": "A", "destination": "B"}),
                 # PIPELINE-1 §9.2: matched notification precedes the dispatch. The
                 # fallback match_type is the .request topic; it bears no ':' so
                 # skill_id/intent_name resolve to that topic.
diff --git a/test/unittests/test_fallback_service.py b/test/unittests/test_fallback_service.py
index 75da5025c01..4a645a0753f 100644
--- a/test/unittests/test_fallback_service.py
+++ b/test/unittests/test_fallback_service.py
@@ -242,8 +242,8 @@ def run():
         self.assertEqual(ping.msg_type, "skill_a.fallback.ping")
         self.assertEqual(ping.data, {"utterances": [], "lang": None})
         self.assertNotIn("fallback_request_id", ping.context)
-        self.assertEqual(ping.context["source"], "client")
-        self.assertEqual(ping.context["destination"], "skills")
+        self.assertEqual(ping.context["source"], "skills")
+        self.assertEqual(ping.context["destination"], "client")
 
     def test_skill_responds_can_handle_false_excluded(self):
         """A skill that replies can_handle=False is not included."""

From 9d7e066fafc0d70c9582bbb1b339b500ee70ed7b Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Ga=C3=ABtan=20Trellu?= <gaetan.trellu@gmail.com>
Date: Wed, 12 Aug 2026 09:28:49 -0400
Subject: [PATCH 6/7] fix: normalize session pipeline blacklist matching

---
 ovos_core/intent_services/service.py          | 23 +++++++++++++----
 .../unittests/test_intent_service_extended.py | 25 ++++++++++++++++++-
 2 files changed, 42 insertions(+), 6 deletions(-)

diff --git a/ovos_core/intent_services/service.py b/ovos_core/intent_services/service.py
index a1deaa95102..acf52e3e181 100644
--- a/ovos_core/intent_services/service.py
+++ b/ovos_core/intent_services/service.py
@@ -303,10 +303,19 @@ def get_pipeline(self, session=None) -> List[Tuple[str, Callable]]:
         # orchestrator-only: no `match` call is made and no bus event is
         # emitted for the skip, it is observable only as a non-invocation.
         # Unknown pipeline_ids in the blacklist are harmless no-ops.
-        blacklisted = set(session.blacklisted_pipelines or [])
-        requested = [p for p in session.pipeline if p not in blacklisted]
+        blacklisted = {
+            _PIPELINE_MIGRATION_MAP.get(pipeline_id, pipeline_id)
+            for pipeline_id in session.blacklisted_pipelines or []
+        }
+
+        def is_blacklisted(matcher_id: str) -> bool:
+            normalized = _PIPELINE_MIGRATION_MAP.get(matcher_id, matcher_id)
+            plugin_id = _PIPELINE_RE.sub('', normalized)
+            return normalized in blacklisted or plugin_id in blacklisted
+
+        requested = [p for p in session.pipeline if not is_blacklisted(p)]
         if blacklisted:
-            skipped = [p for p in session.pipeline if p in blacklisted]
+            skipped = [p for p in session.pipeline if is_blacklisted(p)]
             if skipped:
                 LOG.debug(f"Session '{session.session_id}' blacklisted "
                           f"pipelines skipped: {skipped}")
@@ -630,7 +639,11 @@ def handle_utterance(self, message: Message):
                 langs = [lang]
                 if self.config.get("multilingual_matching"):
                     # if multilingual matching is enabled, attempt to match all user languages if main fails
-                    langs += [l for l in get_valid_languages() if l != lang]
+                    langs += [
+                        candidate_lang
+                        for candidate_lang in get_valid_languages()
+                        if candidate_lang != lang
+                    ]
                 for intent_lang in langs:
                     try:
                         match = match_func(utterances, intent_lang, message)
@@ -867,4 +880,4 @@ def launch_standalone():
 
 
 if __name__ == "__main__":
-    launch_standalone()
\ No newline at end of file
+    launch_standalone()
diff --git a/test/unittests/test_intent_service_extended.py b/test/unittests/test_intent_service_extended.py
index 11bc4aa2ea3..885c64e14b5 100644
--- a/test/unittests/test_intent_service_extended.py
+++ b/test/unittests/test_intent_service_extended.py
@@ -17,7 +17,7 @@
 from unittest.mock import MagicMock, patch
 
 from ovos_bus_client.message import Message
-from ovos_bus_client.session import Session, SessionManager
+from ovos_bus_client.session import Session
 from ovos_plugin_manager.templates.pipeline import (
     IntentHandlerMatch,
     ConfidenceMatcherPipeline,
@@ -310,6 +310,29 @@ def test_blacklist_overrides_explicit_pipeline_preference(self):
         result = svc.get_pipeline(session=sess)
         self.assertEqual(result, [])
 
+    def test_plugin_blacklist_skips_all_confidence_matchers_before_lookup(self):
+        """A base plugin policy ID blocks its suffixed matcher variants.
+
+        The deployment blacklist is expressed in installed plugin IDs while a
+        session pipeline contains confidence-suffixed matcher IDs.  Filtering
+        must happen before matcher lookup so an intentionally disabled plugin
+        is neither invoked nor reported as unknown.
+        """
+        svc = self._svc_with_adapt_fallback()
+        sess = Session("s")
+        sess.pipeline = [
+            "ovos-adapt-pipeline-plugin-high",
+            "fallback_high",
+        ]
+        sess.blacklisted_pipelines = ["ovos-adapt-pipeline-plugin"]
+
+        with patch.object(svc, "get_pipeline_matcher",
+                          wraps=svc.get_pipeline_matcher) as get_matcher:
+            result = svc.get_pipeline(session=sess)
+
+        self.assertEqual([matcher[0] for matcher in result], ["fallback_high"])
+        get_matcher.assert_called_once_with("fallback_high")
+
     def test_unknown_blacklisted_id_is_harmless_noop(self):
         """Unknown pipeline_ids in blacklisted_pipelines are ignored without
         error and don't affect the effective pipeline (§5.2)."""

From 28b37143edf7320203cc0a0e2f74116cb1967251 Mon Sep 17 00:00:00 2001
From: JarbasAi <jarbasai@mailfence.com>
Date: Wed, 12 Aug 2026 22:58:55 +0100
Subject: [PATCH 7/7] fix: keep the broadcast fallback poll alongside
 skill-addressed pings for the migration window
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The skill-addressed fallback ping/pong machinery landed with no
compatibility window: any released ovos-workshop (pre-#465, broadcast-only)
paired with this branch never gets its fallback ping answered, since the
broadcast ping/pong collector was dropped outright. FALLBACK-1 §6.1 makes
the addressed topics normative but explicitly sanctions the broadcast poll
as an observably-equivalent optimisation, so restore it for one
deprecation window (kill-switch #837 conventions):

- emit the general `ovos.skills.fallback.ping` broadcast once per poll
  round alongside the addressed pings, and keep a `ovos.skills.fallback.pong`
  collector with the same session filter as the addressed collectors.
- dedup pongs by skill_id: a skill running fixed ovos-workshop (#465)
  answers BOTH ping families during the window, and must only count once
  (first answer wins).
- delete the end2end `_wire_skill_addressed_probe` fake and let the real
  (released) ovos-workshop installed by the test run answer the ping
  honestly; expected_messages updated to reflect the addressed ping being
  emitted-but-unanswered against a pre-#465 workshop.
- fix the CodeRabbit-flagged flaky session-lock unit test's docstring
  (it exercises the session-id filter, not the lock) and add a real
  same-session lock serialization test.
- revert the unrelated pipeline-blacklist normalization change to
  intent_services/service.py (and its test) that had leaked into this
  branch; it is being split into its own PR.

Executed matrix (probe-free harness, real ovoscope + real workshop):
fixed-core+fixed-ws, fixed-core+ws-dev, core-dev+fixed-ws, and
fixed-core+PyPI ovos-workshop==9.3.9a1 are all GREEN (ping answered,
dispatched exactly once, skill spoke exactly once).

Narrows #807 (same-session stale-pong residue remains; needs a round
nonce -- pre-existing on dev).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 ovos_core/intent_services/fallback_service.py |  54 +++++-
 ovos_core/intent_services/service.py          |  31 ++--
 test/end2end/test_fallback.py                 |  43 ++---
 test/unittests/test_fallback_service.py       | 160 +++++++++++++++++-
 .../unittests/test_intent_service_extended.py |  63 ++++---
 5 files changed, 261 insertions(+), 90 deletions(-)

diff --git a/ovos_core/intent_services/fallback_service.py b/ovos_core/intent_services/fallback_service.py
index a5c3b67a117..c80bc2d67da 100644
--- a/ovos_core/intent_services/fallback_service.py
+++ b/ovos_core/intent_services/fallback_service.py
@@ -186,6 +186,18 @@ def _collect_fallback_skills(self, message: Message,
         response_lock = threading.Lock()
         handlers: Dict[str, Callable] = {}
 
+        def _record(expected_skill_id: str, can_handle) -> None:
+            """First answer for a skill_id wins, regardless of which pong
+            topic (addressed or broadcast) it arrived on -- a skill running
+            fixed ovos-workshop (#465) answers BOTH ping families during the
+            migration window and must only count once."""
+            valid = isinstance(can_handle, bool)
+            with response_lock:
+                if responses[expected_skill_id] is not None:
+                    return
+                responses[expected_skill_id] = can_handle if valid else False
+            response_event.set()
+
         def make_handler(expected_skill_id: str) -> Callable:
             def handle_ack(msg: Message) -> None:
                 response_session = SessionManager.get(msg)
@@ -194,16 +206,30 @@ def handle_ack(msg: Message) -> None:
                     return
                 skill_id = msg.data.get("skill_id")
                 can_handle = msg.data.get("can_handle")
-                valid = skill_id == expected_skill_id and \
-                    isinstance(can_handle, bool)
-                with response_lock:
-                    if responses[expected_skill_id] is not None:
-                        return
-                    responses[expected_skill_id] = can_handle if valid else False
-                response_event.set()
+                if skill_id != expected_skill_id:
+                    _record(expected_skill_id, False)
+                    return
+                _record(expected_skill_id, can_handle)
 
             return handle_ack
 
+        def handle_broadcast_pong(msg: Message) -> None:
+            # DEPRECATION WINDOW (ovos-core kill-switch #837 conventions):
+            # the general `ovos.skills.fallback.pong` collector is kept
+            # alongside the skill-addressed one for one deprecation window,
+            # so that a released ovos-workshop (pre-#465, only answering the
+            # broadcast ping) still gets picked up. Removable once the
+            # ovos-workshop floor pin guarantees dual-binding (#465).
+            response_session = SessionManager.get(msg)
+            if response_session is None or \
+                    response_session.session_id != session_id:
+                return
+            skill_id = msg.data.get("skill_id")
+            can_handle = msg.data.get("can_handle")
+            if skill_id not in responses:
+                return
+            _record(skill_id, can_handle)
+
         try:
             LOG.info("checking for FallbackSkill candidates")
             for skill_id in pool:
@@ -212,6 +238,13 @@ def handle_ack(msg: Message) -> None:
                 handlers[pong_type] = handler
                 self.bus.on(pong_type, handler)
 
+            # DEPRECATION WINDOW (ovos-core kill-switch #837 conventions):
+            # bind the broadcast pong collector once per poll round, with
+            # the same session filter as the addressed collectors above.
+            broadcast_pong_type = "ovos.skills.fallback.pong"
+            handlers[broadcast_pong_type] = handle_broadcast_pong
+            self.bus.on(broadcast_pong_type, handle_broadcast_pong)
+
             query_data = {
                 "utterances": list(message.data.get("utterances", [])),
                 "lang": message.data.get("lang")
@@ -221,6 +254,13 @@ def handle_ack(msg: Message) -> None:
                 # reply derived from the inbound utterance envelope.
                 self.bus.emit(message.reply(
                     f"{skill_id}.fallback.ping", query_data))
+            # DEPRECATION WINDOW: also broadcast the legacy general ping once
+            # per poll round, so a released ovos-workshop (pre-#465) that
+            # only binds `ovos.skills.fallback.ping` still answers. Removable
+            # when the ovos-workshop floor pin guarantees dual-binding
+            # (#465) -- see ovos-core kill-switch #837 conventions.
+            self.bus.emit(message.forward(
+                "ovos.skills.fallback.ping", query_data))
 
             try:
                 timeout = max(0.0, float(self.config.get(
diff --git a/ovos_core/intent_services/service.py b/ovos_core/intent_services/service.py
index acf52e3e181..f1e84c74e21 100644
--- a/ovos_core/intent_services/service.py
+++ b/ovos_core/intent_services/service.py
@@ -252,9 +252,11 @@ def disambiguate_lang(message):
         for k in lang_keys:
             if k in message.context:
                 v = standardize_lang(message.context[k])
-                # closest_lang already applies the "distance below 10" threshold
-                # and returns None when no candidate is close enough
-                best_lang = closest_lang(v, valid_langs, max_distance=10)
+                # closest_lang applies the language-distance threshold and
+                # returns None when no candidate is close enough. The bound is
+                # inclusive, so a member language still matches its
+                # macrolanguage (distance 10, eg. "arz" against "ar")
+                best_lang = closest_lang(v, valid_langs)
                 if best_lang is None:
                     LOG.warning(f"ignoring {k}, {v} is not in enabled languages: {valid_langs}")
                     continue
@@ -303,19 +305,10 @@ def get_pipeline(self, session=None) -> List[Tuple[str, Callable]]:
         # orchestrator-only: no `match` call is made and no bus event is
         # emitted for the skip, it is observable only as a non-invocation.
         # Unknown pipeline_ids in the blacklist are harmless no-ops.
-        blacklisted = {
-            _PIPELINE_MIGRATION_MAP.get(pipeline_id, pipeline_id)
-            for pipeline_id in session.blacklisted_pipelines or []
-        }
-
-        def is_blacklisted(matcher_id: str) -> bool:
-            normalized = _PIPELINE_MIGRATION_MAP.get(matcher_id, matcher_id)
-            plugin_id = _PIPELINE_RE.sub('', normalized)
-            return normalized in blacklisted or plugin_id in blacklisted
-
-        requested = [p for p in session.pipeline if not is_blacklisted(p)]
+        blacklisted = set(session.blacklisted_pipelines or [])
+        requested = [p for p in session.pipeline if p not in blacklisted]
         if blacklisted:
-            skipped = [p for p in session.pipeline if is_blacklisted(p)]
+            skipped = [p for p in session.pipeline if p in blacklisted]
             if skipped:
                 LOG.debug(f"Session '{session.session_id}' blacklisted "
                           f"pipelines skipped: {skipped}")
@@ -639,11 +632,7 @@ def handle_utterance(self, message: Message):
                 langs = [lang]
                 if self.config.get("multilingual_matching"):
                     # if multilingual matching is enabled, attempt to match all user languages if main fails
-                    langs += [
-                        candidate_lang
-                        for candidate_lang in get_valid_languages()
-                        if candidate_lang != lang
-                    ]
+                    langs += [l for l in get_valid_languages() if l != lang]
                 for intent_lang in langs:
                     try:
                         match = match_func(utterances, intent_lang, message)
@@ -880,4 +869,4 @@ def launch_standalone():
 
 
 if __name__ == "__main__":
-    launch_standalone()
+    launch_standalone()
\ No newline at end of file
diff --git a/test/end2end/test_fallback.py b/test/end2end/test_fallback.py
index 5a2ae0a5f55..59b26a92684 100644
--- a/test/end2end/test_fallback.py
+++ b/test/end2end/test_fallback.py
@@ -44,31 +44,6 @@ class TestFallback(TestCase):
 
     skill_id = "ovos-skill-fallback-unknown.openvoiceos"
 
-    @classmethod
-    def _wire_skill_addressed_probe(cls, minicroft):
-        """Expose the companion Workshop FALLBACK-1 probe in this test.
-
-        The Ovoscope workflow intentionally installs released sibling packages
-        while testing this Core checkout.  Until the coordinated Workshop
-        change is released, adapt only the loaded test skill's capability probe
-        to the skill-addressed FALLBACK-1 topics.  The fallback request itself
-        still runs through the real skill and its normal lifecycle handlers.
-        """
-        skill = minicroft.plugin_skills[cls.skill_id].instance
-        ping_type = f"{cls.skill_id}.fallback.ping"
-        pong_type = f"{cls.skill_id}.fallback.pong"
-
-        def handle_ping(message: Message) -> None:
-            minicroft.bus.emit(message.reply(
-                pong_type,
-                data={"skill_id": cls.skill_id,
-                      "can_handle": skill.can_answer(message)},
-                context={"skill_id": cls.skill_id}
-            ))
-
-        minicroft.bus.on(ping_type, handle_ping)
-        return handle_ping
-
     def setUp(self):
         LOG.set_level("DEBUG")
 
@@ -79,7 +54,6 @@ def _run_fallback_match(self, namespace: str) -> None:
         modernize, emit_legacy, utt_topic = NAMESPACE_PATHS[namespace]
         minicroft = get_minicroft([self.skill_id], modernize=modernize,
                                   emit_legacy=emit_legacy)
-        probe_handler = self._wire_skill_addressed_probe(minicroft)
         try:
 
             session = Session("123")
@@ -97,7 +71,6 @@ def _run_fallback_match(self, namespace: str) -> None:
                 eof_msgs=[UTTERANCE_HANDLED],
                 flip_points=[
                     utt_topic,
-                    f"{self.skill_id}.fallback.pong",
                 ],
                 entry_points=[utt_topic],
                 final_session=final_session,
@@ -107,12 +80,20 @@ def _run_fallback_match(self, namespace: str) -> None:
                 source_message=message,
                 expected_messages=[
                     message,
+                # DEPRECATION WINDOW (kill-switch #837 conventions): core
+                # still polls both ping families. The released ovos-workshop
+                # installed by this test run (pre-#465, dev floor pin) only
+                # binds the legacy broadcast ping, so the skill-addressed
+                # ping is emitted but goes unanswered here -- it is only
+                # honored once ovos-workshop >=#465 is the floor pin.
                     Message(f"{self.skill_id}.fallback.ping",
                             {"utterances": ["hello world"],
                              "lang": session.lang}),
-                    Message(f"{self.skill_id}.fallback.pong",
-                            {"skill_id": self.skill_id, "can_handle": True},
-                            {"source": "A", "destination": "B"}),
+                    Message("ovos.skills.fallback.ping",
+                            {"utterances": ["hello world"],
+                             "lang": session.lang}),
+                    Message("ovos.skills.fallback.pong",
+                            {"skill_id": self.skill_id, "can_handle": True}),
                 # PIPELINE-1 §9.2: matched notification precedes the dispatch. The
                 # fallback match_type is the .request topic; it bears no ':' so
                 # skill_id/intent_name resolve to that topic.
@@ -158,8 +139,6 @@ def _run_fallback_match(self, namespace: str) -> None:
 
             test.execute(timeout=10)
         finally:
-            minicroft.bus.remove(f"{self.skill_id}.fallback.ping",
-                                 probe_handler)
             minicroft.stop()
 
     def test_fallback_match(self):
diff --git a/test/unittests/test_fallback_service.py b/test/unittests/test_fallback_service.py
index 4a645a0753f..b7c3d8609d2 100644
--- a/test/unittests/test_fallback_service.py
+++ b/test/unittests/test_fallback_service.py
@@ -238,12 +238,17 @@ def run():
             svc.shutdown()
 
         self.assertIn("skill_a", result_holder[0])
-        ping = svc.bus.emit.call_args[0][0]
-        self.assertEqual(ping.msg_type, "skill_a.fallback.ping")
+        emitted = [call.args[0] for call in svc.bus.emit.call_args_list]
+        # DEPRECATION WINDOW: both the skill-addressed ping and the legacy
+        # broadcast ping are emitted per poll round (kill-switch #837
+        # conventions) -- see fallback_service._collect_fallback_skills.
+        ping = next(m for m in emitted if m.msg_type == "skill_a.fallback.ping")
+        broadcast_ping = next(m for m in emitted if m.msg_type == "ovos.skills.fallback.ping")
         self.assertEqual(ping.data, {"utterances": [], "lang": None})
         self.assertNotIn("fallback_request_id", ping.context)
         self.assertEqual(ping.context["source"], "skills")
         self.assertEqual(ping.context["destination"], "client")
+        self.assertEqual(broadcast_ping.data, {"utterances": [], "lang": None})
 
     def test_skill_responds_can_handle_false_excluded(self):
         """A skill that replies can_handle=False is not included."""
@@ -357,7 +362,13 @@ def test_fallback_registry_snapshot_is_isolated_from_mutation(self):
         self.assertEqual(svc.registered_fallbacks, {"skill_b": 40})
 
     def test_concurrent_sessions_do_not_consume_each_others_pongs(self):
-        """Same-topic pongs are correlated by their propagated session."""
+        """Same-topic pongs from DIFFERENT sessions are correlated by their
+        propagated session id and never cross-consumed. This exercises the
+        session-id filter in the pong handler, NOT the same-session
+        serialization lock -- see
+        test_same_session_polls_are_serialized_by_lock for that. A generous
+        explicit fallback_query_timeout keeps this from flaking under CI
+        load (CodeRabbit-flagged)."""
         svc = _make_service(config={"fallback_query_timeout": 30})
         svc.registered_fallbacks = {"skill_a": 50}
         handlers = []
@@ -414,6 +425,140 @@ def run(session_id):
 
         self.assertEqual(results.get("b"), ["skill_a"])
 
+    def test_dedup_pong_across_both_ping_families_counts_once(self):
+        """DEPRECATION WINDOW: a skill running fixed ovos-workshop (#465)
+        answers BOTH the skill-addressed pong and the legacy broadcast pong
+        for the same poll round. It must be counted exactly once in the
+        returned pool, keyed by skill_id, with the first answer winning."""
+        svc = _make_service(config={"fallback_query_timeout": 2})
+        svc.registered_fallbacks = {"skill_a": 50}
+        sess = Session("dedup-session")
+        message = Message("test", context={"session": sess.serialize()})
+
+        addressed_handler = {}
+        broadcast_handler = {}
+        orig_on = svc.bus.on
+
+        def capture_on(event, handler):
+            if event == "skill_a.fallback.pong":
+                addressed_handler["h"] = handler
+            elif event == "ovos.skills.fallback.pong":
+                broadcast_handler["h"] = handler
+            return orig_on(event, handler)
+
+        svc.bus.on = capture_on
+
+        def fake_emit(msg):
+            if msg.msg_type == "skill_a.fallback.ping":
+                # a dual-bound skill answers the addressed ping on BOTH pong
+                # topics: the addressed one wins (first answer), the
+                # broadcast one must be ignored as a dup.
+                addressed_handler["h"](Message(
+                    "skill_a.fallback.pong",
+                    {"skill_id": "skill_a", "can_handle": True},
+                    {"session": sess.serialize()}))
+                broadcast_handler["h"](Message(
+                    "ovos.skills.fallback.pong",
+                    {"skill_id": "skill_a", "can_handle": False},
+                    {"session": sess.serialize()}))
+
+        svc.bus.emit = fake_emit
+
+        result = svc._collect_fallback_skills(message, fb_range=FallbackRange(5, 90))
+        # first answer (addressed pong, can_handle=True) wins; the
+        # contradicting broadcast dup (can_handle=False) is ignored, and
+        # skill_a appears exactly once.
+        self.assertEqual(result, ["skill_a"])
+
+    def test_dedup_first_answer_wins_when_broadcast_arrives_first(self):
+        """Symmetric case: broadcast pong arrives first and wins even though
+        the addressed pong (arriving second) disagrees."""
+        svc = _make_service(config={"fallback_query_timeout": 2})
+        svc.registered_fallbacks = {"skill_a": 50}
+        sess = Session("dedup-session-2")
+        message = Message("test", context={"session": sess.serialize()})
+
+        addressed_handler = {}
+        broadcast_handler = {}
+        orig_on = svc.bus.on
+
+        def capture_on(event, handler):
+            if event == "skill_a.fallback.pong":
+                addressed_handler["h"] = handler
+            elif event == "ovos.skills.fallback.pong":
+                broadcast_handler["h"] = handler
+            return orig_on(event, handler)
+
+        svc.bus.on = capture_on
+
+        def fake_emit(msg):
+            if msg.msg_type == "skill_a.fallback.ping":
+                broadcast_handler["h"](Message(
+                    "ovos.skills.fallback.pong",
+                    {"skill_id": "skill_a", "can_handle": False},
+                    {"session": sess.serialize()}))
+                addressed_handler["h"](Message(
+                    "skill_a.fallback.pong",
+                    {"skill_id": "skill_a", "can_handle": True},
+                    {"session": sess.serialize()}))
+
+        svc.bus.emit = fake_emit
+
+        result = svc._collect_fallback_skills(message, fb_range=FallbackRange(5, 90))
+        # broadcast pong (can_handle=False) arrived first and wins, so
+        # skill_a is NOT selected despite the later addressed pong saying True.
+        self.assertEqual(result, [])
+
+    def test_same_session_polls_are_serialized_by_lock(self):
+        """_acquire_fallback_session_lock/_release_fallback_session_lock
+        serialize two concurrent polls for the SAME session id: the second
+        acquirer only proceeds once the first releases -- verified by a
+        strict entry/exit ordering with no interleaving."""
+        svc = _make_service()
+        session_id = "shared-session"
+        order = []
+        holder_acquired = threading.Event()
+        release_signal = threading.Event()
+
+        def holder():
+            lock = svc._acquire_fallback_session_lock(session_id)
+            order.append("holder-acquired")
+            holder_acquired.set()
+            release_signal.wait(timeout=2)
+            order.append("holder-releasing")
+            svc._release_fallback_session_lock(session_id, lock)
+
+        def waiter():
+            self.assertTrue(holder_acquired.wait(timeout=2))
+            # give the holder a head start to guarantee overlap
+            time.sleep(0.05)
+            order.append("waiter-attempting")
+            lock = svc._acquire_fallback_session_lock(session_id)
+            order.append("waiter-acquired")
+            svc._release_fallback_session_lock(session_id, lock)
+
+        t_holder = threading.Thread(target=holder)
+        t_waiter = threading.Thread(target=waiter)
+        t_holder.start()
+        t_waiter.start()
+
+        # give the waiter time to reach and block on acquire()
+        time.sleep(0.3)
+        self.assertIn("waiter-attempting", order)
+        self.assertNotIn("waiter-acquired", order,
+                         "waiter must still be blocked while holder holds the lock")
+
+        release_signal.set()
+        t_holder.join(timeout=2)
+        t_waiter.join(timeout=2)
+
+        self.assertEqual(order, [
+            "holder-acquired",
+            "waiter-attempting",
+            "holder-releasing",
+            "waiter-acquired",
+        ])
+
     def test_listener_removed_on_timeout(self):
         """bus.remove must be called even when no skill replies (timeout path)."""
         svc = _make_service(config={"fallback_query_timeout": 0})
@@ -427,9 +572,12 @@ def test_listener_removed_on_timeout(self):
                    return_value=sess):
             svc._collect_fallback_skills(Message("test"), fb_range=FallbackRange(5, 90))
 
-        svc.bus.remove.assert_called_once()
-        args = svc.bus.remove.call_args[0]
-        self.assertEqual(args[0], "slow_skill.fallback.pong")
+        # DEPRECATION WINDOW: the broadcast pong collector is also bound and
+        # removed alongside the addressed one per poll round.
+        self.assertEqual(svc.bus.remove.call_count, 2)
+        removed_topics = {call.args[0] for call in svc.bus.remove.call_args_list}
+        self.assertEqual(removed_topics,
+                         {"slow_skill.fallback.pong", "ovos.skills.fallback.pong"})
 
     def test_blacklisted_skill_excluded(self):
         """Skills blacklisted by the session are not collected."""
diff --git a/test/unittests/test_intent_service_extended.py b/test/unittests/test_intent_service_extended.py
index 885c64e14b5..64b79b50850 100644
--- a/test/unittests/test_intent_service_extended.py
+++ b/test/unittests/test_intent_service_extended.py
@@ -17,7 +17,7 @@
 from unittest.mock import MagicMock, patch
 
 from ovos_bus_client.message import Message
-from ovos_bus_client.session import Session
+from ovos_bus_client.session import Session, SessionManager
 from ovos_plugin_manager.templates.pipeline import (
     IntentHandlerMatch,
     ConfidenceMatcherPipeline,
@@ -164,6 +164,44 @@ def test_lang_not_in_valid_langs_falls_through(self):
             result = IntentService.disambiguate_lang(msg)
         self.assertEqual(result, "en-US")
 
+    def test_macrolanguage_member_resolves_to_its_macrolanguage(self):
+        """A tag at the language-distance threshold resolves (arz -> ar)."""
+        for tag in ("arz", "wuu"):
+            macro = "ar" if tag == "arz" else "zh"
+            with self.subTest(tag=tag):
+                msg = Message("test", data={}, context={"stt_lang": tag})
+                with patch("ovos_core.intent_services.service.get_message_lang",
+                           return_value="en-US"), \
+                     patch("ovos_core.intent_services.service.get_valid_languages",
+                           return_value=["en-US", macro]):
+                    result = IntentService.disambiguate_lang(msg)
+                self.assertEqual(result, tag)
+
+    def test_regional_variant_resolves(self):
+        """Regional variants stay inside the threshold."""
+        for tag, supported in (("ar-SA", "ar"), ("en-AU", "en-GB"), ("pt-BR", "pt-PT")):
+            with self.subTest(tag=tag):
+                msg = Message("test", data={}, context={"stt_lang": tag})
+                with patch("ovos_core.intent_services.service.get_message_lang",
+                           return_value="en-US"), \
+                     patch("ovos_core.intent_services.service.get_valid_languages",
+                           return_value=["en-US", supported]):
+                    result = IntentService.disambiguate_lang(msg)
+                self.assertEqual(result, tag)
+
+    def test_unrelated_language_is_ignored(self):
+        """Distant languages stay outside the threshold and fall through."""
+        for tag, supported in (("zh", "en"), ("fr", "es"),
+                               ("de-CH", "fr-CH"), ("nl", "af")):
+            with self.subTest(tag=tag):
+                msg = Message("test", data={}, context={"stt_lang": tag})
+                with patch("ovos_core.intent_services.service.get_message_lang",
+                           return_value="en-US"), \
+                     patch("ovos_core.intent_services.service.get_valid_languages",
+                           return_value=[supported]):
+                    result = IntentService.disambiguate_lang(msg)
+                self.assertEqual(result, "en-US")
+
 
 # ---------------------------------------------------------------------------
 # get_pipeline_matcher
@@ -310,29 +348,6 @@ def test_blacklist_overrides_explicit_pipeline_preference(self):
         result = svc.get_pipeline(session=sess)
         self.assertEqual(result, [])
 
-    def test_plugin_blacklist_skips_all_confidence_matchers_before_lookup(self):
-        """A base plugin policy ID blocks its suffixed matcher variants.
-
-        The deployment blacklist is expressed in installed plugin IDs while a
-        session pipeline contains confidence-suffixed matcher IDs.  Filtering
-        must happen before matcher lookup so an intentionally disabled plugin
-        is neither invoked nor reported as unknown.
-        """
-        svc = self._svc_with_adapt_fallback()
-        sess = Session("s")
-        sess.pipeline = [
-            "ovos-adapt-pipeline-plugin-high",
-            "fallback_high",
-        ]
-        sess.blacklisted_pipelines = ["ovos-adapt-pipeline-plugin"]
-
-        with patch.object(svc, "get_pipeline_matcher",
-                          wraps=svc.get_pipeline_matcher) as get_matcher:
-            result = svc.get_pipeline(session=sess)
-
-        self.assertEqual([matcher[0] for matcher in result], ["fallback_high"])
-        get_matcher.assert_called_once_with("fallback_high")
-
     def test_unknown_blacklisted_id_is_harmless_noop(self):
         """Unknown pipeline_ids in blacklisted_pipelines are ignored without
         error and don't affect the effective pipeline (§5.2)."""