From 8910fd17e0b21d1c5ba5a390c82506f87b91e6c8 Mon Sep 17 00:00:00 2001 From: sinohqb Date: Thu, 30 Jul 2026 13:33:10 +0800 Subject: [PATCH] feat(campaigns): durable scheduler loop with restart recovery and cancel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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). --- .../agenteval/evaluation/campaign_runner.py | 199 +++++++++++++++++- backend/agenteval/storage/repository.py | 8 + backend/agenteval/web/app.py | 13 ++ backend/agenteval/web/routers/campaigns.py | 45 +++- .../test_campaign_scheduler_loop.py | 154 ++++++++++++++ tests/integration/test_campaigns_api.py | 39 ++++ 6 files changed, 441 insertions(+), 17 deletions(-) create mode 100644 tests/integration/test_campaign_scheduler_loop.py diff --git a/backend/agenteval/evaluation/campaign_runner.py b/backend/agenteval/evaluation/campaign_runner.py index af4d889..38e1a9b 100644 --- a/backend/agenteval/evaluation/campaign_runner.py +++ b/backend/agenteval/evaluation/campaign_runner.py @@ -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: - 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) + # 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() diff --git a/backend/agenteval/storage/repository.py b/backend/agenteval/storage/repository.py index 61b33a0..ebe0cfc 100644 --- a/backend/agenteval/storage/repository.py +++ b/backend/agenteval/storage/repository.py @@ -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 diff --git a/backend/agenteval/web/app.py b/backend/agenteval/web/app.py index ada84e9..6d22371 100644 --- a/backend/agenteval/web/app.py +++ b/backend/agenteval/web/app.py @@ -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( diff --git a/backend/agenteval/web/routers/campaigns.py b/backend/agenteval/web/routers/campaigns.py index b473e1b..f9c2eca 100644 --- a/backend/agenteval/web/routers/campaigns.py +++ b/backend/agenteval/web/routers/campaigns.py @@ -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 diff --git a/tests/integration/test_campaign_scheduler_loop.py b/tests/integration/test_campaign_scheduler_loop.py new file mode 100644 index 0000000..1339eb3 --- /dev/null +++ b/tests/integration/test_campaign_scheduler_loop.py @@ -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 diff --git a/tests/integration/test_campaigns_api.py b/tests/integration/test_campaigns_api.py index 29c676c..681efdc 100644 --- a/tests/integration/test_campaigns_api.py +++ b/tests/integration/test_campaigns_api.py @@ -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