meida_backend/tests/integration/test_retrieval_cache.py

"""检索缓存集成（T-P2-10）：真实 PostgreSQL + pgvector，DeterministicEmbedder 不联网.

证明单测证明不了的两条失效路径（任务文档 §4，双路径验收）：
1. 发布新实体 + ``invalidate_cache()`` → 同参数检索立即可见新实体；
2. 不失效、等 TTL 自然到期（TTL=1s，sleep 1.1s）→ 新实体同样可见。

Run: MEIDA_RUN_INTEGRATION=1 .venv/bin/pytest -m integration -q --no-cov
"""

from __future__ import annotations

import asyncio
import os
import uuid
from collections.abc import AsyncIterator
from typing import Any

import pytest
from sqlalchemy import delete

from packages.domain.ids import next_id
from packages.retrieval.cache import TtlCache
from packages.retrieval.embedding import DeterministicEmbedder
from packages.retrieval.repo import ChunkInput, upsert_chunks
from packages.retrieval.service import RetrievalService, invalidate_cache
from packages.shared.db import dispose_engine, session_scope
from packages.shared.models import KbAsset, KbChunk, KbEntity, KbGap


def _infra_available() -> bool:
    return bool(os.getenv("MEIDA_RUN_INTEGRATION"))


pytestmark = [
    pytest.mark.integration,
    pytest.mark.skipif(
        not _infra_available(),
        reason="set MEIDA_RUN_INTEGRATION=1 with PostgreSQL and Redis running",
    ),
]


def _unique(name: str) -> str:
    return f"it-t14-{name}-{uuid.uuid4().hex[:10]}"


@pytest.fixture(autouse=True)
async def _hermetic_engine() -> AsyncIterator[None]:
    """每个用例前后重建全局 engine：asyncpg 连接绑定创建它的事件循环，
    pytest-asyncio 给每个用例一个新 loop，复用池里旧连接会报
    "attached to a different loop"。"""
    await dispose_engine()
    yield
    await dispose_engine()


@pytest.fixture
async def clean_kb() -> AsyncIterator[dict[str, list[Any]]]:
    """登记本用例造的脏数据并在结束时清掉：集成库长活，不能留假实体."""
    registry: dict[str, list[Any]] = {"entities": [], "gaps": []}
    yield registry
    async with session_scope() as session:
        if registry["entities"]:
            ids = registry["entities"]
            await session.execute(delete(KbChunk).where(KbChunk.entity_id.in_(ids)))
            await session.execute(delete(KbAsset).where(KbAsset.entity_id.in_(ids)))
            await session.execute(delete(KbEntity).where(KbEntity.entity_id.in_(ids)))
        for query_text in registry["gaps"]:
            await session.execute(delete(KbGap).where(KbGap.query_text == query_text))


async def _publish_entity_with_chunk(
    registry: dict[str, list[Any]], embedder: DeterministicEmbedder, *, chunk_text: str
) -> int:
    """直接以 published 落库（任务文档允许「SQL 置 published」）.

    用例聚焦缓存失效语义，不走发布状态机；chunk 与查询同文保证向量路
    距离 0，必然被召回。
    """
    entity_id = next_id()
    registry["entities"].append(entity_id)
    async with session_scope() as session:
        session.add(
            KbEntity(
                entity_id=entity_id,
                entity_type="item",
                name=_unique("item"),
                attrs={},
                status="published",
            )
        )
        # sessionmaker 是 autoflush=False：不显式 flush，挂起的实体 INSERT
        # 不会先于 kb_chunk 下发，FK 会拦下 chunk。
        await session.flush()
        await upsert_chunks(session, entity_id, [ChunkInput(text=chunk_text)], embedder)
    return entity_id


def _seen_entities(outcome: Any) -> set[int]:
    return {r.entity_id for r in outcome.results}


# --- 1. 发布 + invalidate_cache() → 新实体立即可见（T-P2-10 验收语义）---------
async def test_publish_then_invalidate_makes_new_entity_visible(
    clean_kb: dict[str, list[Any]],
) -> None:
    embedder = DeterministicEmbedder(dimensions=1024)
    query = _unique("invalidate-query")
    # 防御性登记：若回归导致空结果被记进 kb_gap，用例结束也能清干净。
    clean_kb["gaps"].append(query)

    entity_a = await _publish_entity_with_chunk(clean_kb, embedder, chunk_text=query)
    cache: TtlCache[Any] = TtlCache(ttl_seconds=60.0, max_entries=64)

    async with session_scope() as session:
        svc = RetrievalService(session, embedding=embedder, cache=cache)
        first = await svc.search(query)
        second = await svc.search(query)

    assert first.degraded is False and second.degraded is False
    assert _seen_entities(first) == {entity_a}
    assert second == first
    assert cache.stats()["hits"] == 1  # 第二次命中缓存：零 DB、零 embed

    entity_b = await _publish_entity_with_chunk(clean_kb, embedder, chunk_text=query)

    # 未失效前缓存照旧只吐 A —— 证明缓存真的在工作（不是每次都查库）
    async with session_scope() as session:
        svc = RetrievalService(session, embedding=embedder, cache=cache)
        stale = await svc.search(query)
    assert _seen_entities(stale) == {entity_a}

    invalidate_cache()  # 发布编排（admin/T12）调用的钩子

    async with session_scope() as session:
        svc = RetrievalService(session, embedding=embedder, cache=cache)
        fresh = await svc.search(query)
    assert _seen_entities(fresh) == {entity_a, entity_b}  # 新实体立即可见


# --- 2. 不失效，等 TTL 到期 → 新实体自然可见（双保险）-------------------------
async def test_ttl_expiry_makes_new_entity_visible_without_invalidation(
    clean_kb: dict[str, list[Any]],
) -> None:
    embedder = DeterministicEmbedder(dimensions=1024)
    query = _unique("ttl-query")
    clean_kb["gaps"].append(query)

    entity_a = await _publish_entity_with_chunk(clean_kb, embedder, chunk_text=query)
    cache: TtlCache[Any] = TtlCache(ttl_seconds=1.0, max_entries=64)

    async with session_scope() as session:
        svc = RetrievalService(session, embedding=embedder, cache=cache)
        first = await svc.search(query)
        await svc.search(query)
    assert _seen_entities(first) == {entity_a}
    assert cache.stats()["hits"] == 1  # TTL 内命中缓存

    entity_b = await _publish_entity_with_chunk(clean_kb, embedder, chunk_text=query)
    # 不调用 invalidate_cache()：只等 TTL 自然到期（1s + 余量 0.1s）。
    # asyncio.sleep 与 TtlCache 的默认时钟同为单调时钟，无校时跳变风险。
    await asyncio.sleep(1.1)

    async with session_scope() as session:
        svc = RetrievalService(session, embedding=embedder, cache=cache)
        fresh = await svc.search(query)
    assert _seen_entities(fresh) == {entity_a, entity_b}  # 到期后重新检索