Fix #2291: fix(hermes-adapter): widen _ACTIVE_CLIENTS key with owner_id to prevent intra-pr

apps/memos-local-plugin/adapters/hermes/memos_provider/__init__.py

@@ -507,13 +507,15 @@ def initialize(self, session_id: str, **kwargs: Any) -> None:  # type: ignore[ov
         try:
             runtime_home = self._runtime_home
             runtime_env = dict(self._runtime_env)
+            owner_id = f"provider-{id(self)}"
             if self._shared_bridge:
                 new_bridge = SHARED_BRIDGE_REGISTRY.acquire(
                     _shared_bridge_runtime_key(runtime_home),
-                    client_factory=lambda home=str(runtime_home), env=runtime_env: (
+                    client_factory=lambda home=str(runtime_home), env=runtime_env, owner=owner_id: (
                         MemosBridgeClient(
                             runtime_home=home,
                             extra_env=env,
+                            owner_id=owner,
                         )
                     ),
                     before_spawn=lambda home=runtime_home: _prepare_shared_bridge(home),
@@ -526,6 +528,7 @@ def initialize(self, session_id: str, **kwargs: Any) -> None:  # type: ignore[ov
                 new_bridge = MemosBridgeClient(
                     runtime_home=str(runtime_home),
                     extra_env=runtime_env,
+                    owner_id=owner_id,
                 )
                 new_bridge.register_host_handler(
                     "host.llm.complete",
@@ -2189,12 +2192,14 @@ def _reconnect_bridge(self, session_id: str = "", *, timeout: float = 30.0) -> N
                         self._runtime_env = _memos_runtime_env_snapshot(self._runtime_home)
                     runtime_home = self._runtime_home
                     runtime_env = dict(self._runtime_env)
+                    owner_id = f"provider-{id(self)}"
                     bridge = SHARED_BRIDGE_REGISTRY.acquire(
                         _shared_bridge_runtime_key(runtime_home),
-                        client_factory=lambda home=str(runtime_home), env=runtime_env: (
+                        client_factory=lambda home=str(runtime_home), env=runtime_env, owner=owner_id: (
                             MemosBridgeClient(
                                 runtime_home=home,
                                 extra_env=env,
+                                owner_id=owner,
                             )
                         ),
                         before_spawn=lambda home=runtime_home: _prepare_shared_bridge(home),
@@ -2261,6 +2266,7 @@ def _reconnect_bridge(self, session_id: str = "", *, timeout: float = 30.0) -> N
                 new_bridge = MemosBridgeClient(
                     runtime_home=str(runtime_home),
                     extra_env=runtime_env,
+                    owner_id=f"provider-{id(self)}",
                 )
                 logger.info(
                     "MemOS: new bridge created (pid=%s)",

apps/memos-local-plugin/adapters/hermes/memos_provider/bridge_client.py

@@ -21,6 +21,7 @@
 import shutil
 import subprocess
 import threading
+import warnings
 
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
@@ -38,15 +39,21 @@
 HOST_HANDLER_QUEUE_CAPACITY = 16
 
 # ─── Module-level singleton tracker ─────────────────────────────────────
-# Each entry maps an ``(agent, no_viewer, runtime_home)`` key to the
-# most-recent active ``MemosBridgeClient`` for that slot. When a new client
-# is constructed for an existing key, the previous client is closed
+# Each entry maps an ``(agent, no_viewer, runtime_home, owner_id)`` key to
+# the most-recent active ``MemosBridgeClient`` for that slot. When a new
+# client is constructed for an existing key, the previous client is closed
 # synchronously so the Node-side ``bridge.cjs`` subprocess does not leak.
 #
 # This is the Python-side guard against issue #1910 (bridge process leak:
 # every turn spawns new bridge.cjs). Defence in depth on the Node side
 # lives in ``bridge.cts`` via ``bridge-stdio.pid``.
-_ACTIVE_CLIENTS: dict[tuple[str, bool, str], MemosBridgeClient] = {}
+#
+# The ``owner_id`` component (issue #2291) lets multiple ``MemTensorProvider``
+# instances in one Hermes gateway process each own a distinct slot instead
+# of fighting for the single per-process ``(agent, no_viewer, runtime_home)``
+# entry. Callers that do not pass ``owner_id`` fall back to
+# ``f"anon-{id(self)}"`` which is unique for the client's lifetime.
+_ACTIVE_CLIENTS: dict[tuple[str, bool, str, str], MemosBridgeClient] = {}
 _ACTIVE_CLIENTS_LOCK = threading.Lock()
 
 
@@ -150,6 +157,7 @@ def __init__(
         no_viewer: bool = True,
         extra_env: dict[str, str] | None = None,
         runtime_home: str | None = None,
+        owner_id: str | None = None,
     ) -> None:
         self._lock = threading.Lock()
         self._next_id = 1
@@ -257,14 +265,32 @@ def __init__(
         self._stderr_reader.start()
 
         # Singleton tracking (issue #1910). Register ourselves as the
-        # active client for ``(agent, no_viewer, runtime_home)`` and reap
-        # any previous holder synchronously so its subprocess does not leak.
-        # The reap happens AFTER our reader threads are running, so the
-        # previous client's ``close()`` (which closes stdin and waits for
-        # exit) cannot interfere with our own startup.
+        # active client for ``(agent, no_viewer, runtime_home, owner_id)``
+        # and reap any previous holder synchronously so its subprocess does
+        # not leak. The reap happens AFTER our reader threads are running,
+        # so the previous client's ``close()`` (which closes stdin and
+        # waits for exit) cannot interfere with our own startup.
+        #
+        # The ``owner_id`` component (issue #2291) widens the slot so
+        # concurrent ``MemTensorProvider`` instances in one Hermes gateway
+        # process each keep their own bridge instead of reaping each other.
+        # A ``None`` value falls back to ``anon-<id>`` which is unique for
+        # the client's lifetime.
         self._singleton_agent = agent
         self._singleton_no_viewer = bool(no_viewer)
         self._singleton_runtime_home = str(resolved_runtime_home)
+        if owner_id is None:
+            # Warn callers who silently opt out of the singleton reap-previous
+            # mechanism (issue #1910). ``anon-<id>`` guarantees uniqueness per
+            # instance, so no two anonymous clients ever share a slot and the
+            # leak guard becomes a no-op for them. Force explicit intent.
+            warnings.warn(
+                "MemosBridgeClient created without owner_id; "
+                "the singleton reap-previous mechanism is disabled for this instance. "
+                "Pass an explicit owner_id to prevent bridge process leaks.",
+                stacklevel=2,
+            )
+        self._singleton_owner = owner_id or f"anon-{id(self)}"
         previous = self._register_active()
         if previous is not None and previous is not self:
             prev_pid = getattr(previous, "pid", "?")
@@ -282,6 +308,7 @@ def _register_active(self) -> MemosBridgeClient | None:
             self._singleton_agent,
             self._singleton_no_viewer,
             self._singleton_runtime_home,
+            self._singleton_owner,
         )
         with _ACTIVE_CLIENTS_LOCK:
             previous = _ACTIVE_CLIENTS.get(key)
@@ -294,6 +321,7 @@ def _unregister_active(self) -> None:
             self._singleton_agent,
             self._singleton_no_viewer,
             self._singleton_runtime_home,
+            self._singleton_owner,
         )
         with _ACTIVE_CLIENTS_LOCK:
             if _ACTIVE_CLIENTS.get(key) is self:

apps/memos-local-plugin/tests/python/test_bridge_client.py

@@ -324,17 +324,20 @@ def test_close_is_idempotent(self) -> None:
         client.close()  # second call must not raise
 
     def test_module_singleton_closes_previous_client_same_agent(self) -> None:
-        """Constructing a second client with the same agent must reap the first.
+        """Constructing a second client with the same owner must reap the first.
 
         Regression for issue #1910: each turn the Hermes adapter could
         spawn a fresh bridge subprocess without closing its predecessor,
         accumulating 4+ processes per session. The singleton tracker in
         ``MemosBridgeClient`` prevents that by closing any active client
-        for the same ``(agent, no_viewer)`` slot at construction time.
+        for the same ``(agent, no_viewer, runtime_home, owner_id)`` slot
+        at construction time. Issue #2291 widened the key with
+        ``owner_id`` so this test now pins the same-owner replacement
+        contract.
         """
-        first = MemosBridgeClient(bridge_path="/tmp/bridge.cts")
+        first = MemosBridgeClient(bridge_path="/tmp/bridge.cts", owner_id="same")
         self.assertFalse(first._closed)
-        second = MemosBridgeClient(bridge_path="/tmp/bridge.cts")
+        second = MemosBridgeClient(bridge_path="/tmp/bridge.cts", owner_id="same")
         # The new constructor must have reaped the previous one.
         self.assertTrue(first._closed)
         self.assertFalse(second._closed)
@@ -367,17 +370,109 @@ def test_module_singleton_isolated_for_distinct_runtime_homes(self) -> None:
             first.close()
             second.close()
 
+    def test_module_singleton_isolated_for_distinct_owners(self) -> None:
+        """Distinct ``owner_id`` values must not reap each other (issue #2291).
+
+        Regression for the intra-process bridge fight: the Hermes gateway
+        creates one ``MemTensorProvider`` per concurrent session (email,
+        cron, subagents). Every provider construction used to close the
+        one active client bound to ``(agent, no_viewer, runtime_home)``
+        even when that client belonged to a healthy sibling. With
+        ``owner_id`` widening the singleton key, N providers coexist
+        instead of fighting.
+        """
+        with tempfile.TemporaryDirectory() as root:
+            runtime = str(Path(root) / "shared-home")
+            first = MemosBridgeClient(
+                bridge_path="/tmp/bridge.cts",
+                runtime_home=runtime,
+                owner_id="provider-1",
+            )
+            second = MemosBridgeClient(
+                bridge_path="/tmp/bridge.cts",
+                runtime_home=runtime,
+                owner_id="provider-2",
+            )
+
+            self.assertFalse(first._closed)
+            self.assertFalse(second._closed)
+
+            key_first = (
+                first._singleton_agent,
+                first._singleton_no_viewer,
+                first._singleton_runtime_home,
+                first._singleton_owner,
+            )
+            key_second = (
+                second._singleton_agent,
+                second._singleton_no_viewer,
+                second._singleton_runtime_home,
+                second._singleton_owner,
+            )
+            self.assertNotEqual(key_first, key_second)
+            self.assertIs(bridge_client_mod._ACTIVE_CLIENTS.get(key_first), first)
+            self.assertIs(bridge_client_mod._ACTIVE_CLIENTS.get(key_second), second)
+
+            first.close()
+            second.close()
+
+    def test_module_singleton_same_owner_replacement_still_reaps_previous(self) -> None:
+        """Same ``owner_id`` re-registration must close its predecessor.
+
+        The #1910 guarantee (a single provider does not keep spawning new
+        bridges without closing the previous one) still holds because a
+        client rebuilt under the same owner slot displaces its predecessor.
+        """
+        with tempfile.TemporaryDirectory() as root:
+            runtime = str(Path(root) / "same-home")
+            first = MemosBridgeClient(
+                bridge_path="/tmp/bridge.cts",
+                runtime_home=runtime,
+                owner_id="provider-same",
+            )
+            second = MemosBridgeClient(
+                bridge_path="/tmp/bridge.cts",
+                runtime_home=runtime,
+                owner_id="provider-same",
+            )
+
+            self.assertTrue(first._closed)
+            self.assertFalse(second._closed)
+            second.close()
+
+    def test_module_singleton_anonymous_fallback_isolates_distinct_instances(self) -> None:
+        """Anonymous fallback keys stay unique per client instance (issue #2291)."""
+        with tempfile.TemporaryDirectory() as root:
+            runtime = str(Path(root) / "anon-home")
+            first = MemosBridgeClient(
+                bridge_path="/tmp/bridge.cts",
+                runtime_home=runtime,
+            )
+            second = MemosBridgeClient(
+                bridge_path="/tmp/bridge.cts",
+                runtime_home=runtime,
+            )
+
+            self.assertFalse(first._closed)
+            self.assertFalse(second._closed)
+            self.assertTrue(first._singleton_owner.startswith("anon-"))
+            self.assertTrue(second._singleton_owner.startswith("anon-"))
+            self.assertNotEqual(first._singleton_owner, second._singleton_owner)
+            first.close()
+            second.close()
+
     def test_close_unregisters_active_client_only_when_still_current(self) -> None:
         """A stale close() must not evict the newer registered client."""
-        first = MemosBridgeClient(bridge_path="/tmp/bridge.cts")
-        second = MemosBridgeClient(bridge_path="/tmp/bridge.cts")
+        first = MemosBridgeClient(bridge_path="/tmp/bridge.cts", owner_id="stale-owner")
+        second = MemosBridgeClient(bridge_path="/tmp/bridge.cts", owner_id="stale-owner")
         # First was already closed by second's __init__. Closing it again is
         # a no-op and must not touch the registry's current entry (second).
         first.close()
         key = (
             second._singleton_agent,
             second._singleton_no_viewer,
             second._singleton_runtime_home,
+            second._singleton_owner,
         )
         self.assertIs(bridge_client_mod._ACTIVE_CLIENTS.get(key), second)
         second.close()