fix(openviking): recover pending session commits

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

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.