Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 53 additions & 15 deletions plugins/memory/openviking/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -1179,9 +1179,21 @@ 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 daemon thread still alive at
# interpreter exit causes CPython to abort during Py_FinalizeEx teardown;
# gh-97940).
if should_stop is not None and should_stop():
return False
ok, _message = _validate_openviking_reachability(endpoint)
if ok:
return True
Expand Down Expand Up @@ -2054,11 +2066,15 @@ def _finish_runtime_openviking_start(
if not _wait_for_openviking_health(
endpoint,
timeout_seconds=_LOCAL_OPENVIKING_AUTOSTART_TIMEOUT,
should_stop=lambda: self._shutting_down,

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This new keyword breaks the existing exact-kwargs assertion in tests/plugins/memory/test_openviking_provider.py:915-918, which currently expects only timeout_seconds. Update that test to expect and exercise the callback.

):
Comment on lines 2066 to 2070
_emit_runtime_warning(
_runtime_openviking_timeout_message(endpoint),
warning_callback,
)
# Suppress the timeout warning during clean shutdown —
# should_stop made the health-wait bail out, not a real timeout.
if not self._shutting_down:
_emit_runtime_warning(
_runtime_openviking_timeout_message(endpoint),
warning_callback,
)
return

try:
Expand All @@ -2070,21 +2086,24 @@ def _finish_runtime_openviking_start(
agent=self._agent,
)
if not client.health():
_emit_runtime_warning(
f"OpenViking server at {endpoint} is still not reachable after auto-start; "
"OpenViking memory disabled for this Hermes run.",
warning_callback,
)
# Suppress health-check failure warning during clean shutdown.
if not self._shutting_down:
_emit_runtime_warning(
f"OpenViking server at {endpoint} is still not reachable after auto-start; "
"OpenViking memory disabled for this Hermes run.",
warning_callback,
)
return
except ImportError:
logger.warning("httpx not installed — OpenViking plugin disabled")
return
except Exception as e:
_emit_runtime_warning(
f"OpenViking server at {endpoint} could not be attached after auto-start: {e}. "
"OpenViking memory disabled for this Hermes run.",
warning_callback,
)
if not self._shutting_down:
_emit_runtime_warning(
f"OpenViking server at {endpoint} could not be attached after auto-start: {e}. "
"OpenViking memory disabled for this Hermes run.",
warning_callback,
)
return

self._client = client
Expand Down Expand Up @@ -3330,6 +3349,15 @@ def shutdown(self) -> None:
deferred_workers = list(self._deferred_commit_threads)

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

OpenVikingMemoryProvider has no _prefetch_lock or _prefetch_threads field (verified across the current provider), so this makes every shutdown() raise AttributeError. Please remove this unrelated block, or add the actual tracked prefetch state and its producer if prefetch is intended here.

with self._memory_write_lock:
memory_write_workers = list(self._memory_write_threads)
with self._prefetch_lock:
prefetch_workers = list(self._prefetch_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 (CPython aborts during Py_FinalizeEx teardown when
# daemon threads are still active; gh-97940). 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)
Expand All @@ -3339,6 +3367,16 @@ def shutdown(self) -> None:
for t in memory_write_workers:
if t.is_alive():
t.join(timeout=5.0)
for t in prefetch_workers:
if t.is_alive():
t.join(timeout=5.0)
if runtime_start_thread is not None and runtime_start_thread.is_alive():
# Residual window: should_stop covers the poll loop, but if the
# thread is in the final _VikingClient.health() HTTP call when
# shutdown lands, this join may time out (~5s). Much smaller than
# the original 60s poll window. Fully closing the Viking client
# here (like the Honcho httpx pattern) would close this gap too.
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:
Expand Down
70 changes: 70 additions & 0 deletions tests/plugins/memory/test_openviking_shutdown.py
Original file line number Diff line number Diff line change
@@ -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()