diff --git a/ovos_utils/fakebus.py b/ovos_utils/fakebus.py index cc80c98f..5a9a3acc 100644 --- a/ovos_utils/fakebus.py +++ b/ovos_utils/fakebus.py @@ -1,13 +1,39 @@ import asyncio import warnings +from copy import deepcopy from os import environ from threading import Event from ovos_utils.log import LOG, log_deprecation from ovos_spec_tools import NamespaceTranslator +from ovos_spec_tools.intent_topics import (canonical_intent_topic, + intent_topic_counterpart, + is_intent_topic, + legacy_intent_topic) from pyee import EventEmitter +#: Context flag stamped on a twin intent frame, mirroring +#: ``ovos_bus_client.client.client.INTENT_COMPAT_TWIN_KEY``. Its presence +#: means the canonical spelling of this dispatch was already delivered +#: alongside it, so a receiver that understands the bridge must not +#: modernize the twin a second time (that would fire the handler twice). +INTENT_COMPAT_TWIN_KEY = "_intent_compat_twin" + + +def _verbatim_copy(message, topic: str): + """Retopic ``message`` onto ``topic``, carrying its context byte-for-byte. + + Mirrors ``ovos_bus_client.client.client._verbatim_copy``. NOT + ``Message.forward`` -- ``forward()`` re-stamps the session, which for + the default session would replace the carried session with this + process's own and desync ``lang`` / ``active_skills`` between the + canonical frame and its twin. + """ + return message.__class__(topic, data=deepcopy(message.data), + context=deepcopy(message.context)) + + def dig_for_message(): try: from ovos_bus_client.message import dig_for_message as _dig @@ -69,6 +95,7 @@ def __init__(self, *args, **kwargs): # wins, else env var -> websocket.* config -> default on. self._translator = _resolve_bus_flags(kwargs) self._handler_guards = {} # handler -> shared mirror-guard + self._intent_pair_guards = {} # frozenset({topic, counterpart}) -> shared mirror-guard self._dedup_registrations = {} # handler -> [(msg_type, wrapped), ...] self.on_open() try: @@ -80,14 +107,11 @@ def __init__(self, *args, **kwargs): self.on_default_session_update) def on(self, msg_type, handler): - # wrap handlers on migrated topics so a handler subscribed to both the - # legacy and ovos.* topic fires once (the mirror is dropped) - if self._translator.is_migrated(msg_type): - guard = self._handler_guards.get(handler) - if guard is None: - guard = self._translator.new_mirror_guard() - self._handler_guards[handler] = guard - + # wrap handlers on migrated/bridged topics so a handler subscribed to + # both spellings of a mirrored dispatch fires once (the mirror is + # dropped). See _mirror_guard_for for the guard-scope rationale. + guard = self._mirror_guard_for(msg_type, handler) + if guard is not None: def wrapped(message=None): if guard(message): return @@ -98,10 +122,70 @@ def wrapped(message=None): return self.ee.on(msg_type, handler) + def _mirror_guard_for(self, msg_type, handler): + """The mirror guard a registration on ``msg_type`` must wrap with. + + Mirrors ``ovos_bus_client.client.client.MessageBusClient._mirror_guard_for``. + Two bridges deliver one logical event twice, and each needs a + different guard SCOPE: + + - **namespace migration** (legacy <-> ``ovos.*``): the guard is per + HANDLER, shared across that handler's registrations, so its legacy + ``on()`` and its ``ovos.*`` ``on()`` dedupe against each other. + - **intent-topic compat** (canonical <-> ``.intent``-suffixed): the + guard is per TOPIC PAIR, shared by every registration on either + spelling. + + The intent guard cannot be keyed by handler: ``ovos-workshop`` + 9.3.2a1+ binds the same skill method to both spellings through a + FRESH wrapper closure per binding, so the two registrations are two + distinct ``handler`` objects and a per-handler guard would hand each + its own private state -- the canonical frame runs one closure, the + twin runs the other, and the skill handler fires twice for a single + dispatch. Keying on the pair collapses them. + """ + counterpart = intent_topic_counterpart(msg_type) + if counterpart is not None: + pair_key = frozenset({msg_type, counterpart}) + guard = self._intent_pair_guards.get(pair_key) + if guard is None: + guard = self._translator.new_mirror_guard() + self._intent_pair_guards[pair_key] = guard + return guard + if self._translator.is_migrated(msg_type): + guard = self._handler_guards.get(handler) + if guard is None: + guard = self._translator.new_mirror_guard() + self._handler_guards[handler] = guard + return guard + return None + + def _release_intent_pair_guard(self, msg_type): + """Drop the pair guard once nothing is registered on either spelling. + + Mirrors ``MessageBusClient._release_intent_pair_guard``. + """ + counterpart = intent_topic_counterpart(msg_type) + if counterpart is None: + return + pair_key = frozenset({msg_type, counterpart}) + for regs in self._dedup_registrations.values(): + if any(ev in pair_key for ev, _ in regs): + return + self._intent_pair_guards.pop(pair_key, None) + def once(self, msg_type, handler): self.ee.once(msg_type, handler) def emit(self, message): + # RULE 2 dedup marker: read it, then POP it before any local + # dispatch. A handler invoked below may call message.forward()/ + # reply() to emit a descendant frame on an UNRELATED topic; those + # deep-copy the whole context, so leaving the marker in place would + # brand that unrelated frame a twin and silently suppress its + # modernization. Mirrors MessageBusClient.on_message's pop-before- + # dispatch ordering. + is_intent_twin = message.context.pop(INTENT_COMPAT_TWIN_KEY, False) if "session" not in message.context: try: # replicate side effects from ovos_bus_client.session import Session, SessionManager @@ -140,6 +224,82 @@ def emit(self, message): self.ee.emit(topic, message.forward(topic, translated)) except Exception as e: LOG.exception(f"Error in counterpart dispatch for '{topic}': {e}") + self._bridge_intent_topic(message, is_twin=is_intent_twin) + + def _bridge_intent_topic(self, message, is_twin=False): + """Legacy <-> canonical intent-topic bridge (RULE 1 + RULE 2). + + Mirrors ``ovos_bus_client.client.client.MessageBusClient``'s + ``_send_legacy_intent_twin`` (RULE 1, send-side) and + ``_modernize_intent_topic`` (RULE 2, receive-side). The real client + splits these across the wire (twin goes out on ``emit()``, the + modernized copy is dispatched locally in ``on_message()``); a + ``FakeBus`` has no separate wire hop, so both rules run inline, + against local listeners only, off of the one message being emitted. + + ``is_twin`` carries the marker decision made in ``emit()``, which + pops :data:`INTENT_COMPAT_TWIN_KEY` off the context BEFORE any + dispatch so it cannot leak onto descendant frames a handler derives + from this one (``message.forward()``/``reply()`` deep-copy context). + The marker is therefore never read from ``message.context`` here -- + only the popped value is trusted. + + RULE 2 first, mirroring the real client's receive order, then + RULE 1 -- so a listener on the canonical topic sees the canonical + dispatch before the legacy twin goes out. + + Note on the capture firehose (deliberate FakeBus-wide convention, + shared with the namespace-migration bridge above): the twin/ + modernized copy is dispatched straight to ``self.ee`` on its own + topic and does NOT re-emit ``"message"``. On the real wire, the + twin is a second frame and therefore fires ``on_message`` (and its + ``"message"`` firehose) a second time in every receiving process; + ``FakeBus`` has no wire hop to put it on, so it keeps the + single-process harness's one-emit-one-capture invariant instead of + reproducing the wire's two-frame shape. + """ + # RULE 2 (receive-side modernize): a suffixed frame WITHOUT the twin + # marker came from an emitter old enough to only put the legacy + # spelling on the bus, so nothing canonical was sent alongside it -- + # a canonical-only listener would never hear it without this. + if self._translator.modernize and not is_twin \ + and is_intent_topic(message.msg_type): + canonical = canonical_intent_topic(message.msg_type) + if canonical != message.msg_type: + try: + self.ee.emit(canonical, _verbatim_copy(message, canonical)) + except Exception as e: + LOG.exception(f"Error in intent modernize dispatch for " + f"'{canonical}': {e}") + + # RULE 1 (send-side twin): every canonical intent dispatch is + # twinned onto its legacy spelling so a listener that only knows + # the old suffixed topic still hears it. An already-suffixed + # dispatch is never twinned (legacy_intent_topic is a no-op on it), + # so the mirror cannot cascade. + # + # On the real wire, this twin goes out MARKED (INTENT_COMPAT_TWIN_KEY), + # so an out-of-process receiver's own RULE 2 knows to skip + # re-modernizing it. FakeBus has no wire hop: RULE 2 above already + # made that call inline for THIS dispatch, so the marker's job is + # already done, and the twin delivered to local listeners here must + # NOT carry it. Carrying it forward would break two things: (a) the + # per-topic-pair mirror guard on ``on()`` fingerprints payload+context, + # so a marked twin would fingerprint differently from the canonical + # dispatch it mirrors and the guard would fail to recognize it as a + # duplicate, double-firing a dual-bound handler; and (b) a handler + # that forwards this frame's context to emit an unrelated topic would + # brand that unrelated frame a twin too (see the marker-leak + # regression test), silently suppressing its own modernization. + if self._translator.emit_legacy and is_intent_topic(message.msg_type): + topic = legacy_intent_topic(message.msg_type) + if topic != message.msg_type: + twin = _verbatim_copy(message, topic) + try: + self.ee.emit(topic, twin) + except Exception as e: + LOG.exception(f"Error in intent twin dispatch for " + f"'{topic}': {e}") def on_message(self, *args): """ @@ -239,6 +399,7 @@ def remove(self, msg_type, handler): if not regs: self._dedup_registrations.pop(handler, None) self._handler_guards.pop(handler, None) + self._release_intent_pair_guard(msg_type) return try: self.ee.remove_listener(msg_type, handler) @@ -384,6 +545,7 @@ def __init__(self, *args, **kwargs): # mirror MessageBusClient's namespace migration (see FakeBus.__init__). self._translator = _resolve_bus_flags(kwargs) self._handler_guards = {} # handler -> shared mirror-guard + self._intent_pair_guards = {} # frozenset({topic, counterpart}) -> shared mirror-guard self._dedup_registrations = {} # handler -> [(msg_type, wrapped), ...] self.connected_event = asyncio.Event() self.connected_event.set() @@ -401,15 +563,11 @@ def __init__(self, *args, **kwargs): # ------------------------------------------------------------------ def on(self, msg_type, handler): - # wrap handlers on migrated topics so a handler subscribed to both the - # legacy and ovos.* topic fires once (the mirror is dropped) -- same as - # FakeBus.on / MessageBusClient.on. - if self._translator.is_migrated(msg_type): - guard = self._handler_guards.get(handler) - if guard is None: - guard = self._translator.new_mirror_guard() - self._handler_guards[handler] = guard - + # wrap handlers on migrated/bridged topics so a handler subscribed to + # both spellings of a mirrored dispatch fires once (the mirror is + # dropped) -- same as FakeBus.on / MessageBusClient.on. + guard = self._mirror_guard_for(msg_type, handler) + if guard is not None: def wrapped(message=None): if guard(message): return @@ -420,6 +578,10 @@ def wrapped(message=None): return self.ee.on(msg_type, handler) + # shared with FakeBus -- same guard-scope rules (see FakeBus._mirror_guard_for) + _mirror_guard_for = FakeBus._mirror_guard_for + _release_intent_pair_guard = FakeBus._release_intent_pair_guard + def once(self, msg_type, handler): self.ee.once(msg_type, handler) @@ -435,6 +597,7 @@ def remove(self, msg_type, handler): if not regs: self._dedup_registrations.pop(handler, None) self._handler_guards.pop(handler, None) + self._release_intent_pair_guard(msg_type) return try: self.ee.remove_listener(msg_type, handler) @@ -466,6 +629,10 @@ async def close(self): # ------------------------------------------------------------------ async def emit(self, message): + # RULE 2 dedup marker: pop before any dispatch -- see FakeBus.emit + # for the rationale (a handler-derived forward()/reply() must not + # inherit this frame's twin marker). + is_intent_twin = message.context.pop(INTENT_COMPAT_TWIN_KEY, False) if "session" not in message.context: try: # replicate side effects from ovos_bus_client.session import Session, SessionManager @@ -493,11 +660,14 @@ async def emit(self, message): self.ee.emit(topic, message.forward(topic, translated)) except Exception as e: LOG.exception(f"Error in counterpart dispatch for '{topic}': {e}") + self._bridge_intent_topic(message, is_twin=is_intent_twin) # ------------------------------------------------------------------ # Sync helpers used internally — same as FakeBus # ------------------------------------------------------------------ + _bridge_intent_topic = FakeBus._bridge_intent_topic + def on_message(self, *args): """Handle an incoming websocket message. diff --git a/test/unittests/test_fakebus_intent_topic_bridge.py b/test/unittests/test_fakebus_intent_topic_bridge.py new file mode 100644 index 00000000..988ef084 --- /dev/null +++ b/test/unittests/test_fakebus_intent_topic_bridge.py @@ -0,0 +1,195 @@ +"""FakeBus mirrors MessageBusClient's legacy<->canonical INTENT-topic bridge +(RULE 1 send-side twin / RULE 2 receive-side modernize), so in-process tests +raw-emitting a legacy ``:IntentName.intent`` topic reach a +canonical-only listener the same way a real websocket deployment does, and +vice-versa. + +Root cause this closes: FakeBus already wired ovos_spec_tools's +NamespaceTranslator for the fixed SpecMessage pairs (speak <-> ovos.utterance.speak +etc, see test_fakebus_namespace_migration.py) but NOT the per-intent +dispatch-topic bridge that ovos_bus_client.client.client.MessageBusClient +applies via ``_send_legacy_intent_twin`` / ``_modernize_intent_topic``. Since +ovos-workshop >= 9.3.11a2 dropped its own dual-bind (only registers the +canonical listener), an in-process test emitting the legacy suffixed topic +directly never reached the handler -- while a real deployment, whose bus +client performs this bridge, dealiased fine. +""" +import asyncio +import unittest +from unittest.mock import patch + +from ovos_utils.fakebus import AsyncFakeBus, FakeBus, Message, INTENT_COMPAT_TWIN_KEY + + +def _run(coro): + return asyncio.run(coro) + + +LEGACY = "myskill.foo:HelloIntent.intent" +CANONICAL = "myskill.foo:HelloIntent" + + +class TestFakeBusIntentTopicBridge(unittest.TestCase): + def test_legacy_emit_reaches_canonical_listener(self): + # RULE 2: a raw legacy-suffixed emit (no bus-client, no twin marker) + # must still fire a canonical-only listener. + bus = FakeBus() # both flags default on + got = [] + bus.on(CANONICAL, lambda m: got.append(m.msg_type)) + bus.emit(Message(LEGACY, {"utterance": "hi"})) + self.assertEqual(got, [CANONICAL]) + + def test_canonical_emit_also_fires_legacy_listener(self): + # RULE 1: every canonical intent dispatch is twinned onto its legacy + # spelling so an old suffix-only listener still hears it. + bus = FakeBus() + got = [] + bus.on(LEGACY, lambda m: got.append(m.msg_type)) + bus.emit(Message(CANONICAL, {"utterance": "hi"})) + self.assertEqual(got, [LEGACY]) + + def test_canonical_emit_twin_not_marked_locally(self): + # On the real wire the RULE-1 twin goes out MARKED so an + # out-of-process receiver's RULE 2 knows to skip re-modernizing it. + # FakeBus has no wire hop: it already made that RULE-2 call inline + # for this dispatch, so the twin delivered to LOCAL listeners must + # NOT carry the marker -- matching the real client, whose receiving + # process pops the marker before any local handler ever sees it + # (client.py:351, before local dispatch). Carrying it into the local + # twin would also break the per-topic-pair mirror guard's + # payload+context fingerprint match (see + # test_no_double_fire_dual_listener) and leak onto any descendant + # frame a handler derives via forward()/reply(). + bus = FakeBus() + got = [] + bus.on(LEGACY, lambda m: got.append(m)) + bus.emit(Message(CANONICAL, {"utterance": "hi"})) + self.assertEqual(len(got), 1) + self.assertNotIn(INTENT_COMPAT_TWIN_KEY, got[0].context) + + def test_no_double_fire_dual_listener(self): + # a handler subscribed to BOTH the legacy and canonical topic must + # not see the same logical dispatch twice -- matching the real + # MessageBusClient, whose per-topic-pair mirror guard (shared by + # every registration on either spelling) drops the twin as a + # re-delivery of the same logical event. ovos-workshop 9.3.2a1+ + # binds the skill method to both spellings via a FRESH wrapper + # closure per registration, so this must hold even though the two + # ``bus.on()`` calls below pass the SAME underlying handler object. + bus = FakeBus() + calls = [] + handler = lambda m: calls.append(m.msg_type) + bus.on(LEGACY, handler) + bus.on(CANONICAL, handler) + bus.emit(Message(CANONICAL, {"utterance": "hi"})) + self.assertEqual(len(calls), 1) + self.assertEqual(calls, [CANONICAL]) + + def test_independent_handlers_legacy_only_starves(self): + # two INDEPENDENT handlers -- one on the canonical topic, one on the + # legacy-only spelling -- share the per-topic-pair guard (it cannot + # be scoped to a single handler, see FakeBus._mirror_guard_for), so + # a canonical emit arms the guard and the legacy-only handler + # starves. This matches real-bus behavior: a process holding both + # handlers is unreachable from a single workshop version, so the + # starvation is an accepted trade-off, not a defect. + bus = FakeBus() + canonical_calls = [] + legacy_calls = [] + bus.on(CANONICAL, lambda m: canonical_calls.append(m.msg_type)) + bus.on(LEGACY, lambda m: legacy_calls.append(m.msg_type)) + bus.emit(Message(CANONICAL, {"utterance": "hi"})) + self.assertEqual(canonical_calls, [CANONICAL]) + self.assertEqual(legacy_calls, []) + + def test_twin_marker_does_not_leak_onto_unrelated_forward(self): + # RULE 2 dedup marker regression: a handler on the LEGACY spelling + # that forwards its received message's context onto an UNRELATED + # suffixed topic must not brand that unrelated frame a twin. The + # marker is popped BEFORE dispatch (mirrors + # MessageBusClient.on_message's pop-before-dispatch ordering), so it + # cannot survive onto a descendant frame created by + # Message.forward(), which deep-copies context. + bus = FakeBus() + seen = [] + other_legacy = "other.skill:OtherIntent.intent" + other_canonical = "other.skill:OtherIntent" + + def legacy_handler(m): + bus.emit(m.forward(other_legacy, {})) + + bus.on(LEGACY, legacy_handler) + bus.on(other_canonical, lambda m: seen.append("canonical-modernized")) + bus.emit(Message(CANONICAL, {"utterance": "hi"})) + self.assertEqual(seen, ["canonical-modernized"]) + + def test_twin_marker_suppresses_rule2_recascade(self): + # if a caller manually emits a message already carrying the twin + # marker (simulating what a real client would receive as the twin + # half of a pair), RULE 2 must not modernize it again -- proving the + # bridge cannot cascade into a modernize/twin loop. + bus = FakeBus() + canonical_hits = [] + bus.on(CANONICAL, lambda m: canonical_hits.append(1)) + msg = Message(LEGACY, {"utterance": "hi"}, + {INTENT_COMPAT_TWIN_KEY: True}) + bus.emit(msg) + self.assertEqual(canonical_hits, []) + + def test_non_intent_topic_untouched(self): + bus = FakeBus() + got = [] + bus.on("my.custom.topic", lambda m: got.append(m.msg_type)) + bus.emit(Message("my.custom.topic", {"x": 1})) + self.assertEqual(got, ["my.custom.topic"]) + # and no stray listeners fired for unrelated suffixed-looking topics + got2 = [] + bus.on("speak", lambda m: got2.append(m.msg_type)) + bus.emit(Message("my.custom.topic", {"x": 1})) + self.assertEqual(got2, []) + + def test_flags_off_no_bridging(self): + # each direction gets its own bus/listener pair: a listener on the + # SAME topic as what's emitted always fires (plain same-topic + # dispatch, unrelated to the bridge) -- only the OTHER namespace's + # listener proves whether bridging happened. + bus1 = FakeBus(modernize=False, emit_legacy=False) + got_canonical = [] + bus1.on(CANONICAL, lambda m: got_canonical.append(1)) + bus1.emit(Message(LEGACY, {"utterance": "hi"})) + self.assertEqual(got_canonical, []) # RULE 2 suppressed + + bus2 = FakeBus(modernize=False, emit_legacy=False) + got_legacy = [] + bus2.on(LEGACY, lambda m: got_legacy.append(1)) + bus2.emit(Message(CANONICAL, {"utterance": "hi"})) + self.assertEqual(got_legacy, []) # RULE 1 suppressed + + def test_already_canonical_not_re_twinned_into_itself(self): + # a topic with no legacy counterpart (canonical == legacy, e.g. a + # non-suffixed topic that is not an intent topic at all) is a no-op. + bus = FakeBus() + calls = [] + bus.on(CANONICAL, lambda m: calls.append(1)) + bus.emit(Message(CANONICAL, {"utterance": "hi"})) + # exactly one direct dispatch; the RULE-1 twin went to LEGACY, not + # back onto CANONICAL, so no double count here. + self.assertEqual(calls, [1]) + + def test_async_fakebus_legacy_emit_reaches_canonical_listener(self): + bus = AsyncFakeBus() + got = [] + bus.on(CANONICAL, lambda m: got.append(m.msg_type)) + _run(bus.emit(Message(LEGACY, {"utterance": "hi"}))) + self.assertEqual(got, [CANONICAL]) + + def test_async_fakebus_canonical_emit_fires_legacy_listener(self): + bus = AsyncFakeBus() + got = [] + bus.on(LEGACY, lambda m: got.append(m.msg_type)) + _run(bus.emit(Message(CANONICAL, {"utterance": "hi"}))) + self.assertEqual(got, [LEGACY]) + + +if __name__ == "__main__": + unittest.main()