diff --git a/plugins/memory/openviking/__init__.py b/plugins/memory/openviking/__init__.py index b2c44c2bf8c6..ec2f2772c066 100644 --- a/plugins/memory/openviking/__init__.py +++ b/plugins/memory/openviking/__init__.py @@ -1222,9 +1222,20 @@ def _start_local_openviking_server(endpoint: str) -> tuple[bool, str]: return True, f"Started openviking-server on {host}:{port} in the background. Logs: {log_path}" -def _wait_for_openviking_health(endpoint: str, *, timeout_seconds: float = 15.0) -> bool: +def _wait_for_openviking_health( + endpoint: str, + *, + timeout_seconds: float = 15.0, + should_stop=None, +) -> bool: deadline = time.monotonic() + timeout_seconds while time.monotonic() < deadline: + # Bail out promptly if the provider is being torn down, so the daemon + # thread running this waiter can be join()ed at shutdown instead of + # lingering up to ``timeout_seconds`` (a worker still alive at + # interpreter exit aborts CPython with SIGABRT at Py_FinalizeEx). + if should_stop is not None and should_stop(): + return False ok, _message = _validate_openviking_reachability(endpoint) if ok: return True @@ -2114,6 +2125,7 @@ class OpenVikingMemoryProvider(MemoryProvider): if not _wait_for_openviking_health( endpoint, timeout_seconds=_LOCAL_OPENVIKING_AUTOSTART_TIMEOUT, + should_stop=lambda: self._shutting_down, ): _emit_runtime_warning( _runtime_openviking_timeout_message(endpoint), @@ -4144,6 +4156,12 @@ class OpenVikingMemoryProvider(MemoryProvider): deferred_workers = list(self._deferred_commit_threads) with self._memory_write_lock: memory_write_workers = list(self._memory_write_threads) + # The runtime-autostart waiter is a tracked daemon thread that blocks on + # network health probes; it must be joined too, or it can be left alive + # at interpreter exit (SIGABRT at Py_FinalizeEx). Setting _shutting_down + # above makes its health-wait loop bail out promptly so the join lands. + with self._runtime_start_lock: + runtime_start_thread = self._runtime_start_thread for t in all_workers: if t.is_alive(): t.join(timeout=5.0) @@ -4153,6 +4171,8 @@ class OpenVikingMemoryProvider(MemoryProvider): for t in memory_write_workers: if t.is_alive(): t.join(timeout=5.0) + if runtime_start_thread is not None and runtime_start_thread.is_alive(): + runtime_start_thread.join(timeout=5.0) # Clear atexit reference so it doesn't double-commit. global _last_active_provider if _last_active_provider is self: diff --git a/tests/plugins/memory/test_openviking_provider.py b/tests/plugins/memory/test_openviking_provider.py index 87246fec2eb9..e7c7f666b44c 100644 --- a/tests/plugins/memory/test_openviking_provider.py +++ b/tests/plugins/memory/test_openviking_provider.py @@ -946,10 +946,14 @@ def test_runtime_openviking_waiter_attaches_client_after_health_recovers(monkeyp assert provider._client is not None assert provider._client.endpoint == "http://127.0.0.1:1934" assert provider._client.api_key == "secret" - assert wait_calls == [( - "http://127.0.0.1:1934", - {"timeout_seconds": openviking_module._LOCAL_OPENVIKING_AUTOSTART_TIMEOUT}, - )] + assert len(wait_calls) == 1 + endpoint, wait_kwargs = wait_calls[0] + assert endpoint == "http://127.0.0.1:1934" + assert wait_kwargs["timeout_seconds"] == openviking_module._LOCAL_OPENVIKING_AUTOSTART_TIMEOUT + assert callable(wait_kwargs["should_stop"]) + assert wait_kwargs["should_stop"]() is False + provider._shutting_down = True + assert wait_kwargs["should_stop"]() is True assert any("OpenViking memory is active" in message for message in statuses) diff --git a/tests/plugins/memory/test_openviking_shutdown.py b/tests/plugins/memory/test_openviking_shutdown.py new file mode 100644 index 000000000000..650d444f948d --- /dev/null +++ b/tests/plugins/memory/test_openviking_shutdown.py @@ -0,0 +1,70 @@ +"""Tests for OpenViking memory-provider shutdown teardown. + +The runtime-autostart waiter is a tracked ``daemon=True`` thread that blocks +on network health probes. If ``shutdown()`` doesn't join it (and the waiter +doesn't bail on the shutdown flag), it can be left alive at interpreter exit, +which crashes CPython with SIGABRT at ``Py_FinalizeEx``. These tests assert +the waiter short-circuits on shutdown and that ``shutdown()`` waits for the +runtime-start thread. +""" + +from __future__ import annotations + +import threading +import time +from unittest.mock import patch + +import plugins.memory.openviking as openviking_module +from plugins.memory.openviking import OpenVikingMemoryProvider + + +def test_wait_for_health_short_circuits_on_should_stop(): + """The health waiter returns False without probing when should_stop is set, + so the daemon thread running it can be join()ed promptly at shutdown.""" + probes: list[str] = [] + + def _reach(endpoint): + probes.append(endpoint) + return (False, "down") + + with patch.object( + openviking_module, "_validate_openviking_reachability", _reach + ): + result = openviking_module._wait_for_openviking_health( + "http://example.invalid", + timeout_seconds=60.0, + should_stop=lambda: True, + ) + + assert result is False + assert probes == [] # bailed before the first network probe + + +def test_shutdown_waits_for_runtime_start_thread(): + """shutdown() must join the runtime-autostart waiter thread. + + The fake waiter does post-stop work (a short sleep) once it observes the + shutdown flag. If shutdown() joins it, that work has completed by the time + shutdown() returns; without the join, shutdown() returns early and the + thread is still running (the SIGABRT-at-exit failure mode). + """ + provider = OpenVikingMemoryProvider() + started = threading.Event() + finished = threading.Event() + + def _runtime(): + started.set() + while not provider._shutting_down: + time.sleep(0.01) + time.sleep(0.2) # work that must finish during shutdown's join + finished.set() + + t = threading.Thread(target=_runtime, daemon=True, name="openviking-runtime-start") + provider._runtime_start_thread = t + t.start() + assert started.wait(2.0) + + provider.shutdown() + + assert finished.is_set() + assert not t.is_alive()