ai-agent/MemoryAgents/backend/tests/test_reme.py

from types import SimpleNamespace

import pytest

import app.adapters.reme as module
from app.adapters.reme import ReMeAgent


def test_reme_heuristic_query_expands_deployment_question():
    queries = ReMeAgent._heuristic_queries("上次部署遇到什么问题？")

    assert queries[0] == "上次部署遇到什么问题？"
    assert "部署" in queries
    assert "索引构建" in queries
    assert "批次" in queries
    assert "超时" in queries


def test_reme_merge_search_results_deduplicates_documents():
    first = "========== daily/2026-07-15.md:5-8 [score=0.4] ==========\n向量索引构建超时"
    second = "========== daily/2026-07-15.md:5-8 [score=0.8] ==========\n向量索引构建超时"

    merged = ReMeAgent._merge_search_results([first, second])

    assert merged.count("daily/2026-07-15.md") == 1
    assert "向量索引构建超时" in merged


@pytest.mark.asyncio
async def test_reme_search_memory_uses_multiple_queries():
    class FakeService:
        def __init__(self):
            self.queries: list[str] = []

        async def run_job(self, job: str, *, query: str, limit: int):
            assert job == "search"
            assert limit == 5
            self.queries.append(query)
            if query == "索引":
                return SimpleNamespace(
                    answer="========== note.md:1 [score=1.0] ==========\n向量索引构建超时"
                )
            return SimpleNamespace(answer="")

    service = FakeService()
    agent = ReMeAgent()

    result, used_queries = await agent._search_memory(service, ["原始问题", "索引"])

    assert service.queries == ["原始问题", "索引"]
    assert used_queries == ["索引"]
    assert "向量索引构建超时" in result


def test_reme_write_classifier_separates_statements_from_queries():
    assert ReMeAgent._should_memorize(
        "上次部署测试环境时遇到向量索引构建超时。"
    ) is True
    assert ReMeAgent._should_memorize(
        "以后修复代码时按这个流程：先复现，再定位，最后验证。"
    ) is True
    assert ReMeAgent._should_memorize("上次部署遇到了什么问题？") is False


def test_reme_disables_automatic_background_and_dream_jobs():
    assert set(ReMeAgent._PASSIVE_JOB_OVERRIDES) == {
        "index_update_loop",
        "resource_watch_loop",
        "digest_watch_loop",
        "dream_cron",
    }
    assert all(
        config == {"backend": "base", "enable_serve": False}
        for config in ReMeAgent._PASSIVE_JOB_OVERRIDES.values()
    )


@pytest.mark.asyncio
async def test_reme_semantic_search_filters_low_similarity(monkeypatch):
    class FakeEmbedding:
        configured = True

        async def embed(self, _texts):
            return [[1.0, 0.0]]

    class FakeRepo:
        async def search_embeddings(self, _system, _query_vector, limit):
            assert limit == 5
            return [
                {
                    "memory_id": "relevant",
                    "source": "daily/relevant.md",
                    "content": "向量索引构建超时，通过分批导入解决。",
                    "score": 0.40,
                },
                {
                    "memory_id": "irrelevant",
                    "source": "daily/irrelevant.md",
                    "content": "前端按钮颜色是蓝色。",
                    "score": 0.20,
                },
            ]

    monkeypatch.setattr(module, "embedding_client", FakeEmbedding())
    agent = ReMeAgent()
    agent.repo = FakeRepo()

    evidence, count = await agent._semantic_search("上次部署遇到了什么问题？")

    assert count == 1
    assert "向量索引构建超时" in evidence
    assert "按钮颜色" not in evidence