AgentEvalTool/tests/integration/test_exploration_patrol_api.py

"""Integration tests for the exploration patrol API (v0.9 票据 03).

One stateless call returns every running production-line campaign that
participates in exploration, the new results since the last patrol watermark
(reusing the campaign report aggregation), and the remaining exploration
budget. The watermark advances after each call so subsequent calls only
report increments.
"""

from datetime import timedelta

import pytest
from agenteval.models import Campaign, EvalRun, RunStatus, RunSummary
from agenteval.storage.db import utc_now
from agenteval.storage.repository import CampaignRepository, RunRepository
from agenteval.web.app import app
from httpx import ASGITransport, AsyncClient

SEEDS = {"personas": ["急性子用户"], "goals": ["查询账单并缴费"]}


def _make_campaign(campaign_id: str, *, time_scale: float = 1.0, status: str = "running", seeds=SEEDS) -> Campaign:
    return Campaign(
        id=campaign_id,
        name=f"campaign-{campaign_id}",
        target_id="t-1",
        window_seconds=86400,
        time_scale=time_scale,
        plan=[{"scenario_id": "s-1", "offset_seconds": 0, "count": 1}],
        status=status,
        started_at=utc_now() - timedelta(hours=2),
        exploration_seeds=seeds,
    )


def _seed_run(db_session, run_id: str, campaign_id: str, *, pass_rate: float, completed_at) -> None:
    RunRepository(db_session).create(EvalRun(
        id=run_id, target_id="t-1", scenario_id="s-1", campaign_id=campaign_id,
        status=RunStatus.COMPLETED, started_at=completed_at - timedelta(minutes=5),
        completed_at=completed_at,
        summary=RunSummary(total_cases=2, pass_rate=pass_rate, avg_latency_ms=120.0),
    ))


@pytest.fixture()
def seeded_db(db_session, monkeypatch):
    from agenteval.models import ChannelType, EvalTarget, PlatformType, TargetStatus
    from agenteval.storage import db as db_module
    from agenteval.storage import repository as repo_module
    from agenteval.storage.repository import TargetRepository
    from agenteval.web import app as app_module

    monkeypatch.setattr(app_module, "init_db", lambda: None)

    def _test_get_session():
        return db_session

    monkeypatch.setattr(db_module, "get_session", _test_get_session)
    monkeypatch.setattr(repo_module, "get_session", _test_get_session)

    from agenteval.web.deps import get_db

    def _test_get_db():
        try:
            yield db_session
        finally:
            pass

    app.dependency_overrides[get_db] = _test_get_db

    TargetRepository(db_session).create(EvalTarget(
        id="t-1", name="mock-target",
        platform=PlatformType.AI_DIGITAL_EMPLOYEE,
        channel_type=ChannelType.TUTU_API,
        channel_config={"base_url": "http://mock", "token": "x"},
        status=TargetStatus.ACTIVE,
    ))
    repo = CampaignRepository(db_session)
    repo.create(_make_campaign("c-prod"))                        # 正式线，参与探索
    repo.create(_make_campaign("c-fast", time_scale=24.0))      # 加速线 → 不巡检
    repo.create(_make_campaign("c-noseed", seeds=None))         # 无种子集 → 不巡检
    repo.create(_make_campaign("c-done", status="completed"))   # 终态 → 不巡检

    _seed_run(db_session, "r-1", "c-prod", pass_rate=1.0, completed_at=utc_now() - timedelta(hours=1))
    _seed_run(db_session, "r-2", "c-prod", pass_rate=0.5, completed_at=utc_now() - timedelta(minutes=30))

    yield db_session
    app.dependency_overrides.clear()


@pytest.fixture()
async def client():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c


async def _patrol(client) -> dict:
    resp = await client.get("/api/exploration/patrol")
    assert resp.status_code == 200, resp.text
    return resp.json()


async def test_patrol_filters_to_running_production_seeded_campaigns(client, seeded_db):
    body = await _patrol(client)
    ids = [c["campaign_id"] for c in body["campaigns"]]
    assert ids == ["c-prod"]


async def test_patrol_entry_content(client, seeded_db):
    body = await _patrol(client)
    entry = body["campaigns"][0]
    assert entry["campaign_name"] == "campaign-c-prod"
    assert entry["target_id"] == "t-1"
    assert entry["target_name"] == "mock-target"
    assert entry["last_patrolled_at"] is None  # 首次巡检无水位

    new_results = entry["new_results"]
    assert new_results is not None
    assert new_results["summary"]["total_runs"] == 2
    assert new_results["summary"]["overall_pass_rate"] == 0.75
    assert new_results["capability_summary"][0]["scenario_id"] == "s-1"

    budget = entry["budget"]
    assert budget["max_sessions"] == 8
    assert budget["sessions_used"] == 0
    assert budget["remaining_sessions"] == 8
    assert budget["max_turns"] == 12
    assert budget["min_interval_seconds"] == 30 * 60
    assert budget["seconds_since_last_session"] is None


async def test_patrol_watermark_advances_and_reports_increments(client, seeded_db):
    await _patrol(client)
    campaign = CampaignRepository(seeded_db).get("c-prod")
    assert campaign.last_patrolled_at is not None

    # 第二次巡检：无新完成的子运行 → 增量为空
    body = await _patrol(client)
    entry = body["campaigns"][0]
    assert entry["new_results"] is None

    # 水位之后新完成一个子运行 → 第三次巡检只报这一个
    _seed_run(seeded_db, "r-3", "c-prod", pass_rate=0.0, completed_at=utc_now())
    body = await _patrol(client)
    entry = body["campaigns"][0]
    assert entry["new_results"]["summary"]["total_runs"] == 1
    assert entry["new_results"]["summary"]["overall_pass_rate"] == 0.0


async def test_patrol_budget_reflects_existing_sessions(client, seeded_db):
    from agenteval.exploration.models import ExplorationSession
    from agenteval.storage.repository import ExplorationSessionRepository

    ExplorationSessionRepository(seeded_db).create(ExplorationSession(
        campaign_id="c-prod", target_id="t-1", goal="查询账单", persona={"name": "x"},
    ))
    body = await _patrol(client)
    budget = body["campaigns"][0]["budget"]
    assert budget["sessions_used"] == 1
    assert budget["remaining_sessions"] == 7
    assert budget["seconds_since_last_session"] is not None


async def test_patrol_budget_honours_campaign_override(client, seeded_db):
    campaign = CampaignRepository(seeded_db).get("c-prod")
    campaign.exploration_budget = {"max_sessions": 3}
    CampaignRepository(seeded_db).update(campaign)

    body = await _patrol(client)
    budget = body["campaigns"][0]["budget"]
    assert budget["max_sessions"] == 3
    assert budget["remaining_sessions"] == 3


async def test_patrol_migration_column_on_existing_db(tmp_path, monkeypatch):
    """last_patrolled_at applies on a DB at the previous head."""
    from pathlib import Path

    from agenteval.storage import db as db_module
    from alembic import command
    from alembic.config import Config
    from sqlalchemy import create_engine, inspect, text
    from sqlmodel import SQLModel

    database_url = f"sqlite:///{tmp_path / 'patrol.db'}"
    monkeypatch.setattr(db_module, "DATABASE_URL", database_url)
    config = Config(str(Path(__file__).resolve().parents[2] / "alembic.ini"))

    SQLModel.metadata.create_all(create_engine(database_url))
    with create_engine(database_url).begin() as connection:
        connection.execute(text("DROP TABLE IF EXISTS intelligent_eval_messages"))
        connection.execute(text("DROP TABLE IF EXISTS intelligent_eval_sessions"))
        connection.execute(text("DROP TABLE IF EXISTS intelligent_evals"))
        connection.execute(text("DROP TABLE IF EXISTS intelligent_eval_config_snapshots"))
        connection.execute(text("DROP TABLE IF EXISTS intelligent_eval_decision_logs"))
        connection.execute(text("DROP TABLE IF EXISTS intelligent_eval_task_queue"))
        connection.execute(text("DROP TABLE IF EXISTS openclaw_cron_pool"))
        connection.execute(text("DROP TABLE IF EXISTS cron_pool_alert_history"))
        connection.execute(text("DROP TABLE IF EXISTS exploration_sessions"))
        connection.execute(text("DROP TABLE IF EXISTS exploration_messages"))
        connection.execute(text("ALTER TABLE campaigns DROP COLUMN exploration_seeds"))
        connection.execute(text("ALTER TABLE campaigns DROP COLUMN exploration_budget"))
        connection.execute(text("ALTER TABLE campaigns DROP COLUMN last_patrolled_at"))
        connection.execute(text("DROP TABLE IF EXISTS alembic_version"))

    command.stamp(config, "b3c7d9e1f5a2")
    command.upgrade(config, "head")

    cols = {c["name"] for c in inspect(create_engine(database_url)).get_columns("campaigns")}
    assert "last_patrolled_at" in cols