AgentLab/backend/tests/test_eval_cleanup.py

"""评测的清理与工件回收（app/eval/cleanup.py、core/artifact_store.forget；细案 01 做法 16）。

- 保留规则：最近 N 次、钉住的基线、A/B 的另几臂，还没结束的不动；
- 清掉的：逐题结果、明细工件、底下的运行（作答、算期望的）和它们的工件引用行；总表永久保留，cleaned_at 写明已清理；
- 工件文件一条引用都不剩才删：被别的运行引用着的同一份内容照留；
- 清理过的评测不能重试、不能重评。
题集、问法都是虚构的。
"""
from __future__ import annotations

from datetime import timedelta

import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy import func, select

from app.core import artifact_store
from app.db.base import SessionLocal, utcnow
from app.db.models import Artifact, Run
from app.db.models_eval import EvalResult, EvalRun
from app.engine.runner import run_manager
from app.eval import answer as answer_mod
from app.eval import cleanup, runner
from app.main import app
from tests import eval_samples as es
from tests.fixtures.sources import drop_source


@pytest.fixture(autouse=True)
async def engine_up():
    await run_manager.setup()
    yield
    await runner.manager.shutdown()
    await es.drop_suites()
    await run_manager.shutdown()


def test_the_keep_rule():
    now = utcnow()

    def run(i: int, **extra) -> EvalRun:
        return EvalRun(id=f"r{i}", suite_id="s", status=extra.pop("status", "succeeded"),
                       created_at=now - timedelta(minutes=i), **extra)

    runs = [run(1), run(2, pair_id="r2"), run(3, pair_id="r2"), run(4, pinned=True), run(5), run(6, status="running")]
    kept = cleanup.keep_set(runs, 2)
    assert kept == {"r1", "r2", "r3", "r4", "r6"}, "最近 2 次 + A/B 另一臂 + 钉住的 + 还没结束的"


async def test_forget_only_removes_files_nobody_refers_to():
    lonely = await artifact_store.put_json({"评测": "只有一处引用", "n": 1}, kind="test_forget")
    shared = await artifact_store.put_json({"评测": "两处引用", "n": 2}, kind="test_forget")
    async with SessionLocal() as session:
        session.add(Artifact(id=shared, run_id="x" * 32, kind="test_forget"))
        await session.execute(Artifact.__table__.delete().where(Artifact.id.in_([lonely, shared]),
                                                                Artifact.run_id == ""))
        await session.commit()
    removed = await artifact_store.forget([lonely, shared, "not-a-hash", lonely])
    assert removed == [lonely]
    assert artifact_store.load(lonely) is None and artifact_store.load(shared)["n"] == 2
    async with SessionLocal() as session:
        await session.execute(Artifact.__table__.delete().where(Artifact.id == shared))
        await session.commit()
    assert await artifact_store.forget([shared]) == [shared]


async def test_cleaning_keeps_the_summary_and_shared_files(monkeypatch, tmp_path):
    async def drive(self):
        self.turn.path, self.turn.reply = "reply", "不查了"
        return self.turn

    monkeypatch.setattr(answer_mod.Driver, "drive", drive)
    name, sid = await es.make_source(tmp_path)
    try:
        suite_id = await es.make_suite(sid)
        made: list[str] = []
        for i in range(3):
            # 中间那一次换一下数：它的期望查询结果只有它自己有（别的两次查到的是同一份，文件共用）
            _amount(tmp_path, 101.0 if i == 1 else 100.0)
            async with SessionLocal() as session:
                from app.db.models_eval import EvalSuite
                from app.eval.variants import normalize_variant

                suite = await session.get(EvalSuite, suite_id)
                [run] = await runner.start(session, suite, [normalize_variant({})], actor="评测测试")
            await es.wait_eval(run.id)
            made.append(run.id)
        first, middle, last = made
        async with SessionLocal() as session:
            (await session.get(EvalRun, first)).pinned = True
            gold_runs = (await session.execute(select(EvalResult.gold_run_ids).where(
                EvalResult.eval_run_id == middle))).scalar_one()
            owned = list((await session.execute(select(Artifact.id).where(Artifact.run_id.in_(gold_runs)))).scalars())
            details = (await session.execute(select(EvalResult.details_artifact).where(
                EvalResult.eval_run_id == middle))).scalar_one()
            # 同一份内容另有一次运行（不是这次评测的）也引用着：文件要留
            shared = owned[0]
            session.add(Artifact(id=shared, run_id="y" * 32, kind="query_snapshot"))
            await session.commit()
        summary_before = (await _run(middle)).summary
        out = await cleanup.clean_suite(suite_id, keep=1)
        assert out["runs"] == 1 and out["results"] == 1 and out["deleted_runs"] == 2 and out["files"] >= 1
        cleaned = await _run(middle)
        assert cleaned.cleaned_at is not None and cleaned.summary == summary_before, "总表永久保留"
        assert (await _run(first)).cleaned_at is None and (await _run(last)).cleaned_at is None
        async with SessionLocal() as session:
            assert await session.scalar(select(func.count()).select_from(EvalResult).where(
                EvalResult.eval_run_id == middle)) == 0
            assert await session.scalar(select(func.count()).select_from(Run).where(Run.id.in_(gold_runs))) == 0
            assert await session.scalar(select(func.count()).select_from(Artifact).where(
                Artifact.run_id.in_(gold_runs))) == 0
        assert artifact_store.load(shared) is not None, "别的运行还引用着的文件不删"
        async with SessionLocal() as session:
            still = set((await session.execute(select(Artifact.id).where(Artifact.id.in_(owned)))).scalars())
        # 同样的内容被另外两次评测的运行引用着的（同一条期望 SQL 查到同一份结果）也留着；一条引用都不剩的才删
        for artifact in owned:
            assert (artifact_store.load(artifact) is not None) == (artifact in still), artifact
        assert any(a not in still for a in owned), "这次评测独有的工件文件删掉了"
        assert artifact_store.load(details) is None, "明细工件跟着删"
        with pytest.raises(runner.EvalRefused):
            await runner.retry(middle)
        async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
            refused = await client.post(f"/api/evals/evaluations/{middle}/rescore")
            assert refused.status_code == 409 and refused.json()["code"] == "eval_details_cleaned"
            compared = (await client.get(f"/api/evals/evaluations/{last}/compare",
                                         params={"base": middle})).json()
            assert compared["cleaned"] is True and compared["transitions"] is None
            again = (await client.post(f"/api/evals/suites/{suite_id}/cleanup")).json()
            assert again["runs"] == 0, "已经清过的不再清"
    finally:
        await drop_source(sid)


def _amount(tmp_path, value: float) -> None:
    import sqlite3

    with sqlite3.connect(tmp_path / "shop.db") as db:
        db.execute("UPDATE orders SET amount = ? WHERE id = 1", (value,))


async def _run(run_id: str) -> EvalRun:
    async with SessionLocal() as session:
        return await session.get(EvalRun, run_id)