fix(openviking): re-arm the commit guard after in-place compression (#74695)

plugins/memory/openviking/__init__.py

@@ -2716,6 +2716,18 @@ def _mark_session_committed(self, sid: str) -> None:
         with self._committed_session_lock:
             self._committed_session_ids.add(sid)
 
+    def _clear_session_committed(self, sid: str) -> None:
+        """Re-arm the commit guard for a session that is still live.
+
+        A permanent per-sid latch is correct for a session being left behind:
+        it dedupes that id's ``_finalize_session_async`` against the commit
+        compression already performed. In-place compression keeps the *same*
+        id, so the latch would otherwise reject every later commit for a
+        session that is still accumulating turns (#74695).
+        """
+        with self._committed_session_lock:
+            self._committed_session_ids.discard(sid)
+
     def _pending_session_dir(self) -> Optional[Path]:
         if not self._hermes_home:
             return None
@@ -4182,6 +4194,20 @@ def on_session_switch(
             self._profile_prefetched_sessions.discard(old_session_id)
             self._profile_prefetched_sessions.discard(new_id)
 
+            if not rotate and old_session_id:
+                # In-place compression (the default) keeps the same session id.
+                # compress_context() has just committed it, latching the guard —
+                # but the session is still live, so every later commit for it
+                # (the next compression, /new, normal session end, startup
+                # recovery) would be rejected and post-compression turns would
+                # never be extracted. Re-arm the guard now that compression has
+                # finished; turns arriving after this point are genuinely new.
+                #
+                # Rotation mode is untouched: there a fresh child id is minted
+                # and the old id stays latched, which is what dedupes its
+                # _finalize_session_async against this same commit.
+                self._clear_session_committed(old_session_id)
+
         if not rotate:
             # Same-session rewind (/undo) or no-op rotation: no commit and no
             # counter reset.

tests/plugins/memory/test_openviking_provider.py

@@ -993,3 +993,96 @@ def post(self, path, payload=None, **kwargs):
     assert "target_uri" not in payload
 
 
+
+
+def test_in_place_compression_rearms_commit_guard():
+    """Post-compression turns must still be committable (#74695).
+
+    ``compress_context()`` commits before rewriting the transcript, which
+    latches the per-sid guard. In-place mode (the default) keeps the SAME sid,
+    so the latch then rejected every later commit for a still-live session —
+    the next compression, /new, normal session end and startup recovery all
+    silently did nothing, and post-compression turns were never extracted.
+    """
+    provider = _make_provider_with_session("sid-123", turn_count=4)
+    provider._ensure_client = lambda: True
+
+    # Compression commits the live session, latching the guard.
+    provider._mark_session_committed("sid-123")
+    assert provider._session_needs_commit("sid-123", 4) is False
+
+    # In-place compression: same id in, no rotation.
+    provider.on_session_switch("sid-123", reason="compression")
+
+    # The session is still live, so new turns must be committable again.
+    assert provider._has_committed_session("sid-123") is False
+    assert provider._session_needs_commit("sid-123", 2) is True
+
+
+def test_rotating_compression_keeps_old_session_latched():
+    """Rotation mode must keep the guard, which dedupes the old id's finalize.
+
+    With ``compression.in_place: false`` a fresh child id is minted. The old id
+    stays committed so its ``_finalize_session_async`` does not double-commit
+    what compression already committed — the behavior the guard exists for.
+    """
+    provider = _make_provider_with_session("old-sid", turn_count=4)
+    provider._ensure_client = lambda: True
+    provider._finalize_session_async = MagicMock()
+
+    provider._mark_session_committed("old-sid")
+    provider.on_session_switch("new-sid", reason="compression")
+
+    assert provider._has_committed_session("old-sid") is True
+    assert provider._session_needs_commit("old-sid", 4) is False
+
+
+def test_undo_rewind_does_not_rearm_commit_guard():
+    """Only compression re-arms; a same-session /undo must not."""
+    provider = _make_provider_with_session("sid-123", turn_count=4)
+    provider._ensure_client = lambda: True
+
+    provider._mark_session_committed("sid-123")
+    provider.on_session_switch("sid-123", rewound=True)
+
+    assert provider._has_committed_session("sid-123") is True
+
+
+def test_in_place_compression_lifecycle_allows_a_later_commit(monkeypatch):
+    """End-to-end wiring, not a hand-set latch (#74695).
+
+    Drives the real sequence a session goes through: commit at the compression
+    boundary, same-id ``on_session_switch``, a post-compression turn via
+    ``sync_turn``, then a later commit. Before the fix the second commit never
+    reached the server, so every turn after the first compression was lost.
+    """
+    provider = _make_provider_with_session("sid-123", turn_count=3)
+    provider._ensure_client = lambda: True
+    provider._drain_writers = lambda sid, timeout=None: True
+    # Keep the async write worker out of it; the counter bump is what matters.
+    monkeypatch.setattr(provider, "_queue_memory_write", lambda *a, **k: None, raising=False)
+
+    def _commit_calls():
+        return [
+            c for c in provider._client.post.call_args_list
+            if c.args and str(c.args[0]).endswith("/commit")
+        ]
+
+    # 1. Compression commits the live session through the real path.
+    provider.on_session_end([{"role": "user", "content": "before"}])
+    assert len(_commit_calls()) == 1
+    assert provider._has_committed_session("sid-123") is True
+
+    # 2. In-place compression: same id back in, no rotation.
+    provider.on_session_switch("sid-123", reason="compression")
+
+    # 3. A genuinely new turn lands on the still-live session.
+    provider.sync_turn("after compression", "reply", session_id="sid-123")
+    assert provider._turn_count > 0
+
+    # 4. That turn must still be committable.
+    provider.on_session_end([{"role": "user", "content": "after"}])
+    assert len(_commit_calls()) == 2, (
+        "post-compression turns were never committed: "
+        f"{provider._client.post.call_args_list}"
+    )