fix: give FakeBus the intent-topic bridge (RULE 1/RULE 2 parity with MessageBusClient)

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.