hermes-agent / tests /plugins /memory /test_mem0_v3.py

Add files using upload-large-folder tool

108e227 verified about 2 months ago

24 kB

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

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