diff --git a/backend/agenteval/evaluation/campaign_runner.py b/backend/agenteval/evaluation/campaign_runner.py index 3534fd2..2339060 100644 --- a/backend/agenteval/evaluation/campaign_runner.py +++ b/backend/agenteval/evaluation/campaign_runner.py @@ -30,6 +30,7 @@ from agenteval.evaluation.campaign_scheduler import ( resolve_finalize, ) from agenteval.evaluation.engine import EvalEngine +from agenteval.exploration.settlement import settle_campaign_sessions from agenteval.models import ( Campaign, CampaignStatus, @@ -201,6 +202,18 @@ def _auto_start_analysis(campaign: Campaign, session: Session) -> None: _logger.warning("活动 %s 自动分析触发失败（已跳过）: %s", campaign.id, exc) +def _settle_exploration(campaign_id: str, session: Session) -> None: + """Expire dangling exploration sessions on completion. + + Never blocks completion: a settlement failure is logged and skipped, the + same non-blocking semantics as ``_auto_start_analysis``. + """ + try: + settle_campaign_sessions(campaign_id, session) + except Exception as exc: + _logger.warning("活动 %s 探索会话结算失败（已跳过）: %s", campaign_id, exc) + + # ── durable scheduler loop ────────────────────────────────────────────────── @@ -250,6 +263,7 @@ async def run_campaign_loop( current.status = CampaignStatus.COMPLETED current.completed_at = utc_now() repo.update(current) + _settle_exploration(campaign_id, session) _auto_start_analysis(current, session) return diff --git a/backend/agenteval/exploration/settlement.py b/backend/agenteval/exploration/settlement.py new file mode 100644 index 0000000..ad8f3f7 --- /dev/null +++ b/backend/agenteval/exploration/settlement.py @@ -0,0 +1,29 @@ +"""Exploration session settlement on campaign finalize (活动终态结算). + +活动窗口结束或被取消进入终态时，仍 running 的探索会话转 expired—— +不留悬挂会话。结算后消息端点按既有状态机拒收（409）。 +""" + +from sqlmodel import Session + +from agenteval.exploration.models import ExplorationSessionStatus +from agenteval.storage.db import utc_now +from agenteval.storage.repository import ExplorationSessionRepository + + +def settle_campaign_sessions(campaign_id: str, session: Session) -> int: + """Expire every still-running exploration session of a finalized campaign. + + Returns the number of sessions expired. Completed/failed sessions keep + their evidence untouched. + """ + repo = ExplorationSessionRepository(session) + expired = 0 + for session_obj in repo.list_by_campaign(campaign_id): + if session_obj.status != ExplorationSessionStatus.RUNNING: + continue + session_obj.status = ExplorationSessionStatus.EXPIRED + session_obj.closed_at = utc_now() + repo.update(session_obj) + expired += 1 + return expired diff --git a/backend/agenteval/web/routers/campaigns.py b/backend/agenteval/web/routers/campaigns.py index 802fdf0..2f966a9 100644 --- a/backend/agenteval/web/routers/campaigns.py +++ b/backend/agenteval/web/routers/campaigns.py @@ -26,6 +26,7 @@ from agenteval.evaluation.report import ( summarize_campaign_progress, ) from agenteval.evaluation.report_render import render_campaign_markdown +from agenteval.exploration.settlement import settle_campaign_sessions from agenteval.exploration.summary import summarize_exploration from agenteval.models import Campaign, CampaignPlanEntry, CampaignStatus, ExplorationBudgetConfig, ExplorationSeeds from agenteval.storage.db import iso_utc, utc_now @@ -122,6 +123,7 @@ async def cancel_campaign(campaign_id: str, session: Session = Depends(get_db)) campaign.completed_at = utc_now() repo.update(campaign) request_cancel(campaign_id) + settle_campaign_sessions(campaign_id, session) return campaign.model_dump() diff --git a/tests/integration/test_exploration_settlement.py b/tests/integration/test_exploration_settlement.py new file mode 100644 index 0000000..0932691 --- /dev/null +++ b/tests/integration/test_exploration_settlement.py @@ -0,0 +1,189 @@ +"""Integration tests for exploration settlement on campaign finalize (v0.9 票据 07). + +Covers both terminal transitions: window completion (durable loop COMPLETE) +and mid-flight cancellation. Running sessions must turn expired and reject +further messages (409), leaving no dangling sessions. +""" + +import asyncio + +import pytest +from agenteval.evaluation import campaign_runner as runner_module +from agenteval.evaluation.campaign_runner import run_campaign_loop +from agenteval.exploration.models import ExplorationSessionStatus +from agenteval.models import ( + Campaign, + CampaignPlanEntry, + Case, + CaseType, + ChannelType, + EvalTarget, + PlatformType, + Scenario, + TargetStatus, +) +from agenteval.storage.db import utc_now +from agenteval.storage.repository import ( + CampaignRepository, + ExplorationSessionRepository, + ScenarioRepository, + TargetRepository, +) +from agenteval.web.app import app +from httpx import ASGITransport, AsyncClient + + +def _session_payload(campaign_id: str) -> dict: + return { + "campaign_id": campaign_id, + "persona": {"name": "巡检用户", "traits": ["耐心"]}, + "goal": "查询账单", + "triggered_by": "manual", + } + + +@pytest.fixture() +def seeded_db(db_session, monkeypatch): + """Point every get_session consumer at the test session and seed the + target + scenario needed by both the loop path and the cancel path.""" + from agenteval.channels import factory as factory_module + from agenteval.evaluation import engine as engine_module + from agenteval.storage import db as db_module + from agenteval.storage import repository as repo_module + from agenteval.web import app as app_module + from agenteval.web.routers import exploration as exploration_router + + 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) + monkeypatch.setattr(engine_module, "get_session", _test_get_session) + monkeypatch.setattr(runner_module, "get_session", _test_get_session) + monkeypatch.setattr(runner_module, "_auto_start_analysis", lambda campaign, session: None) + + from tests.unit.mock_channel import MockChannel + + channel = MockChannel(reply_delay=0.0) + monkeypatch.setattr(factory_module.ChannelFactory, "create", lambda target: channel) + monkeypatch.setattr(exploration_router, "start_judge_review", lambda session_id: None) + + 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 + + target = 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, + ) + TargetRepository(db_session).create(target) + scenario = Scenario( + id="s-1", + name="mock-scenario", + cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])], + ) + ScenarioRepository(db_session).create(scenario) + + yield db_session + app.dependency_overrides.clear() + + +def _make_campaign(db_session, campaign_id: str, *, time_scale: float) -> Campaign: + return CampaignRepository(db_session).create( + Campaign( + id=campaign_id, + name=f"campaign-{campaign_id}", + target_id="t-1", + window_seconds=3600, + time_scale=time_scale, + plan=[CampaignPlanEntry(scenario_id="s-1", offset_seconds=0, count=1)], + status="running", + started_at=utc_now(), + ) + ) + + +async def _create_running_session(db_session, campaign_id: str) -> str: + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + resp = await c.post("/api/exploration/sessions", json=_session_payload(campaign_id)) + assert resp.status_code == 200, resp.text + session_id = resp.json()["id"] + assert ExplorationSessionRepository(db_session).get(session_id).status == (ExplorationSessionStatus.RUNNING) + return session_id + + +async def test_window_completion_expires_running_sessions(seeded_db): + campaign = _make_campaign(seeded_db, "c-done", time_scale=3600.0) + session_id = await _create_running_session(seeded_db, "c-done") + + # time_scale 3600 → 1 real second covers the whole window; the loop + # completes on its first tick and settles exploration sessions. + cancel = asyncio.Event() + await asyncio.wait_for(run_campaign_loop(campaign.id, cancel, tick_seconds=0.05), timeout=5) + + assert CampaignRepository(seeded_db).get("c-done").status.value == "completed" + settled = ExplorationSessionRepository(seeded_db).get(session_id) + assert settled.status == ExplorationSessionStatus.EXPIRED + assert settled.closed_at is not None + + +async def test_cancel_expires_running_sessions_and_rejects_messages(seeded_db): + _make_campaign(seeded_db, "c-1", time_scale=1.0) + session_id = await _create_running_session(seeded_db, "c-1") + + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + resp = await client.post("/api/campaigns/c-1/cancel") + assert resp.status_code == 200, resp.text + assert resp.json()["status"] == "cancelled" + + settled = ExplorationSessionRepository(seeded_db).get(session_id) + assert settled.status == ExplorationSessionStatus.EXPIRED + assert settled.closed_at is not None + + resp = await client.post(f"/api/exploration/sessions/{session_id}/messages", json={"content": "还在吗"}) + assert resp.status_code == 409 + + +async def test_completed_session_survives_settlement(seeded_db): + _make_campaign(seeded_db, "c-1", time_scale=1.0) + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + resp = await client.post("/api/exploration/sessions", json=_session_payload("c-1")) + session_id = resp.json()["id"] + resp = await client.post( + f"/api/exploration/sessions/{session_id}/close", + json={"experience": {"goal_achieved": True}}, + ) + assert resp.status_code == 200 + + await client.post("/api/campaigns/c-1/cancel") + + assert ExplorationSessionRepository(seeded_db).get(session_id).status == (ExplorationSessionStatus.COMPLETED) + + +async def test_production_line_manual_trigger_also_settles(seeded_db): + _make_campaign(seeded_db, "c-1", time_scale=1.0) + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + resp = await client.post( + "/api/exploration/sessions", + json={**_session_payload("c-1"), "triggered_by": "manual"}, + ) + assert resp.status_code == 200, resp.text + session_id = resp.json()["id"] + assert resp.json()["triggered_by"] == "manual" + + # 正式线手动会话同样纳入终态结算，触发来源不影响收口。 + await client.post("/api/campaigns/c-1/cancel") + + assert ExplorationSessionRepository(seeded_db).get(session_id).status == (ExplorationSessionStatus.EXPIRED)