diff --git a/ovos_core/intent_services/fallback_service.py b/ovos_core/intent_services/fallback_service.py index dad6a4607ae..c80bc2d67da 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,18 +42,17 @@ 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) 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() @@ -64,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") @@ -84,12 +88,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 +103,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 +162,142 @@ 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 _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) + 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 != 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") - 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) + + # 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") + } + for skill_id in pool: + # 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)) + # 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( + "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,11 +324,12 @@ 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)) - 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/ovos_core/intent_services/service.py b/ovos_core/intent_services/service.py index a1deaa95102..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 diff --git a/test/end2end/test_fallback.py b/test/end2end/test_fallback.py index 51fb4d4f489..59b26a92684 100644 --- a/test/end2end/test_fallback.py +++ b/test/end2end/test_fallback.py @@ -69,22 +69,31 @@ 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, + ], entry_points=[utt_topic], final_session=final_session, - keep_original_src=[ - "ovos.skills.fallback.ping", - # "ovos.skills.fallback.pong", # TODO - ], ignore_messages=["recognizer_loop:audio_output_start", "recognizer_loop:audio_output_end"], activation_points=[f"ovos.skills.fallback.{self.skill_id}.request"], 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("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}), + {"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. @@ -95,7 +104,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 diff --git a/test/unittests/test_fallback_service.py b/test/unittests/test_fallback_service.py index 4b56ecd7c58..b7c3d8609d2 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 @@ -217,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: @@ -225,7 +230,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 +238,17 @@ def run(): svc.shutdown() self.assertIn("skill_a", result_holder[0]) + 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.""" @@ -243,7 +259,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 +282,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 +291,277 @@ 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 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 = [] + 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_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() + svc = _make_service(config={"fallback_query_timeout": 0}) svc.registered_fallbacks = {"slow_skill": 50} svc.bus.on = MagicMock() svc.bus.remove = MagicMock() @@ -285,15 +569,15 @@ 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") + # 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.""" @@ -455,6 +739,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.""" diff --git a/test/unittests/test_intent_service_extended.py b/test/unittests/test_intent_service_extended.py index 11bc4aa2ea3..64b79b50850 100644 --- a/test/unittests/test_intent_service_extended.py +++ b/test/unittests/test_intent_service_extended.py @@ -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