diff --git a/plugins/memory/openviking/__init__.py b/plugins/memory/openviking/__init__.py index 5aef3b908503..b7dd1a135511 100644 --- a/plugins/memory/openviking/__init__.py +++ b/plugins/memory/openviking/__init__.py @@ -26,6 +26,7 @@ Capabilities: from __future__ import annotations import atexit +import errno import json import logging import mimetypes @@ -42,7 +43,7 @@ import zipfile from dataclasses import dataclass, replace from pathlib import Path from typing import Any, Callable, Dict, List, Optional, Set -from urllib.parse import urlparse +from urllib.parse import quote, unquote, urlparse from urllib.request import url2pathname from agent.message_content import flatten_message_text @@ -51,6 +52,11 @@ from agent.skill_commands import extract_user_instruction_from_skill_message from tools.registry import tool_error from utils import atomic_json_write, env_var_enabled +try: + import fcntl +except ImportError: # pragma: no cover - Windows + fcntl = None + logger = logging.getLogger(__name__) _DEFAULT_ENDPOINT = "http://127.0.0.1:1933" @@ -111,6 +117,9 @@ _LOCAL_OPENVIKING_HOSTS = {"localhost", "127.0.0.1", "::1"} _LOCAL_OPENVIKING_AUTOSTART_TIMEOUT = 60.0 _OPENVIKING_SERVER_LOG_RELATIVE_PATH = Path("logs") / "openviking-server.log" _OPENVIKING_RESPONDED_FAILURE_PREFIX = "OpenViking server responded" +_PENDING_SESSIONS_RELATIVE_DIR = Path("openviking") / "pending_sessions" +_RUN_LOCKS_RELATIVE_DIR = Path("openviking") / "runs" +_LOCK_BUSY_ERRNOS = {errno.EWOULDBLOCK, errno.EACCES, errno.EAGAIN} _SETUP_CANCELLED = object() @@ -205,6 +214,11 @@ def _atexit_commit_sessions(): provider.on_session_end([]) except Exception: pass # best-effort at shutdown time + finally: + try: + provider._release_run_lock() + except Exception: + pass atexit.register(_atexit_commit_sessions) @@ -1800,6 +1814,10 @@ class OpenVikingMemoryProvider(MemoryProvider): self._agent = "" self._session_id = "" self._turn_count = 0 + self._hermes_home = "" + self._run_id = uuid.uuid4().hex + self._run_lock_file: Optional[Any] = None + self._run_lock_path: Optional[Path] = None # Guards the (_session_id, _turn_count) pair. sync_turn runs on the # MemoryManager's background sync executor while on_session_end / # on_session_switch run on the caller's thread, so the snapshot+reset @@ -2088,6 +2106,7 @@ class OpenVikingMemoryProvider(MemoryProvider): return self._client = client + self._recover_pending_sessions(self._session_id) _emit_runtime_status( f"Local OpenViking server at {endpoint} is reachable; OpenViking memory is active for later turns.", status_callback, @@ -2139,6 +2158,15 @@ class OpenVikingMemoryProvider(MemoryProvider): self._agent = settings["agent"] self._session_id = session_id self._turn_count = 0 + hermes_home = str(kwargs.get("hermes_home") or "").strip() + if not hermes_home: + try: + from hermes_constants import get_hermes_home + hermes_home = str(get_hermes_home()) + except Exception: + hermes_home = str(Path.home() / ".hermes") + self._hermes_home = hermes_home + self._acquire_run_lock() warning_callback = ( kwargs.get("warning_callback") if kwargs.get("platform") == "cli" @@ -2171,6 +2199,9 @@ class OpenVikingMemoryProvider(MemoryProvider): logger.warning("httpx not installed — OpenViking plugin disabled") self._client = None + if self._client: + self._recover_pending_sessions(session_id) + # Register as the last active provider for atexit safety net global _last_active_provider _last_active_provider = self @@ -2423,6 +2454,278 @@ class OpenVikingMemoryProvider(MemoryProvider): with self._committed_session_lock: self._committed_session_ids.add(sid) + def _pending_session_dir(self) -> Optional[Path]: + if not self._hermes_home: + return None + return Path(self._hermes_home) / _PENDING_SESSIONS_RELATIVE_DIR + + def _pending_session_marker_path(self, sid: str) -> Optional[Path]: + sid = str(sid or "").strip() + directory = self._pending_session_dir() + if not sid or directory is None: + return None + return directory / f"{quote(sid, safe='')}.json" + + def _run_lock_dir(self) -> Optional[Path]: + if not self._hermes_home: + return None + return Path(self._hermes_home) / _RUN_LOCKS_RELATIVE_DIR + + def _run_lock_path_for(self, run_id: str) -> Optional[Path]: + run_id = str(run_id or "").strip() + directory = self._run_lock_dir() + if not run_id or directory is None: + return None + return directory / f"{quote(run_id, safe='')}.lock" + + def _acquire_run_lock(self) -> None: + if self._run_lock_path is not None: + return + path = self._run_lock_path_for(self._run_id) + if path is None: + return + lock_file = None + try: + path.parent.mkdir(parents=True, exist_ok=True) + lock_file = path.open("a+", encoding="utf-8") + self._run_lock_path = path + if fcntl is None: + logger.debug("OpenViking run locks are not supported on this platform") + lock_file.close() + return + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + self._run_lock_file = lock_file + except Exception as e: + if lock_file is not None: + try: + lock_file.close() + except Exception: + pass + self._run_lock_path = None + try: + path.unlink(missing_ok=True) + except Exception: + pass + logger.debug("Could not acquire OpenViking run lock %s: %s", path, e) + + def _release_run_lock(self) -> None: + lock_file = self._run_lock_file + path = self._run_lock_path + self._run_lock_file = None + self._run_lock_path = None + if lock_file is not None: + try: + if fcntl is not None: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + except Exception as e: + logger.debug("Could not unlock OpenViking run lock %s: %s", path, e) + try: + lock_file.close() + except Exception as e: + logger.debug("Could not close OpenViking run lock %s: %s", path, e) + if path is not None: + try: + path.unlink(missing_ok=True) + except Exception as e: + logger.debug("Could not remove OpenViking run lock %s: %s", path, e) + + def _claim_owner_run_for_recovery(self, owner_run_id: str) -> tuple[bool, Optional[Any]]: + owner_run_id = str(owner_run_id or "").strip() + if not owner_run_id: + # Legacy markers predate owner_run_id; recover them without a + # liveness check so old pending sessions do not strand forever. + return True, None + if owner_run_id == self._run_id: + return False, None + path = self._run_lock_path_for(owner_run_id) + if path is None: + return False, None + if not path.exists(): + return True, None + if fcntl is None: + logger.debug( + "Skipping OpenViking pending-session recovery for owner %s; " + "advisory locks are not supported", + owner_run_id, + ) + return False, None + + lock_file = None + try: + lock_file = path.open("a+", encoding="utf-8") + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + return True, lock_file + except FileNotFoundError: + return True, None + except BlockingIOError: + if lock_file is not None: + lock_file.close() + return False, None + except OSError as e: + if lock_file is not None: + lock_file.close() + if e.errno in _LOCK_BUSY_ERRNOS: + return False, None + logger.debug( + "Skipping OpenViking pending-session recovery for owner %s; " + "could not check run lock %s: %s", + owner_run_id, + path, + e, + ) + return False, None + except Exception as e: + if lock_file is not None: + lock_file.close() + logger.debug( + "Skipping OpenViking pending-session recovery for owner %s; " + "could not check run lock %s: %s", + owner_run_id, + path, + e, + ) + return False, None + + def _release_owner_run_claim( + self, + owner_run_id: str, + lock_file: Optional[Any], + *, + cleanup: bool, + ) -> None: + if lock_file is not None: + try: + if fcntl is not None: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + except Exception: + pass + try: + lock_file.close() + except Exception: + pass + if cleanup: + self._cleanup_owner_run_lock(owner_run_id) + + def _cleanup_owner_run_lock(self, owner_run_id: str) -> None: + owner_run_id = str(owner_run_id or "").strip() + if not owner_run_id or owner_run_id == self._run_id: + return + path = self._run_lock_path_for(owner_run_id) + if path is None: + return + try: + path.unlink(missing_ok=True) + except Exception as e: + logger.debug("Could not remove OpenViking owner run lock %s: %s", path, e) + + def _mark_session_pending(self, sid: str) -> None: + if not sid or self._has_committed_session(sid): + return + path = self._pending_session_marker_path(sid) + if path is None: + return + if self._run_lock_path is None: + logger.debug("Could not safely mark OpenViking session %s pending without a run lock", sid) + return + try: + path.parent.mkdir(parents=True, exist_ok=True) + atomic_json_write( + path, + {"session_id": sid, "owner_run_id": self._run_id}, + mode=0o600, + ) + except Exception as e: + logger.debug("Could not mark OpenViking session %s pending: %s", sid, e) + + def _clear_pending_session(self, sid: str) -> None: + path = self._pending_session_marker_path(sid) + if path is None: + return + try: + path.unlink(missing_ok=True) + except Exception as e: + logger.debug("Could not clear OpenViking pending session %s: %s", sid, e) + + def _pending_sessions(self) -> List[tuple[str, str]]: + directory = self._pending_session_dir() + if directory is None or not directory.is_dir(): + return [] + sessions: List[tuple[str, str]] = [] + for path in sorted(directory.glob("*.json")): + sid = "" + owner_run_id = "" + try: + raw = json.loads(path.read_text(encoding="utf-8")) + if isinstance(raw, dict): + sid = str(raw.get("session_id") or "").strip() + owner_run_id = str(raw.get("owner_run_id") or "").strip() + except Exception: + sid = "" + sid = sid or unquote(path.stem).strip() + if sid: + sessions.append((sid, owner_run_id)) + return sessions + + def _recover_pending_sessions(self, current_sid: str) -> None: + if not self._client: + return + pending_by_owner: Dict[str, List[str]] = {} + for sid, owner_run_id in self._pending_sessions(): + pending_by_owner.setdefault(owner_run_id, []).append(sid) + + for owner_run_id, sids in pending_by_owner.items(): + recoverable, owner_lock_file = self._claim_owner_run_for_recovery(owner_run_id) + if not recoverable: + continue + + holder: List[threading.Thread] = [] + + def _recover_owner( + pending_sids: tuple = tuple(sids), + pending_owner_run_id: str = owner_run_id, + pending_owner_lock_file: Optional[Any] = owner_lock_file, + ) -> None: + try: + for pending_sid in pending_sids: + with self._deferred_commit_lock: + if self._shutting_down or pending_sid in self._deferred_commit_sids: + continue + self._deferred_commit_sids.add(pending_sid) + try: + if self._has_committed_session(pending_sid): + self._clear_pending_session(pending_sid) + continue + if self._shutting_down: + continue + self._commit_session( + pending_sid, + 0, + context="during startup recovery", + clear_missing=True, + ) + finally: + with self._deferred_commit_lock: + self._deferred_commit_sids.discard(pending_sid) + finally: + self._release_owner_run_claim( + pending_owner_run_id, + pending_owner_lock_file, + cleanup=True, + ) + with self._deferred_commit_lock: + if holder: + self._deferred_commit_threads.discard(holder[0]) + + thread = threading.Thread( + target=_recover_owner, + daemon=True, + name=f"openviking-recover-owner-{owner_run_id or 'legacy'}", + ) + holder.append(thread) + with self._deferred_commit_lock: + self._deferred_commit_threads.add(thread) + thread.start() + def _session_needs_commit(self, sid: str, turn_count: int) -> bool: # Already-committed sessions never need a second commit, regardless of # the turn counter — a racing sync_turn can re-increment _turn_count @@ -2433,16 +2736,28 @@ class OpenVikingMemoryProvider(MemoryProvider): return True return self._session_has_pending_tokens(sid) - def _commit_session(self, sid: str, turn_count: int, *, context: str) -> bool: + def _commit_session( + self, + sid: str, + turn_count: int, + *, + context: str, + clear_missing: bool = False, + ) -> bool: try: self._client.post( f"/api/v1/sessions/{sid}/commit", {"keep_recent_count": 0}, ) self._mark_session_committed(sid) + self._clear_pending_session(sid) logger.info("OpenViking session %s committed %s (%d turns)", sid, context, turn_count) return True except Exception as e: + if clear_missing and _status_code_from_error(e) == 404: + self._clear_pending_session(sid) + logger.debug("OpenViking pending session %s no longer exists; dropped marker", sid) + return False logger.warning("OpenViking session commit failed for %s: %s", sid, e) return False @@ -3105,6 +3420,8 @@ class OpenVikingMemoryProvider(MemoryProvider): return self._turn_count += 1 + self._mark_session_pending(sid) + def _sync(): def _post_turn(client: _VikingClient) -> None: if batch_messages: @@ -3343,6 +3660,7 @@ class OpenVikingMemoryProvider(MemoryProvider): global _last_active_provider if _last_active_provider is self: _last_active_provider = None + self._release_run_lock() # -- Tool implementations ------------------------------------------------ diff --git a/tests/plugins/memory/test_openviking_provider.py b/tests/plugins/memory/test_openviking_provider.py index 98a99755bfcb..65394ce149fa 100644 --- a/tests/plugins/memory/test_openviking_provider.py +++ b/tests/plugins/memory/test_openviking_provider.py @@ -2791,6 +2791,367 @@ def test_sync_turn_tracks_writer_under_session_id(): assert provider._inflight_writers.get("sid-1", set()) == set() +def test_initialize_recovers_pending_session_from_previous_process(tmp_path, monkeypatch): + _clear_openviking_env(monkeypatch) + monkeypatch.setenv("OPENVIKING_ENDPOINT", "http://test") + + posts = [] + + class StubClient: + def __init__(self, *a, **kw): + pass + + def health_payload(self): + return {"healthy": True} + + def post(self, path, payload=None, **kwargs): + posts.append((path, payload)) + return {} + + monkeypatch.setattr(openviking_module, "_VikingClient", StubClient) + + previous = OpenVikingMemoryProvider() + previous.initialize("old-sid", hermes_home=str(tmp_path)) + previous._spawn_writer = lambda sid, target, name: None + previous.sync_turn("u", "a") + previous.shutdown() + + fresh = OpenVikingMemoryProvider() + fresh.initialize("new-sid", hermes_home=str(tmp_path)) + assert fresh._drain_finalizers(timeout=2.0) + + commit = ("/api/v1/sessions/old-sid/commit", {"keep_recent_count": 0}) + assert posts.count(commit) == 1 + + later = OpenVikingMemoryProvider() + later.initialize("third-sid", hermes_home=str(tmp_path)) + assert later._drain_finalizers(timeout=2.0) + + assert posts.count(commit) == 1 + + +def test_initialize_skips_pending_session_owned_by_live_same_profile_provider(tmp_path, monkeypatch): + _clear_openviking_env(monkeypatch) + monkeypatch.setenv("OPENVIKING_ENDPOINT", "http://test") + + posts = [] + + class StubClient: + def __init__(self, *a, **kw): + pass + + def health_payload(self): + return {"healthy": True} + + def post(self, path, payload=None, **kwargs): + posts.append((path, payload)) + return {} + + monkeypatch.setattr(openviking_module, "_VikingClient", StubClient) + + live_owner = OpenVikingMemoryProvider() + live_owner.initialize("owned-sid", hermes_home=str(tmp_path)) + live_owner._spawn_writer = lambda sid, target, name: None + live_owner.sync_turn("u", "a") + marker = tmp_path / openviking_module._PENDING_SESSIONS_RELATIVE_DIR / "owned-sid.json" + assert json.loads(marker.read_text(encoding="utf-8"))["owner_run_id"] == live_owner._run_id + + other_provider = OpenVikingMemoryProvider() + other_provider.initialize("other-sid", hermes_home=str(tmp_path)) + assert other_provider._drain_finalizers(timeout=2.0) + + assert ( + "/api/v1/sessions/owned-sid/commit", + {"keep_recent_count": 0}, + ) not in posts + + live_owner.shutdown() + other_provider.shutdown() + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX advisory locks") +def test_initialize_recovers_free_owner_lock_once_and_cleans_marker(tmp_path, monkeypatch): + pytest.importorskip("fcntl") + _clear_openviking_env(monkeypatch) + monkeypatch.setenv("OPENVIKING_ENDPOINT", "http://test") + + pending_dir = tmp_path / openviking_module._PENDING_SESSIONS_RELATIVE_DIR + pending_dir.mkdir(parents=True) + marker = pending_dir / "old-sid.json" + marker.write_text( + json.dumps({"session_id": "old-sid", "owner_run_id": "dead-owner"}), + encoding="utf-8", + ) + owner_lock = tmp_path / "openviking" / "runs" / "dead-owner.lock" + owner_lock.parent.mkdir(parents=True) + owner_lock.write_text("", encoding="utf-8") + + posts = [] + + class StubClient: + def __init__(self, *a, **kw): + pass + + def health_payload(self): + return {"healthy": True} + + def post(self, path, payload=None, **kwargs): + posts.append((path, payload)) + return {} + + monkeypatch.setattr(openviking_module, "_VikingClient", StubClient) + + fresh = OpenVikingMemoryProvider() + fresh.initialize("new-sid", hermes_home=str(tmp_path)) + assert fresh._drain_finalizers(timeout=2.0) + + commit = ("/api/v1/sessions/old-sid/commit", {"keep_recent_count": 0}) + assert posts.count(commit) == 1 + assert not marker.exists() + assert not owner_lock.exists() + + later = OpenVikingMemoryProvider() + later.initialize("third-sid", hermes_home=str(tmp_path)) + assert later._drain_finalizers(timeout=2.0) + + assert posts.count(commit) == 1 + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX advisory locks") +def test_initialize_recovers_multiple_pending_sessions_for_one_dead_owner(tmp_path, monkeypatch): + import threading + + pytest.importorskip("fcntl") + _clear_openviking_env(monkeypatch) + monkeypatch.setenv("OPENVIKING_ENDPOINT", "http://test") + + pending_dir = tmp_path / openviking_module._PENDING_SESSIONS_RELATIVE_DIR + pending_dir.mkdir(parents=True) + owner_run_id = "dead-owner" + for sid in ("old-sid-a", "old-sid-b"): + (pending_dir / f"{sid}.json").write_text( + json.dumps({"session_id": sid, "owner_run_id": owner_run_id}), + encoding="utf-8", + ) + owner_lock = tmp_path / "openviking" / "runs" / f"{owner_run_id}.lock" + owner_lock.parent.mkdir(parents=True) + owner_lock.write_text("", encoding="utf-8") + + posts = [] + first_commit_entered = threading.Event() + release_commit = threading.Event() + + class StubClient: + def __init__(self, *a, **kw): + pass + + def health_payload(self): + return {"healthy": True} + + def post(self, path, payload=None, **kwargs): + posts.append((path, payload)) + if path == "/api/v1/sessions/old-sid-a/commit": + first_commit_entered.set() + release_commit.wait(timeout=5.0) + return {} + + monkeypatch.setattr(openviking_module, "_VikingClient", StubClient) + + fresh = OpenVikingMemoryProvider() + fresh.initialize("new-sid", hermes_home=str(tmp_path)) + + assert first_commit_entered.wait(timeout=2.0), "first recovery commit did not start" + release_commit.set() + assert fresh._drain_finalizers(timeout=2.0) + + assert posts.count(( + "/api/v1/sessions/old-sid-a/commit", + {"keep_recent_count": 0}, + )) == 1 + assert posts.count(( + "/api/v1/sessions/old-sid-b/commit", + {"keep_recent_count": 0}, + )) == 1 + assert not (pending_dir / "old-sid-a.json").exists() + assert not (pending_dir / "old-sid-b.json").exists() + assert not owner_lock.exists() + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX advisory locks") +def test_initialize_skips_multiple_pending_sessions_for_one_live_owner(tmp_path, monkeypatch): + import threading + + pytest.importorskip("fcntl") + _clear_openviking_env(monkeypatch) + monkeypatch.setenv("OPENVIKING_ENDPOINT", "http://test") + + commit_called = threading.Event() + release_commit = threading.Event() + posts = [] + + class StubClient: + def __init__(self, *a, **kw): + pass + + def health_payload(self): + return {"healthy": True} + + def post(self, path, payload=None, **kwargs): + posts.append((path, payload)) + if path.endswith("/commit"): + commit_called.set() + release_commit.wait(timeout=5.0) + return {} + + monkeypatch.setattr(openviking_module, "_VikingClient", StubClient) + + live_owner = OpenVikingMemoryProvider() + live_owner.initialize("owned-sid", hermes_home=str(tmp_path)) + + pending_dir = tmp_path / openviking_module._PENDING_SESSIONS_RELATIVE_DIR + pending_dir.mkdir(parents=True) + for sid in ("owned-sid-a", "owned-sid-b"): + (pending_dir / f"{sid}.json").write_text( + json.dumps({"session_id": sid, "owner_run_id": live_owner._run_id}), + encoding="utf-8", + ) + + other_provider = OpenVikingMemoryProvider() + other_provider.initialize("other-sid", hermes_home=str(tmp_path)) + assert other_provider._drain_finalizers(timeout=2.0) + + release_commit.set() + assert not commit_called.is_set() + assert ( + "/api/v1/sessions/owned-sid-a/commit", + {"keep_recent_count": 0}, + ) not in posts + assert ( + "/api/v1/sessions/owned-sid-b/commit", + {"keep_recent_count": 0}, + ) not in posts + + live_owner.shutdown() + other_provider.shutdown() + + +def test_initialize_recovers_legacy_pending_session_marker(tmp_path, monkeypatch): + _clear_openviking_env(monkeypatch) + monkeypatch.setenv("OPENVIKING_ENDPOINT", "http://test") + + pending_dir = tmp_path / openviking_module._PENDING_SESSIONS_RELATIVE_DIR + pending_dir.mkdir(parents=True) + marker = pending_dir / "legacy-sid.json" + marker.write_text(json.dumps({"session_id": "legacy-sid"}), encoding="utf-8") + + posts = [] + + class StubClient: + def __init__(self, *a, **kw): + pass + + def health_payload(self): + return {"healthy": True} + + def post(self, path, payload=None, **kwargs): + posts.append((path, payload)) + return {} + + monkeypatch.setattr(openviking_module, "_VikingClient", StubClient) + + fresh = OpenVikingMemoryProvider() + fresh.initialize("new-sid", hermes_home=str(tmp_path)) + assert fresh._drain_finalizers(timeout=2.0) + + assert posts.count(( + "/api/v1/sessions/legacy-sid/commit", + {"keep_recent_count": 0}, + )) == 1 + assert not marker.exists() + + +def test_initialize_skips_owned_pending_marker_when_fcntl_unavailable(tmp_path, monkeypatch): + _clear_openviking_env(monkeypatch) + monkeypatch.setenv("OPENVIKING_ENDPOINT", "http://test") + monkeypatch.setattr(openviking_module, "fcntl", None) + + pending_dir = tmp_path / openviking_module._PENDING_SESSIONS_RELATIVE_DIR + pending_dir.mkdir(parents=True) + marker = pending_dir / "owned-sid.json" + marker.write_text( + json.dumps({"session_id": "owned-sid", "owner_run_id": "owner-run"}), + encoding="utf-8", + ) + owner_lock = tmp_path / "openviking" / "runs" / "owner-run.lock" + owner_lock.parent.mkdir(parents=True) + owner_lock.write_text("", encoding="utf-8") + posts = [] + + class StubClient: + def __init__(self, *a, **kw): + pass + + def health_payload(self): + return {"healthy": True} + + def post(self, path, payload=None, **kwargs): + posts.append((path, payload)) + return {} + + monkeypatch.setattr(openviking_module, "_VikingClient", StubClient) + + fresh = OpenVikingMemoryProvider() + fresh.initialize("new-sid", hermes_home=str(tmp_path)) + assert fresh._drain_finalizers(timeout=2.0) + + assert ( + "/api/v1/sessions/owned-sid/commit", + {"keep_recent_count": 0}, + ) not in posts + assert marker.exists() + + +def test_initialize_recovers_pending_session_without_blocking_startup(tmp_path, monkeypatch): + import threading + + _clear_openviking_env(monkeypatch) + monkeypatch.setenv("OPENVIKING_ENDPOINT", "http://test") + + post_entered = threading.Event() + release_post = threading.Event() + + class StubClient: + def __init__(self, *a, **kw): + pass + + def health_payload(self): + return {"healthy": True} + + def post(self, path, payload=None, **kwargs): + if path == "/api/v1/sessions/old-sid/commit": + post_entered.set() + release_post.wait(timeout=5.0) + return {} + + monkeypatch.setattr(openviking_module, "_VikingClient", StubClient) + + previous = OpenVikingMemoryProvider() + previous.initialize("old-sid", hermes_home=str(tmp_path)) + previous._spawn_writer = lambda sid, target, name: None + previous.sync_turn("u", "a") + previous.shutdown() + + start = time.monotonic() + fresh = OpenVikingMemoryProvider() + fresh.initialize("new-sid", hermes_home=str(tmp_path)) + elapsed = time.monotonic() - start + + assert elapsed < 3.0, f"startup recovery blocked initialize() for {elapsed:.2f}s" + assert post_entered.wait(timeout=2.0), "recovery commit did not start" + release_post.set() + assert fresh._drain_finalizers(timeout=2.0) + + # --------------------------------------------------------------------------- # on_memory_write: explicit memory writes use content/write and stay outside # the session transcript/commit boundary.