fix(openviking): join runtime-autostart thread on shutdown (SIGABRT-at-exit)

`OpenVikingMemoryProvider.shutdown()` joins in-flight writers, deferred-commit
threads, and prefetch threads, but not `_runtime_start_thread` — the tracked
`daemon=True` waiter that runs `_finish_runtime_openviking_start`, which blocks
on network health probes (`_wait_for_openviking_health` polling + a
`_VikingClient.health()` request).

If the local OpenViking runtime is slow or unreachable, that waiter can still
be blocked in network I/O at interpreter exit. CPython then forcibly kills it
during `Py_FinalizeEx` (`PyThread_exit_thread` -> `__pthread_unwind` ->
`abort()`), producing SIGABRT (exit 134) with no traceback — the same daemon-
thread-at-exit failure class fixed for the Honcho provider.

Fix:
- `shutdown()` now joins `_runtime_start_thread` (timeout-bounded) alongside the
  other tracked threads.
- `_wait_for_openviking_health()` gains a `should_stop` callback; the waiter
  passes `lambda: self._shutting_down` so the poll loop bails out promptly once
  `shutdown()` flips the flag, instead of lingering up to the 60s autostart
  timeout and timing out the join (which would leave the thread alive).
- Add tests/plugins/memory/test_openviking_shutdown.py covering the short-circuit
  and the shutdown-joins-runtime-thread behaviour.

(cherry picked from commit 5471ec7021)

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:

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)
 
 

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()