feat(mem0-plugin): improve auto-triggering — pre-fetch, dedup, skill enforcement v0.2.3 (#5237)

mem0-plugin/tests/test_on_file_read.py

@@ -0,0 +1,105 @@
+"""Tests for on_file_read.sh hook — skip rules and dedup logic."""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+
+SCRIPT = os.path.join(os.path.dirname(__file__), "..", "scripts", "on_file_read.sh")
+
+
+def _run_hook(file_path: str, env_overrides: dict | None = None) -> subprocess.CompletedProcess:
+    env = {
+        **os.environ,
+        "MEM0_API_KEY": "",
+        "CLAUDE_PLUGIN_OPTION_API_KEY": "",
+        "CLAUDE_PLUGIN_OPTION_MEM0_API_KEY": "",
+        "USER": "testuser",
+    }
+    if env_overrides:
+        env.update(env_overrides)
+
+    payload = json.dumps({"tool_name": "Read", "tool_input": {"file_path": file_path}})
+    return subprocess.run(
+        ["bash", SCRIPT],
+        input=payload,
+        capture_output=True,
+        text=True,
+        env=env,
+        timeout=8,
+    )
+
+
+def test_skip_png():
+    result = _run_hook("/project/logo.png")
+    assert result.returncode == 0
+    assert result.stdout == ""
+
+
+def test_skip_lockfile():
+    result = _run_hook("/project/package-lock.json")
+    assert result.returncode == 0
+    assert result.stdout == ""
+
+
+def test_skip_min_js():
+    result = _run_hook("/project/dist/bundle.min.js")
+    assert result.returncode == 0
+    assert result.stdout == ""
+
+
+def test_skip_node_modules():
+    result = _run_hook("/project/node_modules/lodash/index.js")
+    assert result.returncode == 0
+    assert result.stdout == ""
+
+
+def test_skip_git_dir():
+    result = _run_hook("/project/.git/config")
+    assert result.returncode == 0
+    assert result.stdout == ""
+
+
+def test_skip_pycache():
+    result = _run_hook("/project/__pycache__/module.cpython-311.pyc")
+    assert result.returncode == 0
+    assert result.stdout == ""
+
+
+def test_skip_no_api_key(tmp_path):
+    """Code file but no API key — should exit cleanly with no output."""
+    result = _run_hook("/project/src/app.py", {"MEM0_API_KEY": "", "HOME": str(tmp_path)})
+    assert result.returncode == 0
+    assert result.stdout == ""
+
+
+def test_skip_empty_file_path():
+    """Empty file_path — should exit cleanly."""
+    result = _run_hook("")
+    assert result.returncode == 0
+    assert result.stdout == ""
+
+
+def test_dedup_repeated_reads(tmp_path):
+    """Same file read twice — second should be skipped (via dedup file)."""
+    dedup_file = str(tmp_path / "mem0_recent_reads_testuser")
+    env = {"MEM0_API_KEY": "fake-key", "USER": "testuser"}
+
+    # Write the file path into the dedup tracking file
+    with open(dedup_file, "w") as f:
+        f.write("/project/src/app.py\n")
+
+    # Patch RECENT_FILE by providing the USER env and ensuring /tmp has the file
+    # Since the script uses /tmp/mem0_recent_reads_${USER}, we create it there
+    tmp_dedup = "/tmp/mem0_recent_reads_testuser"
+    try:
+        with open(tmp_dedup, "w") as f:
+            f.write("/project/src/app.py\n")
+
+        result = _run_hook("/project/src/app.py", env)
+        assert result.returncode == 0
+        assert result.stdout == ""
+    finally:
+        if os.path.exists(tmp_dedup):
+            os.remove(tmp_dedup)

mem0-plugin/tests/test_rubric_dedup.py

@@ -0,0 +1,58 @@
+"""Tests for rubric deduplication in on_user_prompt.sh."""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+
+import pytest
+
+SCRIPTS_DIR = os.path.join(os.path.dirname(__file__), "..", "scripts")
+
+
+@pytest.fixture(autouse=True)
+def _clean_rubric_flag(tmp_path, monkeypatch):
+    """Use a temp dir for the rubric flag file."""
+    monkeypatch.setenv("MEM0_RUBRIC_DIR", str(tmp_path))
+    yield
+
+
+def _run_hook(prompt: str, env_overrides: dict | None = None, session_id: str = "test-sess-001") -> str:
+    """Run on_user_prompt.sh with a simulated prompt and return stdout."""
+    env = {
+        **os.environ,
+        "USER": "testuser",
+        "MEM0_API_KEY": "test-key-123",
+        "MEM0_RESOLVED_USER_ID": "testuser",
+        "MEM0_PROJECT_ID": "test-project",
+        "MEM0_BRANCH": "main",
+    }
+    if env_overrides:
+        env.update(env_overrides)
+
+    input_json = json.dumps({"prompt": prompt, "session_id": session_id})
+    result = subprocess.run(
+        ["bash", os.path.join(SCRIPTS_DIR, "on_user_prompt.sh")],
+        input=input_json,
+        capture_output=True,
+        text=True,
+        env=env,
+        timeout=10,
+    )
+    return result.stdout
+
+
+def test_first_prompt_gets_full_rubric():
+    """First substantial prompt of session gets full memory check rubric."""
+    output = _run_hook("How should we refactor the auth module?")
+    assert "Search mem0" in output
+    assert "Search tips" in output
+    assert "metadata.type" in output
+
+
+def test_second_prompt_gets_no_rubric():
+    """Second prompt of session emits nothing — rubric and tips only on first prompt."""
+    _run_hook("How should we refactor the auth module?")
+    output = _run_hook("What about the database layer?")
+    assert output.strip() == ""

mem0-plugin/tests/test_search.py

@@ -0,0 +1,116 @@
+"""Tests for _search.py — shared mem0 search API helper."""
+
+from __future__ import annotations
+
+import json
+from unittest.mock import MagicMock, patch
+
+
+def test_search_memories_returns_results():
+    from _search import search_memories
+
+    fake_results = [
+        {"id": "abc123", "memory": "Use Postgres for auth", "metadata": {"type": "decision"}},
+        {"id": "def456", "memory": "Never use floats for money", "metadata": {"type": "anti_pattern"}},
+    ]
+
+    def mock_urlopen(req, timeout=None):
+        resp = MagicMock()
+        resp.read.return_value = json.dumps({"results": fake_results}).encode()
+        resp.__enter__ = lambda s: s
+        resp.__exit__ = MagicMock(return_value=False)
+        return resp
+
+    with patch("urllib.request.urlopen", side_effect=mock_urlopen):
+        results = search_memories("test-key", "user1", "proj1", "auth decisions")
+
+    assert len(results) == 2
+    assert results[0]["id"] == "abc123"
+
+
+def test_search_memories_with_metadata_type():
+    from _search import search_memories
+
+    captured_body = {}
+
+    def mock_urlopen(req, timeout=None):
+        captured_body.update(json.loads(req.data.decode()))
+        resp = MagicMock()
+        resp.read.return_value = json.dumps({"results": []}).encode()
+        resp.__enter__ = lambda s: s
+        resp.__exit__ = MagicMock(return_value=False)
+        return resp
+
+    with patch("urllib.request.urlopen", side_effect=mock_urlopen):
+        search_memories("key", "user", "proj", "query", metadata_type="decision")
+
+    filters = captured_body["filters"]
+    assert {"metadata": {"type": "decision"}} in filters["AND"]
+
+
+def test_search_memories_handles_api_error():
+    from _search import search_memories
+
+    with patch("urllib.request.urlopen", side_effect=Exception("timeout")):
+        results = search_memories("key", "user", "proj", "query")
+
+    assert results == []
+
+
+def test_search_memories_handles_list_response():
+    from _search import search_memories
+
+    fake_results = [{"id": "abc", "memory": "test"}]
+
+    def mock_urlopen(req, timeout=None):
+        resp = MagicMock()
+        resp.read.return_value = json.dumps(fake_results).encode()
+        resp.__enter__ = lambda s: s
+        resp.__exit__ = MagicMock(return_value=False)
+        return resp
+
+    with patch("urllib.request.urlopen", side_effect=mock_urlopen):
+        results = search_memories("key", "user", "proj", "query")
+
+    assert len(results) == 1
+
+
+def test_search_memories_respects_top_k():
+    from _search import search_memories
+
+    captured_body = {}
+
+    def mock_urlopen(req, timeout=None):
+        captured_body.update(json.loads(req.data.decode()))
+        resp = MagicMock()
+        resp.read.return_value = json.dumps({"results": []}).encode()
+        resp.__enter__ = lambda s: s
+        resp.__exit__ = MagicMock(return_value=False)
+        return resp
+
+    with patch("urllib.request.urlopen", side_effect=mock_urlopen):
+        search_memories("key", "user", "proj", "query", top_k=5)
+
+    assert captured_body["top_k"] == 5
+
+
+def test_search_memories_no_api_key_returns_empty():
+    from _search import search_memories
+
+    results = search_memories("", "user", "proj", "query")
+    assert results == []
+
+
+def test_format_results_for_context():
+    from _search import format_results_for_context
+
+    memories = [
+        {"id": "abc12345-long-id", "memory": "Use Postgres for auth", "metadata": {"type": "decision"}},
+        {"id": "def67890-long-id", "memory": "JWT tokens expire in 1h", "metadata": {"type": "convention"}},
+    ]
+
+    output = format_results_for_context(memories, heading="Relevant memories")
+    assert "Relevant memories" in output
+    assert "[decision]" in output
+    assert "Use Postgres for auth" in output
+    assert "abc12345" in output

mem0-plugin/tests/test_write_path.py

@@ -190,5 +190,6 @@ def test_resolve_api_key_returns_empty_when_neither_set(monkeypatch):
     from _identity import resolve_api_key
 
     monkeypatch.delenv("MEM0_API_KEY", raising=False)
+    monkeypatch.delenv("CLAUDE_PLUGIN_OPTION_API_KEY", raising=False)
     monkeypatch.delenv("CLAUDE_PLUGIN_OPTION_MEM0_API_KEY", raising=False)
     assert resolve_api_key() == ""