"""Tests for the after_turn hook (stdin/stdout interface).""" import importlib.util import json import subprocess import sys import tempfile from pathlib import Path HOOK = Path(__file__).parent.parent / "hooks" / "after_turn.py" def run_hook(payload: dict, env=None) -> dict: import os run_env = os.environ.copy() if env: run_env.update(env) result = subprocess.run( [sys.executable, str(HOOK)], input=json.dumps(payload), capture_output=True, text=True, env=run_env, ) return json.loads(result.stdout.strip()) def _load_module(env_overrides=None): import os saved = {} if env_overrides: for k, v in env_overrides.items(): saved[k] = os.environ.get(k) os.environ[k] = v spec = importlib.util.spec_from_file_location("after_turn", HOOK) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) for k, v in saved.items(): if v is None: os.environ.pop(k, None) else: os.environ[k] = v return mod _mod = _load_module() # --------------------------------------------------------------------------- # Skip cases # --------------------------------------------------------------------------- def test_empty_messages_skipped(): out = run_hook({"type": "after_turn", "agent_id": "a1", "messages": []}) assert out["status"] == "skip" assert out["reason"] == "no messages" def test_bad_json_skipped(): result = subprocess.run( [sys.executable, str(HOOK)], input="not json", capture_output=True, text=True, ) out = json.loads(result.stdout.strip()) assert out["status"] == "skip" assert out["reason"] == "bad input" def test_short_exchange_skipped(): messages = [ {"role": "user", "content": "hi"}, {"role": "assistant", "content": "hello"}, ] out = run_hook({"type": "after_turn", "agent_id": "a1", "messages": messages}) assert out["status"] == "skip" assert out["reason"] == "too short" def test_irrelevant_exchange_skipped(): messages = [ {"role": "user", "content": "What is the capital of France and why is it historically significant?"}, {"role": "assistant", "content": "Paris has been the capital since the 10th century and is the cultural hub."}, ] out = run_hook({"type": "after_turn", "agent_id": "a1", "messages": messages}) assert out["status"] == "skip" assert out["reason"] == "not relevant" def test_agent_mempalace_tool_call_skipped(): messages = [ {"role": "user", "content": "Save this to my memory please."}, {"role": "tool", "content": '{"tool": "mcp_mempalace_add_drawer", "result": "ok"}'}, {"role": "assistant", "content": "Done, I saved it using mcp_mempalace."}, ] out = run_hook({"type": "after_turn", "agent_id": "a1", "messages": messages}) assert out["status"] == "skip" assert out["reason"] == "agent used mcp_mempalace" def test_user_mentioning_mempalace_not_skipped(): """User saying 'mcp_mempalace' should NOT trigger the dedup skip.""" messages = [ {"role": "user", "content": "Can you use mcp_mempalace to save my dentist appointment on April 15th?"}, {"role": "assistant", "content": "I will remember your dentist appointment on April 15th."}, ] out = run_hook({"type": "after_turn", "agent_id": "a1", "messages": messages}) assert out.get("reason") != "agent used mcp_mempalace" # --------------------------------------------------------------------------- # Code block stripping # --------------------------------------------------------------------------- def test_noise_re_does_not_match_common_prose(): """'exception' and 'traceback' as normal words must not trigger noise filter.""" assert not _mod.NOISE_RE.search("There's no exception to this rule.") assert not _mod.NOISE_RE.search("Let me traceback the history of this decision.") def test_noise_re_matches_python_traceback(): assert _mod.NOISE_RE.search("Traceback (most recent call last):\n File foo.py") def test_noise_re_matches_error_line(): assert _mod.NOISE_RE.search("Exception: something went wrong") assert _mod.NOISE_RE.search("Error: connection refused") def test_code_block_stripped_not_whole_message_skipped(): """A message with a code block but meaningful surrounding text should not be skipped.""" text = "My email is alice@example.com. Here is the script:\n```bash\necho hello\n```\nRun it daily." result = _mod._strip_code_blocks(text) assert "alice@example.com" in result assert "```" not in result assert "[code]" in result def test_pure_code_block_skipped_after_strip(): """A message that is only a code block becomes empty after stripping → too short → skip.""" messages = [ {"role": "user", "content": "here's the script"}, {"role": "assistant", "content": "```python\nfor i in range(10):\n print(i)\n```"}, ] out = run_hook({"type": "after_turn", "agent_id": "a1", "messages": messages}) # Should be skipped due to length or irrelevance after stripping assert out["status"] == "skip" # --------------------------------------------------------------------------- # Multi-room classification # --------------------------------------------------------------------------- def test_classify_single_room(): assert _mod._classify_rooms("Her email is alice@example.com") == [("people", "contacts")] def test_classify_multiple_rooms(): # "meeting" → calendar, "email" → contacts — both should match rooms = _mod._classify_rooms("Schedule a meeting with alice@example.com next Tuesday") assert ("people", "contacts") in rooms assert ("time", "calendar") in rooms def test_classify_finance(): assert _mod._classify_rooms("Invoice for $500 is due next week") == [("finance", "transactions")] def test_classify_logistics(): assert _mod._classify_rooms("The shipment arrived today") == [("logistics", "orders")] def test_classify_decisions(): assert _mod._classify_rooms("We decided to use Postgres going forward") == [("knowledge", "decisions")] def test_classify_default_fallback(): assert _mod._classify_rooms("The sky is blue and the grass is green.") == [("default", "sessions")] # --------------------------------------------------------------------------- # Content deduplication # --------------------------------------------------------------------------- def test_duplicate_detection(): with tempfile.TemporaryDirectory() as tmpdir: mod = _load_module({"MEMPALACE_PALACE_PATH": tmpdir}) text = "I have a dentist appointment on April 15th with Dr. Smith." assert mod._is_duplicate(text, "agent-1") is False # first time: not a duplicate assert mod._is_duplicate(text, "agent-1") is True # second time: duplicate def test_different_content_not_duplicate(): with tempfile.TemporaryDirectory() as tmpdir: mod = _load_module({"MEMPALACE_PALACE_PATH": tmpdir}) assert mod._is_duplicate("dentist appointment April 15th", "agent-1") is False assert mod._is_duplicate("meeting with Bob on Friday afternoon", "agent-1") is False def test_dedup_is_per_agent(): """Two different agents should have independent dedup stores.""" with tempfile.TemporaryDirectory() as tmpdir: mod = _load_module({"MEMPALACE_PALACE_PATH": tmpdir}) text = "I have a dentist appointment on April 15th with Dr. Smith." mod._is_duplicate(text, "agent-1") # agent-1 saves it assert mod._is_duplicate(text, "agent-2") is False # agent-2 has not seen it def test_dedup_rolling_max(): with tempfile.TemporaryDirectory() as tmpdir: mod = _load_module({"MEMPALACE_PALACE_PATH": tmpdir, "MEMPALACE_DEDUP_MAX": "3"}) for i in range(5): mod._is_duplicate(f"unique content number {i} with enough chars to matter", "agent-1") seen = json.loads((Path(tmpdir) / ".after_turn_seen_agent-1.json").read_text()) assert len(seen) == 3 # capped at DEDUP_MAX # --------------------------------------------------------------------------- # Configurable parameters # --------------------------------------------------------------------------- def test_custom_min_chars_env(): mod = _load_module({"MEMPALACE_MIN_CHARS": "200"}) assert mod.MIN_CONTENT_LENGTH == 200 def test_custom_window_size_env(): mod = _load_module({"MEMPALACE_WINDOW_SIZE": "10"}) assert mod.WINDOW_SIZE == 10 # --------------------------------------------------------------------------- # Language detection # --------------------------------------------------------------------------- def test_detect_language_returns_string(): lang = _mod._detect_language("I have a meeting with the client tomorrow afternoon.") assert isinstance(lang, str) assert len(lang) > 0 def test_is_english_known_english(): assert _mod._is_english("en") is True def test_is_english_unknown_treated_as_english(): # unknown = langdetect unavailable or failed → don't block on keyword check assert _mod._is_english("unknown") is True def test_is_english_other_language(): assert _mod._is_english("zh") is False assert _mod._is_english("ja") is False assert _mod._is_english("fr") is False def test_lang_detect_disabled_returns_unknown(): mod = _load_module({"MEMPALACE_LANG_DETECT": "0"}) lang = mod._detect_language("Ich habe morgen einen Termin beim Zahnarzt.") assert lang == "unknown" def test_non_english_skips_relevance_check(): """A non-English exchange long enough to pass length check should not be blocked by the English-only RELEVANCE_RE.""" # This text is in Chinese and contains no English keywords from RELEVANCE_RE, # so without language detection it would be skipped as "not relevant". messages = [ {"role": "user", "content": "我明天下午三点有个牙医预约，在城市医院，请帮我记住这件事。"}, {"role": "assistant", "content": "好的，我已经记住了您明天下午三点在城市医院的牙医预约。"}, ] out = run_hook({"type": "after_turn", "agent_id": "a1", "messages": messages}) # Should NOT be skipped for "not relevant" — may fail at mempalace import (error) or dedup assert out.get("reason") != "not relevant"