plugin-registry/plugins/mempalace-indexer/tests/test_after_turn.py

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