diff --git a/tests/memory/long_term_memory_backends/test_opensearch_ltm_backend.py b/tests/memory/long_term_memory_backends/test_opensearch_ltm_backend.py new file mode 100644 index 000000000..947190c41 --- /dev/null +++ b/tests/memory/long_term_memory_backends/test_opensearch_ltm_backend.py @@ -0,0 +1,142 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""OpensearchLTMBackend must not map distinct (index, user_id) pairs to one index. + +``f"{self.index}_{user_id}"`` is ambiguous: backends with +``index="app"`` / ``user_id="u1_x"`` and ``index="app_u1"`` / ``user_id="x"`` +both resolve to ``app_u1_x`` and therefore share memories. +""" + +from __future__ import annotations + +import pytest +from llama_index.core.embeddings import MockEmbedding + +from veadk.configs.database_configs import OpensearchConfig +from veadk.memory.long_term_memory_backends import opensearch_backend +from veadk.memory.long_term_memory_backends.opensearch_backend import ( + OpensearchLTMBackend, +) + + +@pytest.fixture +def patch_embed(monkeypatch): + monkeypatch.setenv("MODEL_EMBEDDING_API_KEY", "test-key") + monkeypatch.setattr( + opensearch_backend, + "create_embedding_model", + lambda **_: MockEmbedding(embed_dim=8), + ) + + +def _backend(index: str) -> OpensearchLTMBackend: + return OpensearchLTMBackend( + index=index, + opensearch_config=OpensearchConfig( + host="localhost", + port=9200, + username="admin", + password="admin", + use_ssl=False, + ), + ) + + +def test_underscore_join_collides_for_nested_index_and_user_id(): + """Document the bug shape before the scoped-index helper is applied.""" + assert f"{'app'}_{'u1_x'}" == f"{'app_u1'}_{'x'}" == "app_u1_x" + + +def test_user_scoped_index_disambiguates_nested_underscore_pairs(patch_embed): + app = _backend("app") + app_u1 = _backend("app_u1") + + left = app._user_scoped_index("u1_x") + right = app_u1._user_scoped_index("x") + + assert left != right + assert left.islower() + assert right.islower() + assert not left.startswith(("_", "-")) + assert not right.startswith(("_", "-")) + + +def test_user_scoped_index_is_stable_for_same_pair(patch_embed): + backend = _backend("my-app") + + assert backend._user_scoped_index("alice") == backend._user_scoped_index("alice") + + +def test_user_scoped_index_differs_for_raw_values_that_sanitize_alike(patch_embed): + """Sanitizing alone would collapse ``u1_x`` and ``u1-x``; digests must not.""" + backend = _backend("app") + + assert backend._user_scoped_index("u1_x") != backend._user_scoped_index("u1-x") + + +def test_save_and_search_use_the_same_scoped_index(patch_embed, monkeypatch): + backend = _backend("app") + seen: list[str] = [] + + def fake_create(index: str): + seen.append(index) + + class _Index: + def insert_nodes(self, nodes): + return None + + def as_retriever(self, similarity_top_k: int): + class _Retriever: + def retrieve(self, query): + return [] + + return _Retriever() + + return _Index() + + monkeypatch.setattr(backend, "_create_vector_index", fake_create) + + backend.save_memory(user_id="u1_x", event_strings=["hello"]) + backend.search_memory(user_id="u1_x", query="hello", top_k=3) + + assert seen == [ + backend._user_scoped_index("u1_x"), + backend._user_scoped_index("u1_x"), + ] + + +def test_distinct_backends_do_not_share_scoped_index_on_collision_pair( + patch_embed, monkeypatch +): + app = _backend("app") + app_u1 = _backend("app_u1") + seen: list[str] = [] + + def fake_create(index: str): + seen.append(index) + + class _Index: + def insert_nodes(self, nodes): + return None + + return _Index() + + monkeypatch.setattr(app, "_create_vector_index", fake_create) + monkeypatch.setattr(app_u1, "_create_vector_index", fake_create) + + app.save_memory(user_id="u1_x", event_strings=["from-app"]) + app_u1.save_memory(user_id="x", event_strings=["from-app-u1"]) + + assert seen[0] != seen[1] diff --git a/veadk/memory/long_term_memory_backends/opensearch_backend.py b/veadk/memory/long_term_memory_backends/opensearch_backend.py index 373d1dcb8..4540aceb0 100644 --- a/veadk/memory/long_term_memory_backends/opensearch_backend.py +++ b/veadk/memory/long_term_memory_backends/opensearch_backend.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import hashlib import re from llama_index.core import Document, VectorStoreIndex @@ -72,6 +73,32 @@ def precheck_index_naming(self, index: str): f"The index name {index} does not conform to the naming rules of OpenSearch" ) + def _user_scoped_index(self, user_id: str) -> str: + """Build an OpenSearch index name unique to ``(self.index, user_id)``. + + ``f"{self.index}_{user_id}"`` is ambiguous: ``index="app"`` with + ``user_id="u1_x"`` and ``index="app_u1"`` with ``user_id="x"`` both + become ``app_u1_x`` and therefore share memories. Each part is + sanitized for OpenSearch naming and tagged with a short digest of the + raw value so pairs that sanitize alike still stay distinct. + """ + + def part(value: str, *, fallback: str) -> str: + raw = str(value or "").strip() + normalized = re.sub(r"[^a-z0-9]+", "-", raw.lower()).strip("-") + if not normalized: + normalized = fallback + digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()[:12] + return f"{normalized[:48]}_{digest}" + + name = ( + "veadk-ltm-" + f"{part(self.index, fallback='app')}-" + f"{part(user_id, fallback='user')}" + ) + # OpenSearch index names are capped at 255 bytes. + return name[:255] + def _create_vector_index(self, index: str) -> VectorStoreIndex: logger.info(f"Create OpenSearch vector index with index={index}") @@ -102,7 +129,7 @@ def _create_vector_index(self, index: str) -> VectorStoreIndex: @override def save_memory(self, user_id: str, event_strings: list[str], **kwargs) -> bool: - index = f"{self.index}_{user_id}" + index = self._user_scoped_index(user_id) vector_index = self._create_vector_index(index) for event_string in event_strings: @@ -115,7 +142,7 @@ def save_memory(self, user_id: str, event_strings: list[str], **kwargs) -> bool: def search_memory( self, user_id: str, query: str, top_k: int, **kwargs ) -> list[str]: - index = f"{self.index}_{user_id}" + index = self._user_scoped_index(user_id) vector_index = self._create_vector_index(index)