From 8dc5bf971c6eea7a44a4d8a33467963524dd0fb3 Mon Sep 17 00:00:00 2001 From: autodev Date: Fri, 28 Aug 2026 03:24:37 +0800 Subject: [PATCH 1/2] fix(hermes-adapter): widen _ACTIVE_CLIENTS key with owner_id (#2291) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When the Hermes gateway hosts multiple concurrent sessions (email threads, cron jobs, subagents) in one process, each session gets its own MemTensorProvider instance. The module-level _ACTIVE_CLIENTS singleton in bridge_client.py was keyed by (agent, no_viewer, runtime_home) — one slot per process — so every new provider's MemosBridgeClient closed whichever client held that slot, even when the slot belonged to a different, healthy session's provider. The gateway process oscillated between 2+ bridge.mjs children replaced every 2-3 seconds. Widen the singleton key to (agent, no_viewer, runtime_home, owner_id) so concurrent provider instances coexist. Callers pass owner_id=f"provider-{id(self)}"; the anonymous fallback is f"anon-{id(self)}" and stays unique for the client's lifetime. The issue #1910 guarantee (a single provider does not keep spawning bridges per turn) is preserved: re-registration under the same owner_id still reaps the previous holder. - bridge_client.py: key widened tuple[str, bool, str] -> tuple[str, bool, str, str]; owner_id: str | None added to __init__; _singleton_owner field set from owner_id or f"anon-{id(self)}"; key construction in _register_active / _unregister_active updated. - __init__.py: both construction sites (initialize() and _reconnect_bridge()) in shared and legacy modes pass owner_id=f"provider-{id(self)}". - tests/python/test_bridge_client.py: adds regression tests covering distinct-owner coexistence, same-owner reap, anonymous-fallback isolation; existing #1910 tests updated to pass explicit owner_id. Tests: 126 unit tests pass (python3 -m unittest discover -s tests/python). Related: #1910, #1927, #1985 --- .../hermes/memos_provider/__init__.py | 10 +- .../hermes/memos_provider/bridge_client.py | 34 ++++-- .../tests/python/test_bridge_client.py | 107 +++++++++++++++++- 3 files changed, 134 insertions(+), 17 deletions(-) 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..ac10a2997 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 @@ -38,15 +38,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 +156,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 +264,21 @@ 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) + 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 +296,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 +309,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..c80ef982d 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.assertIsNot(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() From 38790db8ce2591a456083d5310cac09a64aa000a Mon Sep 17 00:00:00 2001 From: autodev-bot Date: Fri, 28 Aug 2026 03:35:55 +0800 Subject: [PATCH 2/2] fix(hermes-adapter): warn on anonymous owner_id and tighten singleton-key test Address open-code-review findings on PR #2293: * bridge_client.py: raise a warnings.warn when MemosBridgeClient is constructed without owner_id. The anon- fallback is unique per instance, which silently disables the #1910 reap-previous guard. Warning forces callers to be explicit about opting out. * test_bridge_client.py: replace assertIsNot(key_first, key_second) with assertNotEqual. Two freshly constructed tuples are always distinct objects in CPython, so the identity check was vacuously true and would still pass if both keys carried identical values. Value comparison is what the regression test needs. --- .../adapters/hermes/memos_provider/bridge_client.py | 12 ++++++++++++ .../tests/python/test_bridge_client.py | 2 +- 2 files changed, 13 insertions(+), 1 deletion(-) 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 ac10a2997..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 @@ -278,6 +279,17 @@ def __init__( 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: 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 c80ef982d..3ee0773bf 100644 --- a/apps/memos-local-plugin/tests/python/test_bridge_client.py +++ b/apps/memos-local-plugin/tests/python/test_bridge_client.py @@ -409,7 +409,7 @@ def test_module_singleton_isolated_for_distinct_owners(self) -> None: second._singleton_runtime_home, second._singleton_owner, ) - self.assertIsNot(key_first, key_second) + 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)