feat(mem0-plugin): v0.2.6 — fix memory visibility, remove redundant hooks, reduce latency (#5257)

mem0-plugin/tests/conftest.py

@@ -23,8 +23,9 @@ def _scripts_on_path():
 
 
 @pytest.fixture(autouse=True)
-def _clean_project_map():
-    """Remove project_map.json before each test to prevent cross-test pollution."""
+def _clean_project_map(monkeypatch):
+    """Remove project_map.json and clear MEM0_PROJECT_ID before each test."""
+    monkeypatch.delenv("MEM0_PROJECT_ID", raising=False)
     map_path = os.path.expanduser("~/.mem0/project_map.json")
     if os.path.isfile(map_path):
         os.remove(map_path)

mem0-plugin/tests/test_auto_capture.py

@@ -0,0 +1,142 @@
+"""Tests for auto_capture.py transcript parsing and exchange extraction."""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+
+import pytest
+
+SCRIPTS_DIR = os.path.join(os.path.dirname(__file__), "..", "scripts")
+
+
+@pytest.fixture(autouse=True)
+def _scripts_path():
+    abs_scripts = os.path.abspath(SCRIPTS_DIR)
+    if abs_scripts not in sys.path:
+        sys.path.insert(0, abs_scripts)
+    yield
+    if abs_scripts in sys.path:
+        sys.path.remove(abs_scripts)
+
+
+def _make_transcript(tmp_path, entries):
+    path = tmp_path / "transcript.jsonl"
+    lines = []
+    for entry in entries:
+        lines.append(json.dumps(entry))
+    path.write_text("\n".join(lines) + "\n")
+    return str(path)
+
+
+def _msg(role, content):
+    return {"message": {"role": role, "content": content}}
+
+
+class TestTailLines:
+    def test_reads_last_n_lines(self, tmp_path):
+        from auto_capture import tail_lines
+
+        p = tmp_path / "test.txt"
+        p.write_text("\n".join(f"line{i}" for i in range(100)) + "\n")
+        result = tail_lines(str(p), 5)
+        assert len(result) >= 5
+        assert result[-1] == "line99"
+
+    def test_empty_file(self, tmp_path):
+        from auto_capture import tail_lines
+
+        p = tmp_path / "empty.txt"
+        p.write_text("")
+        assert tail_lines(str(p), 10) == []
+
+    def test_nonexistent_file(self):
+        from auto_capture import tail_lines
+
+        assert tail_lines("/nonexistent/path", 10) == []
+
+
+class TestExtractRecentExchanges:
+    def test_extracts_user_assistant_pairs(self):
+        from auto_capture import extract_recent_exchanges
+
+        lines = [
+            json.dumps(_msg("user", "What is Python used for?" * 3)),
+            json.dumps(_msg("assistant", "Python is used for many things." * 3)),
+            json.dumps(_msg("user", "Tell me about web frameworks." * 3)),
+            json.dumps(_msg("assistant", "Django and Flask are popular." * 3)),
+        ]
+        result = extract_recent_exchanges(lines, max_exchanges=2)
+        assert len(result) == 4
+        assert result[0]["role"] == "user"
+        assert result[1]["role"] == "assistant"
+
+    def test_skips_short_messages(self):
+        from auto_capture import extract_recent_exchanges
+
+        lines = [
+            json.dumps(_msg("user", "ok")),
+            json.dumps(_msg("assistant", "Sure, here is a detailed explanation." * 3)),
+        ]
+        result = extract_recent_exchanges(lines, max_exchanges=2)
+        assert len(result) == 1
+        assert result[0]["role"] == "assistant"
+
+    def test_skips_compact_summaries(self):
+        from auto_capture import extract_recent_exchanges
+
+        lines = [
+            json.dumps({"isCompactSummary": True, "message": {"role": "assistant", "content": "summary " * 20}}),
+            json.dumps(_msg("user", "This is a real user message here." * 2)),
+        ]
+        result = extract_recent_exchanges(lines, max_exchanges=2)
+        assert len(result) == 1
+        assert result[0]["role"] == "user"
+
+    def test_limits_to_max_exchanges(self):
+        from auto_capture import extract_recent_exchanges
+
+        lines = []
+        for i in range(10):
+            lines.append(json.dumps(_msg("user", f"Question number {i} with enough text to pass." * 2)))
+            lines.append(json.dumps(_msg("assistant", f"Answer number {i} with enough text to pass." * 2)))
+        result = extract_recent_exchanges(lines, max_exchanges=2)
+        assert len(result) == 4
+
+    def test_handles_list_content(self):
+        from auto_capture import extract_recent_exchanges
+
+        lines = [
+            json.dumps({"message": {"role": "user", "content": [
+                {"type": "text", "text": "This is block content that is long enough." * 2},
+            ]}}),
+        ]
+        result = extract_recent_exchanges(lines, max_exchanges=2)
+        assert len(result) == 1
+        assert "block content" in result[0]["content"]
+
+    def test_empty_lines(self):
+        from auto_capture import extract_recent_exchanges
+
+        assert extract_recent_exchanges([], max_exchanges=2) == []
+
+    def test_truncates_long_content(self):
+        from auto_capture import extract_recent_exchanges
+
+        long_text = "x" * 5000
+        lines = [json.dumps(_msg("user", long_text))]
+        result = extract_recent_exchanges(lines, max_exchanges=1)
+        assert len(result) == 1
+        assert len(result[0]["content"]) == 2000
+
+    def test_skips_tool_call_assistant_messages(self):
+        from auto_capture import extract_recent_exchanges
+
+        lines = [
+            json.dumps(_msg("assistant", '{"tool_calls": [{"name": "read"}]}')),
+            json.dumps(_msg("user", "Thanks for reading that file for me!" * 2)),
+        ]
+        result = extract_recent_exchanges(lines, max_exchanges=2)
+        assert len(result) == 1
+        assert result[0]["role"] == "user"

mem0-plugin/tests/test_on_file_read.py

@@ -1,105 +0,0 @@
-"""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

@@ -13,9 +13,12 @@ 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."""
+    """Use a temp dir for the rubric flag file and clean msg counter."""
     monkeypatch.setenv("MEM0_RUBRIC_DIR", str(tmp_path))
+    msg_count_file = "/tmp/mem0_msg_count_testuser"
     yield
+    if os.path.exists(msg_count_file):
+        os.unlink(msg_count_file)
 
 
 def _run_hook(prompt: str, env_overrides: dict | None = None, session_id: str = "test-sess-001") -> str:
@@ -46,8 +49,7 @@ def _run_hook(prompt: str, env_overrides: dict | None = None, session_id: str =
 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 "Mem0 searches apply" in output
     assert "metadata.type" in output
 
 

mem0-plugin/tests/test_write_path.py

@@ -192,4 +192,16 @@ def test_resolve_api_key_returns_empty_when_neither_set(monkeypatch):
     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)
+    monkeypatch.setattr("_identity._extract_key_from_shell_profiles", lambda: "")
     assert resolve_api_key() == ""
+
+
+def test_resolve_api_key_falls_back_to_shell_profile(monkeypatch):
+    """resolve_api_key extracts key from shell profile when env vars are empty."""
+    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)
+    monkeypatch.setattr("_identity._extract_key_from_shell_profiles", lambda: "m0-from-profile")
+    assert resolve_api_key() == "m0-from-profile"