feat(exploration): expire dangling sessions on campaign finalize

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
 

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

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()
 
 

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)