From c4d0f1c1d6b6bbe57d852b5d4ee13bccd1b43a80 Mon Sep 17 00:00:00 2001 From: Hao Zhe Date: Tue, 14 Jul 2026 12:46:21 +0800 Subject: [PATCH] fix(openviking): serialize local runtime recovery starts Avoid spawning multiple local OpenViking server processes while a runtime autostart waiter is already active. Remote endpoints still retry on later accesses because they do not install a local waiter. --- plugins/memory/openviking/__init__.py | 88 +++++++++----- tests/openviking_plugin/test_openviking.py | 109 +++++++++++++++++- .../memory/test_openviking_provider.py | 33 ++++++ 3 files changed, 200 insertions(+), 30 deletions(-) diff --git a/plugins/memory/openviking/__init__.py b/plugins/memory/openviking/__init__.py index 193013a6559..68f0f8f9eac 100644 --- a/plugins/memory/openviking/__init__.py +++ b/plugins/memory/openviking/__init__.py @@ -1894,6 +1894,7 @@ class OpenVikingMemoryProvider(MemoryProvider): self._pending_marked_sids: Set[str] = set() self._runtime_start_lock = threading.Lock() self._runtime_start_thread: Optional[threading.Thread] = None + self._runtime_start_pending = False self._memory_write_lock = threading.Lock() self._memory_write_threads: Set[threading.Thread] = set() self._profile_prefetched_sessions: Set[str] = set() @@ -2112,19 +2113,20 @@ class OpenVikingMemoryProvider(MemoryProvider): status_callback=None, warning_callback=None, ) -> None: - with self._runtime_start_lock: - if self._runtime_start_thread and self._runtime_start_thread.is_alive(): - return - self._runtime_start_thread = threading.Thread( - target=self._finish_runtime_openviking_start, - kwargs={ - "status_callback": status_callback, - "warning_callback": warning_callback, - }, - daemon=True, - name="openviking-runtime-start", - ) - self._runtime_start_thread.start() + # Precondition: caller holds _runtime_start_lock. Local process start + # ownership is reserved with _runtime_start_pending before callbacks run. + if self._runtime_start_thread and self._runtime_start_thread.is_alive(): + return + self._runtime_start_thread = threading.Thread( + target=self._finish_runtime_openviking_start, + kwargs={ + "status_callback": status_callback, + "warning_callback": warning_callback, + }, + daemon=True, + name="openviking-runtime-start", + ) + self._runtime_start_thread.start() def _finish_runtime_openviking_start( self, @@ -2194,25 +2196,48 @@ class OpenVikingMemoryProvider(MemoryProvider): self._client = None return - started, start_message = _start_local_openviking_server(endpoint) - if not started: + warning_message = "" + status_message = "" + should_start_waiter = False + with self._runtime_start_lock: + if ( + self._runtime_start_pending + or (self._runtime_start_thread and self._runtime_start_thread.is_alive()) + ): + self._client = None + return + + self._runtime_start_pending = True + started, start_message = _start_local_openviking_server(endpoint) + if not started: + self._runtime_start_pending = False + warning_message = ( + f"Local OpenViking server at {endpoint} is not reachable. {start_message} " + "OpenViking memory disabled for this Hermes run." + ) + self._client = None + else: + self._client = None + status_message = ( + f"{start_message} OpenViking memory is starting in the background and will attach when ready." + ) + should_start_waiter = True + + if warning_message: _emit_runtime_warning( - f"Local OpenViking server at {endpoint} is not reachable. {start_message} " - "OpenViking memory disabled for this Hermes run.", + warning_message, warning_callback, ) - self._client = None return - - self._client = None - _emit_runtime_status( - f"{start_message} OpenViking memory is starting in the background and will attach when ready.", - status_callback, - ) - self._start_runtime_openviking_waiter( - status_callback=status_callback, - warning_callback=warning_callback, - ) + if status_message: + _emit_runtime_status(status_message, status_callback) + if should_start_waiter: + with self._runtime_start_lock: + self._runtime_start_pending = False + self._start_runtime_openviking_waiter( + status_callback=status_callback, + warning_callback=warning_callback, + ) def initialize(self, session_id: str, **kwargs) -> None: settings = _resolve_connection_settings(_load_hermes_openviking_config()) @@ -2309,6 +2334,13 @@ class OpenVikingMemoryProvider(MemoryProvider): ) if config_unchanged and self._client is not None: return self._client + if config_unchanged: + with self._runtime_start_lock: + if ( + self._runtime_start_pending + or (self._runtime_start_thread and self._runtime_start_thread.is_alive()) + ): + return self._client self._endpoint = endpoint self._api_key = api_key diff --git a/tests/openviking_plugin/test_openviking.py b/tests/openviking_plugin/test_openviking.py index b71f0305312..700514aed97 100644 --- a/tests/openviking_plugin/test_openviking.py +++ b/tests/openviking_plugin/test_openviking.py @@ -2,6 +2,7 @@ import json import threading +import time from http.server import BaseHTTPRequestHandler, HTTPServer from typing import Any, cast from urllib.parse import parse_qs, urlparse @@ -1630,13 +1631,15 @@ class TestEnsureClientReloadsEnv: ): from hermes_cli import config as hermes_config - for key in ( + known_hermes_env = set(hermes_config.OPTIONAL_ENV_VARS) | hermes_config._EXTRA_ENV_KEYS + openviking_tenant_env = { "OPENVIKING_ENDPOINT", "OPENVIKING_API_KEY", "OPENVIKING_ACCOUNT", "OPENVIKING_USER", "OPENVIKING_AGENT", - ): + } + for key in known_hermes_env | openviking_tenant_env: monkeypatch.delenv(key, raising=False) hermes_home = tmp_path / "hermes-home" @@ -1701,6 +1704,108 @@ class TestEnsureClientReloadsEnv: assert start_calls == ["http://127.0.0.1:31933"] assert len(waiter_calls) == 1 + def test_repeated_access_while_local_runtime_starts_does_not_spawn_again(self, monkeypatch): + class _AliveThread: + def is_alive(self): + return True + + class _StubClient: + def __init__(self, endpoint, api_key="", account="", user="", agent=""): + self.endpoint = endpoint + + def health(self): + return False + + monkeypatch.setattr(openviking_plugin, "_VikingClient", _StubClient) + monkeypatch.setenv("OPENVIKING_ENDPOINT", "http://127.0.0.1:31933") + monkeypatch.setenv("OPENVIKING_API_KEY", "") + + start_calls = [] + provider = OpenVikingMemoryProvider() + provider._env_refresh_enabled = True + monkeypatch.setattr( + openviking_plugin, + "_start_local_openviking_server", + lambda endpoint: start_calls.append(endpoint) or (True, "started"), + ) + monkeypatch.setattr( + provider, + "_start_runtime_openviking_waiter", + lambda **kwargs: setattr(provider, "_runtime_start_thread", _AliveThread()), + raising=False, + ) + + assert provider._ensure_client() is None + assert provider._ensure_client() is None + + assert start_calls == ["http://127.0.0.1:31933"] + + def test_concurrent_local_runtime_recovery_starts_once(self, monkeypatch): + class _AliveThread: + def is_alive(self): + return True + + health_barrier = threading.Barrier(2) + + class _StubClient: + def __init__(self, endpoint, api_key="", account="", user="", agent=""): + self.endpoint = endpoint + + def health(self): + health_barrier.wait(timeout=2) + return False + + monkeypatch.setattr(openviking_plugin, "_VikingClient", _StubClient) + monkeypatch.setenv("OPENVIKING_ENDPOINT", "http://127.0.0.1:31933") + monkeypatch.setenv("OPENVIKING_API_KEY", "") + + provider = OpenVikingMemoryProvider() + provider._env_refresh_enabled = True + start_calls = [] + start_lock = threading.Lock() + first_start_entered = threading.Event() + release_start = threading.Event() + + def start_local(endpoint): + with start_lock: + start_calls.append(endpoint) + first_start_entered.set() + release_start.wait(timeout=2) + return True, "started" + + monkeypatch.setattr(openviking_plugin, "_start_local_openviking_server", start_local) + monkeypatch.setattr( + provider, + "_start_runtime_openviking_waiter", + lambda **kwargs: setattr(provider, "_runtime_start_thread", _AliveThread()), + raising=False, + ) + + errors = [] + + def access_client(): + try: + provider._ensure_client() + except BaseException as exc: # pragma: no cover - surfaced below + errors.append(exc) + + threads = [ + threading.Thread(target=access_client, name=f"openviking-access-{index}") + for index in range(2) + ] + for thread in threads: + thread.start() + + assert first_start_entered.wait(timeout=2) + time.sleep(0.05) + release_start.set() + for thread in threads: + thread.join(timeout=2) + assert not thread.is_alive() + + assert errors == [] + assert start_calls == ["http://127.0.0.1:31933"] + def test_handle_tool_call_uses_ensure_client(self, monkeypatch): provider = OpenVikingMemoryProvider() provider._env_refresh_enabled = True diff --git a/tests/plugins/memory/test_openviking_provider.py b/tests/plugins/memory/test_openviking_provider.py index 75a221fa81f..2ef2a2686c7 100644 --- a/tests/plugins/memory/test_openviking_provider.py +++ b/tests/plugins/memory/test_openviking_provider.py @@ -953,6 +953,39 @@ def test_initialize_autostarts_local_openviking_in_background_when_runtime_healt assert any("starting in the background" in message for message in statuses) +def test_initialize_emits_starting_status_before_runtime_waiter_can_attach(monkeypatch): + _clear_openviking_env(monkeypatch) + monkeypatch.setenv("OPENVIKING_ENDPOINT", "http://127.0.0.1:1934") + + class FakeVikingClient: + def __init__(self, endpoint, api_key="", account="", user="", agent=""): + assert endpoint == "http://127.0.0.1:1934" + + def health(self): + return False + + monkeypatch.setattr(openviking_module, "_VikingClient", FakeVikingClient) + monkeypatch.setattr( + openviking_module, + "_start_local_openviking_server", + lambda endpoint: (True, "started"), + ) + + provider = OpenVikingMemoryProvider() + statuses = [] + + def start_waiter(*, status_callback=None, warning_callback=None): + assert callable(status_callback) + status_callback("Local OpenViking server is reachable; OpenViking memory is active.") + + monkeypatch.setattr(provider, "_start_runtime_openviking_waiter", start_waiter, raising=False) + + provider.initialize("session-1", platform="cli", status_callback=statuses.append) + + assert "starting in the background" in statuses[0] + assert "memory is active" in statuses[1] + + def test_runtime_openviking_waiter_attaches_client_after_health_recovers(monkeypatch): _clear_openviking_env(monkeypatch) wait_calls = []