solahqb/AgentEvalTool
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat(campaigns): durable scheduler loop with restart recovery and cancel

Browse Source 
Add a thin async loop (run_campaign_loop) that ticks on real wall-clock time,
maps elapsed×time_scale to a window offset via the pure decide_schedule, spawns
due child Runs, and marks the campaign COMPLETED at window end. All authority
lives in the DB (started_at, spawned_indices, status), so the app lifespan can
resume every RUNNING campaign on startup without double-spawning and stop all
loops gracefully on shutdown. A failing plan entry is skipped and recorded
rather than wedging the campaign.

Creating a campaign now starts its loop; POST /api/campaigns/{id}/cancel stops
further spawning (completed child Runs are kept); GET /api/campaigns/{id}
reports live progress (window offset, spawned/completed Run counts).
This commit is contained in:
sinohqb 2026-07-30 13:33:10 +08:00
parent c6b102a9b5
commit 8910fd17e0
6 changed files with 441 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

193
backend/agenteval/evaluation/campaign_runner.py
View File

@ -1,24 +1,30 @@
"""Campaign runner — the side-effect shell around the pure scheduler.
Given a campaign and an injected clock position (real elapsed seconds), it asks
Given a campaign and a clock position (real elapsed seconds), it asks
``campaign_scheduler.decide_schedule`` what is due, then spawns those child Runs
by reusing the existing single-run execution path (``EvalEngine.run`` with an
``existing_run`` that carries ``campaign_id``). Spawned plan-entry indices are
persisted on the campaign so re-advancing the same clock never double-spawns.
This module deliberately has no timer/loop of its own — ticket 02 is driven by
tests or a manual clock advance. The durable async loop and restart recovery
arrive in a later ticket.
This module also hosts the durable scheduler loop: a thin async shell that,
tick by tick, maps real wall-clock elapsed time (since the campaign's persisted
``started_at``) to a window offset and calls ``advance_campaign``. All authority
lives in the DB (window start, spawned progress, status), so the loop can be
torn down and rebuilt on restart without losing or duplicating work.
"""
import asyncio
import logging
from dataclasses import dataclass, field
from datetime import timezone
from typing import Optional
from sqlmodel import Session
from agenteval.evaluation.campaign_scheduler import clock_offset, decide_schedule
from agenteval.evaluation.engine import EvalEngine
from agenteval.models import Campaign, EvalRun, RunStatus, RunTrigger
from agenteval.models import Campaign, CampaignStatus, EvalRun, RunStatus, RunTrigger
from agenteval.storage.db import get_session, utc_now
from agenteval.storage.repository import (
CampaignRepository,
RunRepository,
@ -27,6 +33,18 @@ from agenteval.storage.repository import (
)
_SPAWNED_KEY = "spawned_indices"
_ERRORS_KEY = "errors"
# Real wall-clock seconds between scheduler ticks. time_scale compresses the
# *window*, not the tick cadence — production 24h campaigns still tick slowly.
DEFAULT_TICK_SECONDS = 1.0
_logger = logging.getLogger("agenteval")
# Live loop tasks + cooperative cancel events, keyed by campaign id. Authority
# is the DB; these are only the in-process handles for the running loop.
_tasks: dict[str, asyncio.Task] = {}
_cancel_events: dict[str, asyncio.Event] = {}
@dataclass
@ -37,11 +55,46 @@ class AdvanceResult:
finished: bool = False
def _elapsed_seconds(started_at) -> float:
"""Real seconds since ``started_at`` (tolerating naive UTC from SQLite)."""
if started_at.tzinfo is None:
started_at = started_at.replace(tzinfo=timezone.utc)
return (utc_now() - started_at).total_seconds()
def _spawned_indices(campaign: Campaign) -> set[int]:
scheduler = (campaign.summary or {}).get("scheduler", {})
return set(scheduler.get(_SPAWNED_KEY, []))
def current_window_offset(campaign: Campaign) -> float:
"""The campaign's live window position (seconds), clamped to the window.
RUNNING campaigns derive it from the persisted ``started_at`` and
``time_scale``; not-yet-started campaigns report 0.
"""
if campaign.started_at is None:
return 0.0
offset = clock_offset(
elapsed_seconds=_elapsed_seconds(campaign.started_at),
time_scale=campaign.time_scale,
)
return min(offset, float(campaign.window_seconds))
def campaign_progress(campaign: Campaign, runs: list[EvalRun]) -> dict:
"""Live progress of a campaign, derived from its child Runs.
``completed_runs`` counts only COMPLETED child Runs; failures stay out of
this field (pass_rate semantics are ADR-0002's concern, not this counter).
"""
return {
"current_offset_seconds": current_window_offset(campaign),
"spawned_runs": len(runs),
"completed_runs": sum(1 for r in runs if r.status == RunStatus.COMPLETED),
}
async def _spawn_child_run(campaign: Campaign, scenario_id: str, session: Session) -> str:
"""Create a Run owned by the campaign and drive it through the engine.
@ -75,11 +128,14 @@ async def advance_campaign(
campaign_id: str,
elapsed_seconds: float,
session: Session,
cancel_event: Optional[asyncio.Event] = None,
) -> Optional[AdvanceResult]:
"""Advance the campaign clock to ``elapsed_seconds`` and spawn due Runs.
``elapsed_seconds`` is real wall-clock time since the window started; it is
mapped to a window offset via ``time_scale``. Returns ``None`` if the
mapped to a window offset via ``time_scale``. A due entry whose spawning
raises is skipped and recorded (so one bad entry never wedges the loop),
but still marked spawned to avoid infinite retries. Returns ``None`` if the
campaign does not exist.
"""
repo = CampaignRepository(session)
@ -97,17 +153,138 @@ async def advance_campaign(
)
result = AdvanceResult(finished=decision.finished)
errors: list[dict] = list((campaign.summary or {}).get("scheduler", {}).get(_ERRORS_KEY, []))
for due in decision.due:
# Cancellation stops further spawning; runs already in flight finish.
if cancel_event is not None and cancel_event.is_set():
break
try:
for _ in range(due.entry.count):
run_id = await _spawn_child_run(campaign, due.entry.scenario_id, session)
result.spawned_run_ids.append(run_id)
except Exception as exc: # one entry failing must not wedge the campaign
errors.append({"entry_index": due.index, "error": str(exc)})
_logger.warning("活动 %s 计划条目 %d 派生失败（跳过）: %s", campaign_id, due.index, exc)
spawned.add(due.index)
# Persist progress per entry: a failure partway through a multi-entry
# advance must never lose which entries already spawned, since restart
# recovery (later ticket) reads this back from the DB.
# recovery reads this back from the DB.
scheduler_state: dict = {_SPAWNED_KEY: sorted(spawned)}
if errors:
scheduler_state[_ERRORS_KEY] = errors
summary = dict(campaign.summary or {})
summary["scheduler"] = {_SPAWNED_KEY: sorted(spawned)}
summary["scheduler"] = scheduler_state
campaign.summary = summary
repo.update(campaign)
return result
# ── durable scheduler loop ──────────────────────────────────────────────────
async def run_campaign_loop(campaign_id: str, *, tick_seconds: float = DEFAULT_TICK_SECONDS) -> None:
"""Drive one campaign to completion, ticking on real wall-clock time.
The loop holds no authority: every tick it reloads the campaign from the DB,
recomputes elapsed time from the persisted ``started_at``, and advances.
It exits when the window finishes, the campaign leaves RUNNING (e.g. it was
cancelled), or the cooperative cancel event fires.
"""
cancel = _cancel_events.setdefault(campaign_id, asyncio.Event())
session = get_session()
try:
while not cancel.is_set():
repo = CampaignRepository(session)
campaign = repo.get(campaign_id)
if not campaign or campaign.status != CampaignStatus.RUNNING or campaign.started_at is None:
return
elapsed = _elapsed_seconds(campaign.started_at)
result = await advance_campaign(
campaign_id=campaign_id,
elapsed_seconds=elapsed,
session=session,
cancel_event=cancel,
)
if result and result.finished:
current = repo.get(campaign_id)
# Only complete if still running (not cancelled meanwhile).
if current and current.status == CampaignStatus.RUNNING:
current.status = CampaignStatus.COMPLETED
current.completed_at = utc_now()
repo.update(current)
return
try:
await asyncio.wait_for(cancel.wait(), timeout=tick_seconds)
except asyncio.TimeoutError:
pass
except Exception as exc: # a crashing loop must not take down the app
_logger.warning("活动调度循环 %s 异常退出: %s", campaign_id, exc)
finally:
session.close()
_cancel_events.pop(campaign_id, None)
_tasks.pop(campaign_id, None)
def start_campaign(campaign_id: str, session: Session, *, tick_seconds: float = DEFAULT_TICK_SECONDS) -> Optional[asyncio.Task]:
"""Move a campaign into RUNNING (stamping ``started_at`` on first start) and
launch its loop. Reused for both create-then-start and restart recovery:
a PLANNED campaign gets a fresh ``started_at``; an already-RUNNING one keeps
its original window start so recovery resumes at the correct offset.
"""
repo = CampaignRepository(session)
campaign = repo.get(campaign_id)
if not campaign or campaign.status in (CampaignStatus.COMPLETED, CampaignStatus.CANCELLED, CampaignStatus.FAILED):
return None
if campaign_id in _tasks:
return _tasks[campaign_id]
if campaign.status == CampaignStatus.PLANNED:
campaign.status = CampaignStatus.RUNNING
campaign.started_at = utc_now()
repo.update(campaign)
_cancel_events.setdefault(campaign_id, asyncio.Event())
task = asyncio.create_task(
run_campaign_loop(campaign_id, tick_seconds=tick_seconds),
name=f"campaign-{campaign_id}",
)
_tasks[campaign_id] = task
return task
def request_cancel(campaign_id: str) -> None:
"""Signal the loop (if live) to stop spawning and exit promptly."""
event = _cancel_events.get(campaign_id)
if event is not None:
event.set()
def resume_running_campaigns(session: Session, *, tick_seconds: float = DEFAULT_TICK_SECONDS) -> int:
"""On startup, relaunch a loop for every campaign left in RUNNING."""
resumed = 0
for campaign in CampaignRepository(session).list_all():
if campaign.status == CampaignStatus.RUNNING and campaign.id:
start_campaign(campaign.id, session, tick_seconds=tick_seconds)
resumed += 1
return resumed
async def shutdown_all() -> None:
"""Gracefully stop all live loops (Web app shutdown)."""
for event in list(_cancel_events.values()):
event.set()
tasks = list(_tasks.values())
for task in tasks:
task.cancel()
for task in tasks:
# CancelledError is expected here (we just cancelled the task) and does
# not derive from Exception in 3.8+, so it must be listed explicitly.
try:
await task
except (asyncio.CancelledError, Exception):
pass
_tasks.clear()
_cancel_events.clear()

8
backend/agenteval/storage/repository.py
View File

@ -303,6 +303,14 @@ class RunRepository:
statement = select(EvalRunDB).order_by(EvalRunDB.started_at.desc())
return [_run_from_db(r) for r in self.session.exec(statement).all()]
def list_by_campaign(self, campaign_id: str) -> list[EvalRun]:
statement = (
select(EvalRunDB)
.where(EvalRunDB.campaign_id == campaign_id)
.order_by(EvalRunDB.started_at)
)
return [_run_from_db(r) for r in self.session.exec(statement).all()]
def get(self, run_id: str) -> Optional[EvalRun]:
db = self.session.get(EvalRunDB, run_id)
return _run_from_db(db) if db else None

13
backend/agenteval/web/app.py
View File

@ -27,11 +27,24 @@ async def lifespan(_: FastAPI):
count = RunRepository(session).mark_orphans_failed()
if count:
logging.getLogger("agenteval").warning("启动清理：%d 个中断的运行已标记为 failed", count)
# 据库恢复所有未完成的评估活动，重建其调度循环（不重复派生已到点条目）
from agenteval.evaluation.campaign_runner import resume_running_campaigns
resumed = resume_running_campaigns(session)
if resumed:
logging.getLogger("agenteval").warning("启动恢复：%d 个进行中的评估活动已续跑", resumed)
finally:
session.close()
except Exception as exc:
logging.getLogger("agenteval").warning("启动清理失败（忽略）: %s", exc)
yield
# 优雅停止所有活动调度循环
try:
from agenteval.evaluation.campaign_runner import shutdown_all
await shutdown_all()
except Exception as exc:
logging.getLogger("agenteval").warning("活动调度停止失败（忽略）: %s", exc)
app = FastAPI(

45
backend/agenteval/web/routers/campaigns.py
View File

@ -1,15 +1,25 @@
"""API routes for evaluation campaigns (评估活动).
This ticket covers persistence and create/query only — no scheduling or child
Run spawning. Those arrive in later tickets.
Creating a campaign starts its durable scheduler loop (``campaign_runner``),
which spawns child Runs across the (optionally compressed) service-cycle window
until it finishes. Progress is authoritative in the DB, so detail queries report
the live window position and spawned/completed Run counts, and a campaign can be
cancelled mid-flight.
"""
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field
from sqlmodel import Session
from agenteval.models import Campaign, CampaignPlanEntry
from agenteval.storage.repository import CampaignRepository, ScenarioRepository, TargetRepository
from agenteval.evaluation.campaign_runner import campaign_progress, request_cancel, start_campaign
from agenteval.models import Campaign, CampaignPlanEntry, CampaignStatus
from agenteval.storage.db import utc_now
from agenteval.storage.repository import (
CampaignRepository,
RunRepository,
ScenarioRepository,
TargetRepository,
)
from agenteval.web.deps import get_db
router = APIRouter()
@ -51,7 +61,26 @@ async def create_campaign(
time_scale=request.time_scale,
plan=request.plan,
)
campaign = CampaignRepository(session).create(campaign)
repo = CampaignRepository(session)
campaign = repo.create(campaign)
# Kick off the durable loop; it moves the campaign into RUNNING.
start_campaign(campaign.id, session)
return (repo.get(campaign.id) or campaign).model_dump()
@router.post("/{campaign_id}/cancel")
async def cancel_campaign(campaign_id: str, session: Session = Depends(get_db)) -> dict:
repo = CampaignRepository(session)
campaign = repo.get(campaign_id)
if not campaign:
raise HTTPException(status_code=404, detail="campaign not found")
if campaign.status not in (CampaignStatus.PLANNED, CampaignStatus.RUNNING):
raise HTTPException(status_code=400, detail="campaign is not in a cancellable state")
campaign.status = CampaignStatus.CANCELLED
campaign.completed_at = utc_now()
repo.update(campaign)
request_cancel(campaign_id)
return campaign.model_dump()
@ -60,4 +89,8 @@ async def get_campaign(campaign_id: str, session: Session = Depends(get_db)) ->
campaign = CampaignRepository(session).get(campaign_id)
if not campaign:
raise HTTPException(status_code=404, detail="campaign not found")
return campaign.model_dump()
runs = RunRepository(session).list_by_campaign(campaign_id)
data = campaign.model_dump()
data["progress"] = campaign_progress(campaign, runs)
return data

154
tests/integration/test_campaign_scheduler_loop.py Normal file
View File

@ -0,0 +1,154 @@
"""Integration tests for the durable campaign scheduler loop (ticket 03).
Drives ``campaign_runner.start_campaign`` with a tiny tick and a compressed
time scale so a whole window elapses in a few real milliseconds. Covers:
auto-run-to-completion, restart recovery (no double-spawn, no lost progress),
and cancellation (no further spawning).
"""
from datetime import timedelta
import pytest
from agenteval.evaluation import campaign_runner
from agenteval.evaluation.campaign_runner import request_cancel, start_campaign
from agenteval.models import (
Campaign, CampaignPlanEntry, CampaignStatus, Case, CaseType, ChannelType,
EvalTarget, PlatformType, RunStatus, Scenario, TargetStatus,
)
from agenteval.storage.db import utc_now
from agenteval.storage.repository import CampaignRepository, RunRepository, ScenarioRepository, TargetRepository
from tests.unit.mock_channel import MockChannel
TICK = 0.01
@pytest.fixture()
def seeded_db(db_session, monkeypatch):
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
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(campaign_runner, "get_session", _test_get_session)
channel = MockChannel(reply_delay=0.0)
monkeypatch.setattr(factory_module.ChannelFactory, "create", lambda target: channel)
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,
))
ScenarioRepository(db_session).create(Scenario(
id="s-1", name="mock-scenario",
cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])],
))
return db_session
def _make_campaign(session, **overrides) -> Campaign:
payload = dict(
name="loop",
target_id="t-1",
window_seconds=1,
time_scale=1000.0, # elapsed×1000 → window ends in ~1ms of real time
plan=[
CampaignPlanEntry(scenario_id="s-1", offset_seconds=0, count=1),
CampaignPlanEntry(scenario_id="s-1", offset_seconds=1, count=1),
],
)
payload.update(overrides)
return CampaignRepository(session).create(Campaign(**payload))
async def _await_task(campaign_id, timeout=3.0):
import asyncio
task = campaign_runner._tasks.get(campaign_id)
if task is not None:
await asyncio.wait_for(task, timeout=timeout)
async def test_loop_runs_to_completion(seeded_db):
campaign = _make_campaign(seeded_db)
start_campaign(campaign.id, seeded_db, tick_seconds=TICK)
await _await_task(campaign.id)
final = CampaignRepository(seeded_db).get(campaign.id)
assert final.status == CampaignStatus.COMPLETED
assert final.completed_at is not None
runs = RunRepository(seeded_db).list_by_campaign(campaign.id)
assert len(runs) == 2 # both plan entries spawned exactly once
assert all(r.status == RunStatus.COMPLETED for r in runs)
async def test_restart_recovery_does_not_respawn(seeded_db):
# Simulate a campaign that was already RUNNING before a restart, with its
# window start well in the past and entry 0 already recorded as spawned.
campaign = _make_campaign(seeded_db)
campaign.status = CampaignStatus.RUNNING
campaign.started_at = utc_now() - timedelta(seconds=10)
campaign.summary = {"scheduler": {"spawned_indices": [0]}}
CampaignRepository(seeded_db).update(campaign)
# Recovery relaunches the loop for the already-RUNNING campaign.
start_campaign(campaign.id, seeded_db, tick_seconds=TICK)
await _await_task(campaign.id)
final = CampaignRepository(seeded_db).get(campaign.id)
assert final.status == CampaignStatus.COMPLETED
runs = RunRepository(seeded_db).list_by_campaign(campaign.id)
# Entry 0 was already spawned pre-restart (not re-spawned); only entry 1 runs.
assert len(runs) == 1
async def test_restart_preserves_original_window_start(seeded_db):
original = utc_now() - timedelta(seconds=5)
campaign = _make_campaign(seeded_db, window_seconds=100000, time_scale=1.0)
campaign.status = CampaignStatus.RUNNING
campaign.started_at = original
CampaignRepository(seeded_db).update(campaign)
# Resume must NOT reset started_at (that would rewind the window clock).
start_campaign(campaign.id, seeded_db, tick_seconds=TICK)
request_cancel(campaign.id) # stop the long-window loop promptly
await _await_task(campaign.id)
reloaded = CampaignRepository(seeded_db).get(campaign.id)
assert abs((reloaded.started_at.replace(tzinfo=None) - original.replace(tzinfo=None)).total_seconds()) < 1
async def test_cancel_stops_further_spawning(seeded_db):
import asyncio
# Long real window (scale 1.0) so entry 1 (offset 50s) never comes due fast.
campaign = _make_campaign(
seeded_db, window_seconds=100, time_scale=1.0,
plan=[
CampaignPlanEntry(scenario_id="s-1", offset_seconds=0, count=1),
CampaignPlanEntry(scenario_id="s-1", offset_seconds=50, count=1),
],
)
start_campaign(campaign.id, seeded_db, tick_seconds=TICK)
await asyncio.sleep(0.05) # let the first tick spawn entry 0
# Cancel: mark DB authoritative + signal the live loop.
repo = CampaignRepository(seeded_db)
current = repo.get(campaign.id)
current.status = CampaignStatus.CANCELLED
current.completed_at = utc_now()
repo.update(current)
request_cancel(campaign.id)
await _await_task(campaign.id)
final = repo.get(campaign.id)
assert final.status == CampaignStatus.CANCELLED
runs = RunRepository(seeded_db).list_by_campaign(campaign.id)
assert len(runs) == 1 # entry 1 (offset 50s) never spawned

39
tests/integration/test_campaigns_api.py
View File

@ -23,8 +23,12 @@ def seeded_db(db_session, monkeypatch):
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 campaigns as campaigns_module
monkeypatch.setattr(app_module, "init_db", lambda: None)
# These tests cover persistence/validation/CRUD only — stub out the durable
# scheduler so creation stays PLANNED and no background loop is launched.
monkeypatch.setattr(campaigns_module, "start_campaign", lambda *a, **k: None)
def _test_get_session():
return db_session
@ -177,3 +181,38 @@ async def test_run_can_belong_to_campaign(seeded_db):
assert run.campaign_id == "camp-1"
fetched = repo.get(run.id)
assert fetched.campaign_id == "camp-1"
# ── cancel endpoint + progress in detail (ticket 03; scheduler stubbed) ─────
async def test_cancel_campaign_then_state(client, seeded_db):
campaign_id = (await client.post("/api/campaigns", json=_valid_payload())).json()["id"]
resp = await client.post(f"/api/campaigns/{campaign_id}/cancel")
assert resp.status_code == 200
assert resp.json()["status"] == "cancelled"
got = (await client.get(f"/api/campaigns/{campaign_id}")).json()
assert got["status"] == "cancelled"
async def test_cancel_already_cancelled_rejected(client, seeded_db):
campaign_id = (await client.post("/api/campaigns", json=_valid_payload())).json()["id"]
await client.post(f"/api/campaigns/{campaign_id}/cancel")
again = await client.post(f"/api/campaigns/{campaign_id}/cancel")
assert again.status_code == 400
async def test_cancel_missing_campaign_404(client, seeded_db):
resp = await client.post("/api/campaigns/nope/cancel")
assert resp.status_code == 404
async def test_detail_includes_progress_fields(client, seeded_db):
campaign_id = (await client.post("/api/campaigns", json=_valid_payload())).json()["id"]
got = (await client.get(f"/api/campaigns/{campaign_id}")).json()
assert "progress" in got
progress = got["progress"]
assert progress["spawned_runs"] == 0
assert progress["completed_runs"] == 0
assert progress["current_offset_seconds"] == 0.0