timmy-config/hermes-sovereign/mempalace/tests/test_retrieval_enforcer.py

"""Tests for retrieval_enforcer.py.

Refs: Epic #367, Sub-issue #369
"""

from __future__ import annotations

import json
import os
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch, MagicMock

import pytest

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))

from mempalace.retrieval_enforcer import (
    is_recall_query,
    load_identity,
    load_scratchpad,
    enforce_retrieval_order,
    search_skills,
    RECALL_PATTERNS,
)


class TestRecallDetection:
    """Test the recall-query pattern matcher."""

    @pytest.mark.parametrize("query", [
        "what did we work on yesterday",
        "status of the mempalace integration",
        "remember the fleet audit results",
        "last time we deployed the nexus",
        "previously you mentioned a CI fix",
        "we discussed the sovereign deployment",
    ])
    def test_recall_queries_detected(self, query):
        assert is_recall_query(query) is True

    @pytest.mark.parametrize("query", [
        "create a new file called test.py",
        "run the test suite",
        "deploy to production",
        "write a function that sums numbers",
        "install the package",
    ])
    def test_non_recall_queries_skipped(self, query):
        assert is_recall_query(query) is False


class TestLoadIdentity:
    def test_loads_existing_identity(self, tmp_path):
        identity_file = tmp_path / "identity.txt"
        identity_file.write_text("I am Timmy. A sovereign AI.")
        with patch("mempalace.retrieval_enforcer.IDENTITY_PATH", identity_file):
            result = load_identity()
            assert "Timmy" in result

    def test_returns_empty_on_missing_file(self, tmp_path):
        identity_file = tmp_path / "nonexistent.txt"
        with patch("mempalace.retrieval_enforcer.IDENTITY_PATH", identity_file):
            result = load_identity()
            assert result == ""

    def test_truncates_long_identity(self, tmp_path):
        identity_file = tmp_path / "identity.txt"
        identity_file.write_text(" ".join(["word"] * 300))
        with patch("mempalace.retrieval_enforcer.IDENTITY_PATH", identity_file):
            result = load_identity()
            assert result.endswith("...")
            assert len(result.split()) <= 201  # 200 words + "..."


class TestLoadScratchpad:
    def test_loads_valid_scratchpad(self, tmp_path):
        scratch_file = tmp_path / "session123.json"
        scratch_file.write_text(json.dumps({"note": "test value", "key2": 42}))
        with patch("mempalace.retrieval_enforcer.SCRATCHPAD_DIR", tmp_path):
            result = load_scratchpad("session123")
            assert "note: test value" in result
            assert "key2: 42" in result

    def test_returns_empty_on_missing_file(self, tmp_path):
        with patch("mempalace.retrieval_enforcer.SCRATCHPAD_DIR", tmp_path):
            result = load_scratchpad("nonexistent")
            assert result == ""

    def test_returns_empty_on_invalid_json(self, tmp_path):
        scratch_file = tmp_path / "bad.json"
        scratch_file.write_text("not valid json{{{")
        with patch("mempalace.retrieval_enforcer.SCRATCHPAD_DIR", tmp_path):
            result = load_scratchpad("bad")
            assert result == ""


class TestEnforceRetrievalOrder:
    def test_skips_non_recall_query(self):
        result = enforce_retrieval_order("create a new file")
        assert result["retrieved_from"] is None
        assert result["tokens"] == 0

    def test_runs_for_recall_query(self, tmp_path):
        identity_file = tmp_path / "identity.txt"
        identity_file.write_text("I am Timmy.")
        with patch("mempalace.retrieval_enforcer.IDENTITY_PATH", identity_file), \
             patch("mempalace.retrieval_enforcer.search_palace", return_value=""), \
             patch("mempalace.retrieval_enforcer.search_gitea", return_value=""), \
             patch("mempalace.retrieval_enforcer.search_skills", return_value=""):
            result = enforce_retrieval_order("what did we work on yesterday")
            assert "Identity" in result["context"]
            assert "L0" in result["layers_checked"]

    def test_palace_hit_sets_l1(self, tmp_path):
        identity_file = tmp_path / "identity.txt"
        identity_file.write_text("I am Timmy.")
        with patch("mempalace.retrieval_enforcer.IDENTITY_PATH", identity_file), \
             patch("mempalace.retrieval_enforcer.search_palace", return_value="Found: fleet audit results"), \
             patch("mempalace.retrieval_enforcer.search_gitea", return_value=""):
            result = enforce_retrieval_order("what did we discuss yesterday")
            assert result["retrieved_from"] == "L1"
            assert "Palace Memory" in result["context"]

    def test_falls_through_to_l5(self, tmp_path):
        identity_file = tmp_path / "nonexistent.txt"
        with patch("mempalace.retrieval_enforcer.IDENTITY_PATH", identity_file), \
             patch("mempalace.retrieval_enforcer.search_palace", return_value=""), \
             patch("mempalace.retrieval_enforcer.search_gitea", return_value=""), \
             patch("mempalace.retrieval_enforcer.search_skills", return_value=""):
            result = enforce_retrieval_order("remember the old deployment", skip_if_not_recall=True)
            assert result["retrieved_from"] == "L5"

    def test_force_mode_skips_recall_check(self, tmp_path):
        identity_file = tmp_path / "identity.txt"
        identity_file.write_text("I am Timmy.")
        with patch("mempalace.retrieval_enforcer.IDENTITY_PATH", identity_file), \
             patch("mempalace.retrieval_enforcer.search_palace", return_value=""), \
             patch("mempalace.retrieval_enforcer.search_gitea", return_value=""), \
             patch("mempalace.retrieval_enforcer.search_skills", return_value=""):
            result = enforce_retrieval_order("deploy now", skip_if_not_recall=False)
            assert "Identity" in result["context"]