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

@@ -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

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

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 = []