tests/plugins/memory/test_mem0_v3.py

"""Tests for Mem0 v3 API — new tool names, paginated responses, update/delete tools."""

import json
import threading
import time
import pytest

from agent import secret_scope
import plugins.memory.mem0 as mem0_plugin
from plugins.memory.mem0 import Mem0MemoryProvider


class FakeBackend:
    """Fake Mem0Backend for provider-level tests."""

    def __init__(self, search_results=None, all_results=None):
        self._search_results = search_results or []
        self._all_results = all_results or {"results": [], "count": 0}
        self.captured = []

    def search(self, query, *, filters, top_k=10, rerank=True):
        self.captured.append(("search", query, {"filters": filters, "top_k": top_k, "rerank": rerank}))
        return self._search_results

    def get_all(self, *, filters, page=1, page_size=100):
        self.captured.append(("get_all", {"filters": filters, "page": page, "page_size": page_size}))
        return self._all_results

    def add(self, messages, *, user_id, agent_id, infer=False, metadata=None):
        self.captured.append((
            "add",
            messages,
            {"user_id": user_id, "agent_id": agent_id, "infer": infer, "metadata": metadata},
        ))
        return {"status": "PENDING", "event_id": "evt-test-123"}

    def update(self, memory_id, text):
        self.captured.append(("update", memory_id, text))
        return {"result": "Memory updated.", "memory_id": memory_id}

    def delete(self, memory_id):
        self.captured.append(("delete", memory_id))
        return {"result": "Memory deleted.", "memory_id": memory_id}


class TestMem0V3Tools:
    """Test v3 tool names and response handling."""

    def _make_provider(self, monkeypatch, backend):
        provider = Mem0MemoryProvider()
        provider.initialize("test-session")
        provider._user_id = "u123"
        provider._agent_id = "hermes"
        provider._backend = backend
        return provider

    def test_search_returns_ids(self, monkeypatch):
        backend = FakeBackend(search_results=[{"id": "mem-1", "memory": "foo", "score": 0.9}])
        provider = self._make_provider(monkeypatch, backend)
        result = json.loads(provider.handle_tool_call("mem0_search", {"query": "test"}))
        assert result["results"][0]["id"] == "mem-1"


    def test_add_uses_content_param(self, monkeypatch):
        backend = FakeBackend()
        provider = self._make_provider(monkeypatch, backend)
        result = json.loads(provider.handle_tool_call("mem0_add", {"content": "user likes dark mode"}))
        assert len(backend.captured) == 1
        call = backend.captured[0]
        assert call[2]["infer"] is False
        assert call[2]["user_id"] == "u123"
        assert call[2]["agent_id"] == "hermes"
        assert "event_id" in result


    def test_old_tool_names_return_unknown(self, monkeypatch):
        backend = FakeBackend()
        provider = self._make_provider(monkeypatch, backend)
        result = json.loads(provider.handle_tool_call("mem0_profile", {}))
        assert "error" in result
        result = json.loads(provider.handle_tool_call("mem0_conclude", {}))
        assert "error" in result


class TestMem0UpdateDelete:

    def _make_provider(self, monkeypatch, backend):
        provider = Mem0MemoryProvider()
        provider.initialize("test-session")
        provider._user_id = "u123"
        provider._agent_id = "hermes"
        provider._backend = backend
        return provider

    def test_update_calls_sdk(self, monkeypatch):
        backend = FakeBackend()
        provider = self._make_provider(monkeypatch, backend)
        result = json.loads(provider.handle_tool_call(
            "mem0_update", {"memory_id": "mem-1", "text": "updated fact"}
        ))
        assert backend.captured[0][1] == "mem-1"
        assert backend.captured[0][2] == "updated fact"
        assert result["result"] == "Memory updated."
        assert result["memory_id"] == "mem-1"


    def test_delete_calls_sdk(self, monkeypatch):
        backend = FakeBackend()
        provider = self._make_provider(monkeypatch, backend)
        result = json.loads(provider.handle_tool_call(
            "mem0_delete", {"memory_id": "mem-1"}
        ))
        assert backend.captured[0][1] == "mem-1"
        assert result["result"] == "Memory deleted."


class TestMem0ErrorHandling:

    def _make_provider(self, monkeypatch, backend):
        provider = Mem0MemoryProvider()
        provider.initialize("test-session")
        provider._user_id = "u123"
        provider._agent_id = "hermes"
        provider._backend = backend
        return provider


class TestMem0V3Internal:

    def _make_provider(self, monkeypatch, backend):
        provider = Mem0MemoryProvider()
        provider.initialize("test-session")
        provider._user_id = "u123"
        provider._agent_id = "hermes"
        provider._backend = backend
        return provider

    def test_sync_turn_explicit_kwargs(self, monkeypatch):
        backend = FakeBackend()
        provider = self._make_provider(monkeypatch, backend)
        provider.sync_turn("user said", "assistant replied", session_id="s1")
        provider._sync_thread.join(timeout=2)
        assert len(backend.captured) == 1
        call = backend.captured[0]
        assert call[2]["user_id"] == "u123"
        assert call[2]["agent_id"] == "hermes"
        assert call[2]["infer"] is True


class TestSyncTurnTruncation:
    """sync_turn must cap messages before ingestion so small-context embedding
    backends (OSS Ollama bge-small-zh-v1.5: 512 tokens; jina-embeddings-v3 token
    limits) don't fail the whole extraction — a failure _try only logs."""

    def _make_provider(self, monkeypatch, backend):
        provider = Mem0MemoryProvider()
        provider.initialize("test-session")
        provider._user_id = "u123"
        provider._agent_id = "hermes"
        provider._backend = backend
        return provider

    def test_small_context_backend_never_sees_oversized_input(self, monkeypatch):
        """Regression for #106235/#37421: an OSS embedding backend with a small
        context window raises on oversized input; truncation up front keeps the
        extraction from being silently dropped (no breaker failures)."""

        class SmallContextBackend(FakeBackend):
            def add(self, messages, **kwargs):
                if any(len(m["content"]) > mem0_plugin._SYNC_MSG_MAX_CHARS for m in messages):
                    raise RuntimeError("HTTP 500: embedding input exceeds model context")
                return super().add(messages, **kwargs)

        backend = SmallContextBackend()
        provider = self._make_provider(monkeypatch, backend)
        provider.sync_turn("Short question?", "".join(f"Fact {i}. " for i in range(200)), session_id="s1")
        provider._sync_thread.join(timeout=2)
        assert len(backend.captured) == 1
        sent = backend.captured[0][1]
        assert sent[0]["content"] == "Short question?"  # under the cap: untouched
        assert len(sent[1]["content"]) <= mem0_plugin._SYNC_MSG_MAX_CHARS and sent[1]["content"].endswith(".")
        assert provider._consecutive_failures == 0

    def test_the_boundary_kept_is_the_last_one_in_the_window_whatever_its_script(self):
        """A mixed-script turn must not be cut back to an early CJK stop.

        The trim exists to keep as much of the turn as the embedder can take; picking the
        first separator KIND that qualifies instead of the last boundary threw away most of
        the allowed window whenever two kinds appeared — an early ``。`` (or ``.``, which
        outranks ``!``/``?``) beat a boundary 240 characters later, so the facts stated in
        the rest of the message never reached extraction.
        """
        cap = mem0_plugin._SYNC_MSG_MAX_CHARS
        early, late = cap // 2, cap - 9

        for early_sep, late_sep in (("。", "."), (".", "!"), ("？", "?"), ("！", ".")):
            text = "a" * early + early_sep + "b" * (late - early - 1) + late_sep + "c" * cap
            assert text[late] == late_sep and len(text) > cap  # both boundaries inside the window
            kept = mem0_plugin._truncate_for_sync(text)
            assert kept == text[:late + 1], f"{early_sep!r} before {late_sep!r} cut back to {len(kept)} chars"
            assert kept.endswith(late_sep)

    def test_a_boundary_only_in_the_first_third_still_falls_back_to_a_hard_cut(self):
        """Unsegmented input keeps the whole window rather than a sliver of a sentence."""
        cap = mem0_plugin._SYNC_MSG_MAX_CHARS
        text = "a" * 10 + "." + "b" * (cap * 2)
        assert mem0_plugin._truncate_for_sync(text) == text[:cap]

    def test_sync_max_chars_config_raises_cap(self, monkeypatch, tmp_path):
        """8k-token embedders should not be stuck at the 512-token default (#106235)."""
        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
        monkeypatch.setenv("MEM0_API_KEY", "test-key")
        (tmp_path / "mem0.json").write_text('{"sync_max_chars": 3000}')
        backend = FakeBackend()
        provider = self._make_provider(monkeypatch, backend)
        provider.sync_turn("hi", "Long answer. " * 200, session_id="s1")  # 2600 chars
        provider._sync_thread.join(timeout=2)
        assert backend.captured[0][1][1]["content"] == "Long answer. " * 200


class TestMem0Prefetch:
    """prefetch() must recall on the CURRENT question, synchronously.

    The old implementation ignored its ``query`` and returned whatever a
    background ``queue_prefetch`` had warmed from the PREVIOUS turn — so the
    first turn injected nothing and later turns injected stale, off-topic
    memories. These lock the corrected behaviour.
    """

    def _make_provider(self, backend):
        provider = Mem0MemoryProvider()
        provider.initialize("test-session")
        provider._user_id = "u123"
        provider._agent_id = "hermes"
        provider._backend = backend
        return provider

    def test_prefetch_searches_current_query(self):
        backend = FakeBackend(search_results=[{"id": "m1", "memory": "user prefers dark mode"}])
        provider = self._make_provider(backend)
        result = provider.prefetch("what theme do I like?")
        kind, query, opts = backend.captured[0]
        assert kind == "search"
        assert query == "what theme do I like?"
        assert opts["filters"] == {"user_id": "u123"}
        assert opts["top_k"] == 10
        assert opts["rerank"] is False
        assert "## Mem0 Memory" in result
        assert "user prefers dark mode" in result


    def test_on_turn_start_queues_current_query(self):
        backend = FakeBackend(search_results=[{"id": "m1", "memory": "lives in Berlin"}])
        provider = self._make_provider(backend)
        provider.on_turn_start(1, "where do I live?")
        provider._prefetch_thread.join(timeout=1)
        result = provider.prefetch("where do I live?")
        assert "lives in Berlin" in result
        assert len([c for c in backend.captured if c[0] == "search"]) == 1

    def test_slow_prefetch_returns_quickly(self, monkeypatch):
        entered = threading.Event()
        release = threading.Event()
        search_returned = threading.Event()

        class SlowBackend(FakeBackend):
            def search(self, query, *, filters, top_k=10, rerank=True):
                entered.set()
                try:
                    release.wait(30)
                    return super().search(
                        query, filters=filters, top_k=top_k, rerank=rerank
                    )
                finally:
                    search_returned.set()

        monkeypatch.setattr(mem0_plugin, "_PREFETCH_WAIT_SECS", 0.01)
        provider = self._make_provider(
            SlowBackend(search_results=[{"id": "m1", "memory": "lives in Berlin"}])
        )
        # DETERMINISTIC non-blocking witness — replaces `assert elapsed < 0.1`.
        #
        # The old form slept 0.2s in the backend and asserted prefetch returned
        # in under 0.1s. That makes the OS scheduler part of the assertion: on
        # a loaded box thread startup alone can eat the 100ms budget, so the
        # inequality flips with nothing wrong in the code under test. Observed
        # failing in a full-directory run of tests/plugins/memory.
        #
        # The real contract is that prefetch gives up on the slow backend
        # instead of waiting for it. Assert it directly: the backend search is
        # STILL PARKED (release unset, so `search_returned` cannot be set). If
        # prefetch ever waited for the backend, the search would have returned
        # first and this fails. No wall-clock constant.
        assert provider.prefetch("where do I live?") == ""
        assert entered.wait(30), "prefetch never reached the backend"
        assert not search_returned.is_set(), (
            "prefetch blocked on the slow backend: the backend search had "
            "already returned by the time prefetch did"
        )

        release.set()
        provider._prefetch_thread.join(timeout=30)
        assert "lives in Berlin" in provider.prefetch("where do I live?")


    def test_queue_prefetch_fires_no_search(self):
        # prefetch is synchronous now, so the post-turn warm is redundant and
        # must not fire a wasted backend search.
        backend = FakeBackend(search_results=[{"id": "m1", "memory": "x"}])
        provider = self._make_provider(backend)
        provider.queue_prefetch("previous turn text")
        assert backend.captured == []


class TestMem0V3Config:

    def test_tool_schemas_four_tools(self):
        provider = Mem0MemoryProvider()
        schemas = provider.get_tool_schemas()
        names = [s["name"] for s in schemas]
        assert names == ["mem0_search", "mem0_add", "mem0_update", "mem0_delete"]

    def test_system_prompt_new_tool_names(self):
        provider = Mem0MemoryProvider()
        provider._user_id = "test"
        block = provider.system_prompt_block()
        assert "mem0_search" in block
        assert "mem0_add" in block
        assert "mem0_update" in block
        assert "mem0_delete" in block
        assert "mem0_list" not in block
        assert "mem0_profile" not in block
        assert "mem0_conclude" not in block


class TestMem0ModeSwitch:

    def test_oss_mode_initializes_without_platform_key_in_scope(
        self, monkeypatch, tmp_path
    ):
        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
        monkeypatch.delenv("MEM0_API_KEY", raising=False)
        (tmp_path / "mem0.json").write_text(
            json.dumps(
                {
                    "mode": "oss",
                    "oss": {"vector_store": {"provider": "qdrant"}},
                }
            )
        )

        # Contract (#99121, restated for fail-loud reads): every production caller is scoped
        # (turn/cron/kanban scope installers); an OSS profile whose scope simply lacks MEM0_API_KEY
        # must initialize. A scope-LESS multiplex caller is a spawn-site bug and raises instead —
        # see test_load_config_fails_closed_without_scope_even_for_identity_settings.
        token = secret_scope.set_secret_scope({})
        secret_scope.set_multiplex_active(True)
        try:
            provider = Mem0MemoryProvider()
            provider._create_backend = lambda: None  # type: ignore[method-assign]
            provider.initialize("test")
            available = provider.is_available()
        finally:
            secret_scope.set_multiplex_active(False)
            secret_scope.reset_secret_scope(token)

        assert provider._mode == "oss"
        assert provider._api_key == ""
        assert available is True

    def test_platform_config_still_fails_closed_without_profile_scope(
        self, monkeypatch, tmp_path
    ):
        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
        monkeypatch.delenv("MEM0_API_KEY", raising=False)

        token = secret_scope.set_secret_scope(None)
        secret_scope.set_multiplex_active(True)
        try:
            with pytest.raises(secret_scope.UnscopedSecretError):
                Mem0MemoryProvider().is_available()
        finally:
            secret_scope.set_multiplex_active(False)
            secret_scope.reset_secret_scope(token)

    def test_load_config_fails_closed_without_scope_even_for_identity_settings(
        self, monkeypatch, tmp_path
    ):
        """A scope-less multiplex caller is a spawn-site bug: identity/mode reads must surface it,
        not degrade to '' and route the turn's memories into the default profile's account."""
        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
        (tmp_path / "mem0.json").write_text(json.dumps({"mode": "oss", "oss": {"vector_store": {"provider": "qdrant"}}}))

        token = secret_scope.set_secret_scope(None)
        secret_scope.set_multiplex_active(True)
        try:
            with pytest.raises(secret_scope.UnscopedSecretError):
                mem0_plugin._load_config()
        finally:
            secret_scope.set_multiplex_active(False)
            secret_scope.reset_secret_scope(token)

    def test_file_api_key_still_overrides_environment(self, monkeypatch, tmp_path):
        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
        monkeypatch.setenv("MEM0_API_KEY", "env-key")
        (tmp_path / "mem0.json").write_text(
            json.dumps({"api_key": "file-key"})
        )

        assert mem0_plugin._load_config()["api_key"] == "file-key"

    def test_default_mode_is_platform(self, monkeypatch, tmp_path):
        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
        monkeypatch.setenv("MEM0_API_KEY", "test-key")
        provider = Mem0MemoryProvider()
        provider.initialize("test")
        assert provider._mode == "platform"

    def test_missing_mode_key_defaults_platform(self, monkeypatch, tmp_path):
        """Backward compat: old mem0.json without mode key works."""
        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
        config_path = tmp_path / "mem0.json"
        config_path.write_text('{"user_id": "old-user"}')
        monkeypatch.setenv("MEM0_API_KEY", "test-key")
        provider = Mem0MemoryProvider()
        provider.initialize("test")
        assert provider._mode == "platform"
        assert provider._user_id == "old-user"

    def test_is_available_platform_needs_key(self, monkeypatch, tmp_path):
        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
        monkeypatch.delenv("MEM0_API_KEY", raising=False)
        provider = Mem0MemoryProvider()
        assert provider.is_available() is False


class TestMem0UserIdResolution:
    """user_id resolution: configured override > gateway-native id > placeholder.

    Same human across CLI / Telegram / Discord / Slack / etc. should map to
    the same memory store when MEM0_USER_ID is set, and only fall back to the
    gateway-native id when it isn't.
    """

    def _provider(self, monkeypatch, tmp_path):
        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
        monkeypatch.setenv("MEM0_API_KEY", "test-key")
        provider = Mem0MemoryProvider()
        # Skip backend instantiation — we only care about identity resolution.
        provider._create_backend = lambda: None  # type: ignore[method-assign]
        return provider

    def test_env_override_beats_gateway_native_id(self, monkeypatch, tmp_path):
        monkeypatch.setenv("MEM0_USER_ID", "ryan@example.com")
        provider = self._provider(monkeypatch, tmp_path)
        provider.initialize("test", user_id="123456789", platform="telegram")
        assert provider._user_id == "ryan@example.com"

    def test_file_override_beats_gateway_native_id(self, monkeypatch, tmp_path):
        monkeypatch.delenv("MEM0_USER_ID", raising=False)
        (tmp_path / "mem0.json").write_text('{"user_id": "ryan@example.com"}')
        provider = self._provider(monkeypatch, tmp_path)
        provider.initialize("test", user_id="123456789", platform="telegram")
        assert provider._user_id == "ryan@example.com"

    def test_unset_falls_back_to_gateway_native_id(self, monkeypatch, tmp_path):
        monkeypatch.delenv("MEM0_USER_ID", raising=False)
        provider = self._provider(monkeypatch, tmp_path)
        provider.initialize("test", user_id="123456789", platform="telegram")
        assert provider._user_id == "123456789"


    def test_legacy_placeholder_in_config_does_not_override_kwargs(self, monkeypatch, tmp_path):
        # Setup wizard historically wrote {"user_id": "hermes-user"} as the
        # suggested default. Treat that placeholder as unset so users on
        # gateways still get gateway-native ids — not silent collisions.
        monkeypatch.delenv("MEM0_USER_ID", raising=False)
        (tmp_path / "mem0.json").write_text('{"user_id": "hermes-user"}')
        provider = self._provider(monkeypatch, tmp_path)
        provider.initialize("test", user_id="123456789", platform="telegram")
        assert provider._user_id == "123456789"


class TestMem0WriteMetadata:
    """Writes carry metadata.channel so per-channel filtered views are possible
    without coupling identity to the channel.
    """

    def _make_provider(self, channel: str = "cli"):
        provider = Mem0MemoryProvider()
        provider._user_id = "u123"
        provider._agent_id = "hermes"
        provider._channel = channel
        provider._backend = FakeBackend()
        return provider


class _SentinelBackend:
    def __init__(self, *args):
        self.args = args


class TestCreateBackendRouting:
    """_create_backend() must pick the backend matching the configured mode/host."""

    def _provider(self, monkeypatch, *, mode="platform", api_key="k", host=""):
        # Neutralize lazy-install so the routing decision is all we exercise.
        monkeypatch.setattr("tools.lazy_deps.ensure", lambda *a, **k: None, raising=False)
        provider = Mem0MemoryProvider()
        provider._mode = mode
        provider._api_key = api_key
        provider._host = host
        provider._config = {"oss": {"vector_store": {"provider": "qdrant"}}}
        return provider

    def test_routes_to_selfhosted_when_host_set(self, monkeypatch):
        captured = {}

        class SH(_SentinelBackend):
            def __init__(self, api_key, host):
                captured["args"] = (api_key, host)

        monkeypatch.setattr("plugins.memory.mem0._backend.SelfHostedBackend", SH)
        provider = self._provider(monkeypatch, host="http://sh:8888", api_key="adminkey")
        backend = provider._create_backend()
        assert isinstance(backend, SH)
        assert captured["args"] == ("adminkey", "http://sh:8888")


    def test_oss_mode_takes_precedence_over_host(self, monkeypatch):
        class OB(_SentinelBackend):
            def __init__(self, cfg):
                pass

        monkeypatch.setattr("plugins.memory.mem0._backend.OSSBackend", OB)
        provider = self._provider(monkeypatch, mode="oss", host="http://sh:8888")
        assert isinstance(provider._create_backend(), OB)

    def test_prompt_label_matches_routing_when_oss_and_host_both_set(self, monkeypatch):
        # system_prompt_block must mirror _create_backend precedence: with both
        # mode=oss and host set, OSS wins the routing, so the prompt must label
        # OSS — not "self-hosted (HTTP API)". Guards the prompt-vs-routing lie.
        provider = self._provider(monkeypatch, mode="oss", host="http://sh:8888")
        provider._user_id = "test"
        block = provider.system_prompt_block()
        assert "OSS" in block
        assert "HTTP API" not in block


class TestSelfHostedConfig:
    """Config plumbing for self-hosted (MEM0_HOST env + is_available)."""

    def test_load_config_reads_mem0_host_env(self, monkeypatch):
        monkeypatch.setenv("MEM0_HOST", "http://localhost:8888")
        assert mem0_plugin._load_config()["host"] == "http://localhost:8888"