diff --git a/apps/memos-local-plugin/adapters/hermes/memos_provider/__init__.py b/apps/memos-local-plugin/adapters/hermes/memos_provider/__init__.py index e2389e498..ae1c02611 100644 --- a/apps/memos-local-plugin/adapters/hermes/memos_provider/__init__.py +++ b/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)", diff --git a/apps/memos-local-plugin/adapters/hermes/memos_provider/bridge_client.py b/apps/memos-local-plugin/adapters/hermes/memos_provider/bridge_client.py index e137729ed..b125d5014 100644 --- a/apps/memos-local-plugin/adapters/hermes/memos_provider/bridge_client.py +++ b/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-`` 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-`` 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: diff --git a/apps/memos-local-plugin/tests/python/test_bridge_client.py b/apps/memos-local-plugin/tests/python/test_bridge_client.py index 5ec1aabe4..3ee0773bf 100644 --- a/apps/memos-local-plugin/tests/python/test_bridge_client.py +++ b/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,10 +370,101 @@ 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() @@ -378,6 +472,7 @@ def test_close_unregisters_active_client_only_when_still_current(self) -> None: 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()