refactor(architecture): deepen campaign runtime modules

This commit is contained in:
sinohqb 2026-08-11 13:18:48 +08:00
parent 10a089e740
commit 1782b245bf
36 changed files with 1909 additions and 903 deletions

View File

@ -8,12 +8,10 @@
import asyncio
import json
import logging
from typing import Any, Awaitable, Callable, Optional
from sqlmodel import Session
from agenteval.evaluation.report import load_campaign_view
from agenteval.model_gateway import ModelGateway
from agenteval.models import Campaign, ModelCapability, RunStatus
from agenteval.services.model_configs import (
@ -21,18 +19,12 @@ from agenteval.services.model_configs import (
ModelConfigService,
ModelRuntimeConfig,
)
from agenteval.storage.db import get_session
from agenteval.storage.model_config_repository import ModelConfigRepository
from agenteval.storage.repository import (
CampaignAnalysisRepository,
CampaignRepository,
RunRepository,
)
from agenteval.task_registry import TaskRegistry
from agenteval.utils.llm import extract_reply_text, parse_json_from_llm_text
_logger = logging.getLogger("agenteval")
# LLM 客户端协议:接收 chat 消息列表,返回文本内容。生产实现走 ModelGateway
# 测试注入假客户端(同 MockChannel 先例)。
ChatClient = Callable[[list[dict[str, str]]], Awaitable[str]]
@ -242,29 +234,6 @@ async def analyze_campaign(
}
def _maybe_enqueue_period_comparison(campaign: Campaign, session: Session) -> None:
"""正式线活动分析完成后自动链到周期对比v0.8)。
前提正式线time_scale == 1分析模型可解析存在自动基线
同活动串且已有 completed 分析任一不满足静默跳过异常仅告警
不影响刚落库的分析结果
"""
try:
if campaign.time_scale != 1:
return
if resolve_analysis_model(campaign, session) is None:
return
# 延迟导入comparison 顶层依赖 analysisresolve_analysis_model
# 反向导入会成环。
from agenteval.evaluation import comparison as comparison_module
if comparison_module.resolve_auto_baseline(campaign, session) is None:
return
comparison_module.start_campaign_comparison(campaign.id, triggered_by="auto")
except Exception as exc:
_logger.warning("活动 %s 自动周期对比跳过: %s", campaign.id, exc)
def gateway_chat_client(runtime: ModelRuntimeConfig) -> ChatClient:
"""Shared ChatClient factory for analysis/comparison background executors."""
gateway = ModelGateway(timeout=180.0)
@ -273,90 +242,3 @@ def gateway_chat_client(runtime: ModelRuntimeConfig) -> ChatClient:
return await gateway.chat(runtime, messages, temperature=0.2)
return _chat
async def execute_campaign_analysis(
campaign_id: str,
*,
triggered_by: str,
chat_client: Optional[ChatClient] = None,
) -> None:
"""后台执行体generating → completed/failed 状态机upsert每活动一行
Runs 同款后台任务约定自持 Sessiontry/finally 关闭失败落 error
"""
session = get_session()
try:
analyses = CampaignAnalysisRepository(session)
campaign = CampaignRepository(session).get(campaign_id)
if not campaign:
return
runtime = resolve_analysis_model(campaign, session)
if runtime is None:
analyses.upsert(
campaign_id, status="failed", triggered_by=triggered_by,
error="未配置分析模型:请在模型配置中心将某个 chat 配置设为「分析默认」",
)
return
analyses.upsert(
campaign_id, status="generating",
model_config_id=runtime.id, triggered_by=triggered_by,
)
try:
client = chat_client or gateway_chat_client(runtime)
runs = RunRepository(session).list_by_campaign(campaign_id)
view = load_campaign_view(session, campaign)
result = await analyze_campaign(
campaign=campaign,
report=view["report"],
failure_samples=collect_failure_samples(campaign_id, session),
valid_run_ids={r.id for r in runs if r.id},
chat_client=client,
exploration_summary=view["exploration"],
)
except Exception as exc:
_logger.warning("活动 %s 智能分析失败: %s", campaign_id, exc)
analyses.upsert(
campaign_id, status="failed", model_config_id=runtime.id,
error=str(exc)[:500], triggered_by=triggered_by,
)
return
analyses.upsert(
campaign_id, status="completed", result=result,
model_config_id=runtime.id, triggered_by=triggered_by,
)
_maybe_enqueue_period_comparison(campaign, session)
finally:
session.close()
analysis_registry = TaskRegistry()
def start_campaign_analysis(campaign_id: str, *, triggered_by: str) -> None:
"""以后台任务启动分析生成(状态经 campaign_analyses 表观测)。
registry 持强引用防 GCshutdown 时统一收敛 id 在跑时幂等不重复派生
"""
analysis_registry.launch(
campaign_id,
lambda _cancel: execute_campaign_analysis(campaign_id, triggered_by=triggered_by),
)
def enqueue_campaign_analysis(campaign_id: str, *, triggered_by: str) -> None:
"""Durably queue analysis, then launch the in-process worker."""
session = get_session()
try:
CampaignAnalysisRepository(session).enqueue(campaign_id, triggered_by=triggered_by)
finally:
session.close()
start_campaign_analysis(campaign_id, triggered_by=triggered_by)
def resume_queued_campaign_analysis(session: Session) -> int:
"""Re-launch analysis jobs persisted before a process interruption."""
rows = CampaignAnalysisRepository(session).list_queued()
for row in rows:
start_campaign_analysis(row.campaign_id, triggered_by=row.triggered_by or "manual")
return len(rows)

View File

@ -15,11 +15,8 @@ from agenteval.models import Campaign, CampaignPlanEntry, CampaignStatus, Explor
from agenteval.storage.db import utc_now
from agenteval.storage.model_config_repository import ModelConfigRepository
from agenteval.storage.repository import (
CampaignAnalysisRepository,
CampaignPeriodComparisonRepository,
CampaignRepository,
CampaignWriteStatus,
RunRepository,
ScenarioRepository,
TargetRepository,
)
@ -41,21 +38,11 @@ class CampaignLifecycleError(Exception):
detail: str
@dataclass(frozen=True)
class CampaignRecoverySummary:
"""Durable work reconciled during process startup."""
interrupted_runs: int = 0
interrupted_analysis: int = 0
resumed_campaigns: int = 0
resumed_analysis: int = 0
def start_campaign(
session: Session,
campaign_id: str,
*,
launch: Optional[Callable[[str, Session], object]] = None,
launch: Optional[Callable[[str], object]] = None,
) -> Optional[Campaign]:
"""Start a planned Campaign through a conditional lifecycle write."""
repo = CampaignRepository(session)
@ -72,7 +59,7 @@ def start_campaign(
if not result.applied or campaign is None:
return None
if launch is not None and campaign.id:
launch(campaign.id, session)
launch(campaign.id)
return campaign
@ -87,7 +74,7 @@ def create_campaign(
analysis_model_config_id: Optional[str] = None,
exploration_seeds: Optional[ExplorationSeeds] = None,
exploration_budget: Optional[ExplorationBudgetConfig] = None,
launch: Optional[Callable[[str, Session], object]] = None,
launch: Optional[Callable[[str], object]] = None,
) -> Campaign:
"""Validate, commit and then launch one running Campaign.
@ -127,7 +114,7 @@ def create_campaign(
# Deliberately after the repository commit. Startup recovery can relaunch
# this Campaign if the process dies before the callback runs.
if launch is not None and campaign.id:
launch(campaign.id, session)
launch(campaign.id)
return campaign
@ -151,31 +138,6 @@ def cancel_campaign(
return campaign
def recover_campaign_runtime(session: Session, *, tick_seconds: float = 1.0) -> CampaignRecoverySummary:
"""Reconcile all durable Campaign work and relaunch safe tasks.
The database is inspected and repaired before process-local tasks are
launched. Child Run reconciliation remains inside each resumed Campaign
loop, so a pending claim is resumed there while an orphaned running claim
is marked interrupted without replaying messages.
"""
interrupted_runs = RunRepository(session).mark_orphans_failed()
interrupted_analysis = CampaignAnalysisRepository(session).mark_orphans_failed()
interrupted_analysis += CampaignPeriodComparisonRepository(session).mark_orphans_failed()
from agenteval.evaluation.analysis import resume_queued_campaign_analysis
from agenteval.evaluation.campaign_runner import resume_running_campaigns
resumed_campaigns = resume_running_campaigns(session, tick_seconds=tick_seconds)
resumed_analysis = resume_queued_campaign_analysis(session)
return CampaignRecoverySummary(
interrupted_runs=interrupted_runs,
interrupted_analysis=interrupted_analysis,
resumed_campaigns=resumed_campaigns,
resumed_analysis=resumed_analysis,
)
def complete_campaign(
session: Session,
campaign_id: str,

View File

@ -0,0 +1,159 @@
"""Stable read projections for evaluation Campaigns."""
from typing import Any, Optional
from sqlmodel import Session
from agenteval.evaluation.campaign_scheduler import clock_offset, elapsed_seconds
from agenteval.evaluation.comparison import load_comparison_view
from agenteval.evaluation.report import (
build_campaign_timeline,
generate_campaign_report,
summarize_campaign_progress,
)
from agenteval.exploration.summary import summarize_campaign_exploration
from agenteval.storage.db import iso_utc, utc_now
from agenteval.storage.repository import (
CampaignAnalysisRepository,
CampaignRepository,
RunRepository,
ScenarioRepository,
TargetRepository,
)
class CampaignReadModel:
"""One interface for Campaign list, detail, report and export projections."""
def __init__(self, session: Session):
self._session = session
self._campaigns = CampaignRepository(session)
self._runs = RunRepository(session)
self._scenarios = ScenarioRepository(session)
self._analyses = CampaignAnalysisRepository(session)
def list_items(self) -> list[dict[str, Any]]:
campaigns = self._campaigns.list_all()
campaign_ids = [campaign.id for campaign in campaigns if campaign.id]
runs_by_campaign = self._runs.list_by_campaigns(campaign_ids)
items = []
for campaign in campaigns:
data = campaign.model_dump()
data["progress"] = summarize_campaign_progress(
campaign,
runs_by_campaign.get(campaign.id, []),
)
items.append(data)
return items
def detail(self, campaign_id: str) -> Optional[dict[str, Any]]:
campaign = self._campaigns.get(campaign_id)
if campaign is None:
return None
data = campaign.model_dump()
runs = self._runs.list_by_campaign(campaign_id)
current_offset = 0.0
if campaign.started_at is not None:
current_offset = min(
clock_offset(
elapsed_seconds=elapsed_seconds(now=utc_now(), started_at=campaign.started_at),
time_scale=campaign.time_scale,
),
float(campaign.window_seconds),
)
data["progress"] = {
"current_offset_seconds": current_offset,
"spawned_runs": len(runs),
"completed_runs": sum(1 for run in runs if run.status.value == "completed"),
}
return data
def report(self, campaign_id: str) -> Optional[dict[str, Any]]:
campaign = self._campaigns.get(campaign_id)
if campaign is None:
return None
report = self._core_report(campaign)
exploration = summarize_campaign_exploration(self._session, campaign_id)
if exploration is not None:
report["exploration"] = exploration
return report
def timeline(self, campaign_id: str) -> Optional[dict[str, Any]]:
campaign = self._campaigns.get(campaign_id)
if campaign is None:
return None
entries = build_campaign_timeline(
campaign,
self._runs.list_by_campaign(campaign_id),
scenario_names=self._scenarios.name_map(),
)
return {"entries": entries}
def analysis(self, campaign_id: str) -> Optional[dict[str, Any]]:
if self._campaigns.get(campaign_id) is None:
return None
row = self._analyses.get_by_campaign(campaign_id)
if row is None:
return {"status": "none"}
return {
"status": row.status,
"result": row.get_result(),
"error": row.error,
"model_config_id": row.model_config_id,
"triggered_by": row.triggered_by,
"updated_at": iso_utc(row.updated_at),
}
def comparison(self, campaign_id: str) -> Optional[dict[str, Any]]:
campaign = self._campaigns.get(campaign_id)
return load_comparison_view(self._session, campaign) if campaign is not None else None
def full_view(self, campaign_id: str) -> Optional[dict[str, Any]]:
campaign = self._campaigns.get(campaign_id)
if campaign is None:
return None
report = self._core_report(campaign)
exploration = summarize_campaign_exploration(self._session, campaign_id)
analysis = self.analysis(campaign_id)
comparison = load_comparison_view(self._session, campaign)
return {
"report": report,
"exploration": exploration,
"analysis": analysis.get("result") if analysis and analysis.get("status") == "completed" else None,
"comparison": comparison if comparison.get("status") != "none" else None,
}
def markdown_projection(self, campaign_id: str) -> Optional[dict[str, Any]]:
campaign = self._campaigns.get(campaign_id)
view = self.full_view(campaign_id)
if campaign is None or view is None:
return None
comparison = None
comparison_view = view["comparison"]
if comparison_view and comparison_view.get("status") == "completed":
row = comparison_view.get("comparison") or {}
baseline = row.get("baseline") or {}
comparison = {
"result": row.get("result"),
"baseline_name": baseline.get("name"),
"baseline_completed_at": baseline.get("completed_at"),
"model_name": row.get("model_name"),
"updated_at": row.get("updated_at"),
"metric_diff": comparison_view.get("metric_diff"),
}
target = TargetRepository(self._session).get(campaign.target_id)
return {
**view,
"comparison": comparison,
"target_name": target.name if target else None,
"scenario_names": self._scenarios.name_map(),
}
def _core_report(self, campaign) -> dict[str, Any]:
return generate_campaign_report(
campaign,
self._runs.list_by_campaign(campaign.id),
scenario_names=self._scenarios.name_map(),
)

View File

@ -7,21 +7,21 @@ plan occurrence before reusing the existing single-run execution path
primary idempotency authority; the legacy campaign summary remains a fallback
for pre-identity Runs created before this migration.
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.
``CampaignRuntime`` owns the durable scheduler loop and exposes only lifecycle-
level operations. Clock mapping, reconciliation, ticking, claims and process-
local task handles remain implementation details. All authority lives in the
DB, so loops can be rebuilt on restart without losing or duplicating work.
"""
import asyncio
import logging
from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime
from typing import Awaitable, Callable, Optional
from sqlmodel import Session
from agenteval.evaluation.analysis import enqueue_campaign_analysis, resolve_analysis_model
from agenteval.evaluation.analysis import resolve_analysis_model
from agenteval.evaluation.campaign_lifecycle import complete_campaign
from agenteval.evaluation.campaign_lifecycle import start_campaign as start_campaign_lifecycle
from agenteval.evaluation.campaign_scheduler import (
@ -33,6 +33,7 @@ from agenteval.evaluation.campaign_scheduler import (
resolve_finalize,
)
from agenteval.evaluation.engine import EvalEngine
from agenteval.evaluation.intelligence_jobs import enqueue_campaign_analysis
from agenteval.models import (
Campaign,
CampaignStatus,
@ -40,6 +41,7 @@ from agenteval.models import (
EvalRun,
RunStatus,
RunTrigger,
Scenario,
SchedulerState,
)
from agenteval.storage.db import get_session, utc_now
@ -58,9 +60,8 @@ DEFAULT_TICK_SECONDS = 1.0
_logger = logging.getLogger("agenteval")
# Live loop tasks + cooperative cancel events, keyed by campaign id. Authority
# is the DB; this registry only holds the in-process handles for the loop.
campaign_registry = TaskRegistry()
ChildRunExecutor = Callable[[Campaign, Scenario, EvalRun, Session], Awaitable[None]]
@dataclass
@ -79,6 +80,14 @@ class CampaignRecoveryResult:
failed_run_ids: list[str] = field(default_factory=list)
@dataclass(frozen=True)
class CampaignRuntimeRecovery:
"""Durable Campaign work reconciled during process startup."""
interrupted_runs: int = 0
resumed_campaigns: int = 0
def _spawned_indices(campaign: Campaign, session: Optional[Session] = None) -> set[int]:
"""Return plan entries completed by durable identities or legacy progress.
@ -99,40 +108,22 @@ def _spawned_indices(campaign: Campaign, session: Optional[Session] = None) -> s
return {
plan_index
for plan_index, entry in enumerate(campaign.plan)
if (
plan_index in occurrences
and set(range(entry.count)).issubset(occurrences[plan_index])
)
if (plan_index in occurrences and set(range(entry.count)).issubset(occurrences[plan_index]))
or (plan_index not in occurrences and plan_index in legacy)
}
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(now=utc_now(), started_at=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 _execute_child_run(
campaign: Campaign,
scenario: Scenario,
run: EvalRun,
session: Session,
) -> None:
target = TargetRepository(session).get(campaign.target_id)
if not target:
raise ValueError(f"campaign target missing: {campaign.target_id}")
engine = EvalEngine(target=target, scenario=scenario, triggered_by=RunTrigger.CAMPAIGN)
await engine.run(existing_run=run)
async def _spawn_child_run(
@ -142,6 +133,7 @@ async def _spawn_child_run(
plan_index: int,
occurrence_index: int,
session: Session,
execute_child_run: ChildRunExecutor = _execute_child_run,
) -> Optional[str]:
"""Claim one occurrence and drive its pending Run through the engine.
@ -168,20 +160,17 @@ async def _spawn_child_run(
raise RuntimeError(f"Campaign child claim returned no Run: {claim.status.value}")
if run.status is RunStatus.PENDING:
target = TargetRepository(session).get(campaign.target_id)
if not target:
raise ValueError(f"campaign target missing: {campaign.target_id}")
engine = EvalEngine(target=target, scenario=scenario, triggered_by=RunTrigger.CAMPAIGN)
await engine.run(existing_run=run)
await execute_child_run(campaign, scenario, run, session)
return run.id or ""
async def advance_campaign(
async def _advance_campaign(
*,
campaign_id: str,
elapsed_seconds: float,
session: Session,
cancel_event: Optional[asyncio.Event] = None,
execute_child_run: ChildRunExecutor = _execute_child_run,
) -> Optional[AdvanceResult]:
"""Advance the campaign clock to ``elapsed_seconds`` and spawn due Runs.
@ -219,6 +208,7 @@ async def advance_campaign(
plan_index=due.index,
occurrence_index=occurrence_index,
session=session,
execute_child_run=execute_child_run,
)
if run_id is None:
claim_rejected = True
@ -259,11 +249,12 @@ def _auto_start_analysis(campaign: Campaign, session: Session) -> None:
_logger.warning("活动 %s 自动分析触发失败(已跳过): %s", campaign.id, exc)
async def reconcile_campaign_child_runs(
async def _reconcile_campaign_child_runs(
campaign_id: str,
session: Session,
*,
cancel_event: Optional[asyncio.Event] = None,
execute_child_run: ChildRunExecutor = _execute_child_run,
) -> Optional[CampaignRecoveryResult]:
"""Resume safe pending claims and fail running Runs without replaying.
@ -278,9 +269,7 @@ async def reconcile_campaign_child_runs(
return CampaignRecoveryResult()
run_repo = RunRepository(session)
result = CampaignRecoveryResult(
failed_run_ids=run_repo.mark_campaign_running_interrupted(campaign_id)
)
result = CampaignRecoveryResult(failed_run_ids=run_repo.mark_campaign_running_interrupted(campaign_id))
for run in run_repo.list_pending_campaign_children(campaign_id):
if cancel_event is not None and cancel_event.is_set():
break
@ -306,6 +295,7 @@ async def reconcile_campaign_child_runs(
plan_index=plan_index,
occurrence_index=occurrence_index,
session=session,
execute_child_run=execute_child_run,
)
if resumed_id is None:
break
@ -326,8 +316,14 @@ async def reconcile_campaign_child_runs(
# ── durable scheduler loop ──────────────────────────────────────────────────
async def run_campaign_loop(
campaign_id: str, cancel: asyncio.Event, *, tick_seconds: float = DEFAULT_TICK_SECONDS
async def _run_campaign_loop(
campaign_id: str,
cancel: asyncio.Event,
*,
session_factory: Callable[[], Session],
now: Callable[[], datetime],
tick_seconds: float,
execute_child_run: ChildRunExecutor,
) -> None:
"""Drive one campaign to completion, ticking on real wall-clock time.
@ -336,18 +332,23 @@ async def run_campaign_loop(
It exits when the window finishes, the campaign leaves RUNNING (e.g. it was
cancelled), or the cooperative cancel event fires.
"""
session = get_session()
session = session_factory()
try:
await reconcile_campaign_child_runs(campaign_id, session, cancel_event=cancel)
await _reconcile_campaign_child_runs(
campaign_id,
session,
cancel_event=cancel,
execute_child_run=execute_child_run,
)
while not cancel.is_set():
repo = CampaignRepository(session)
campaign = repo.get(campaign_id)
if not campaign:
return
now = utc_now()
current_time = now()
decision = decide_tick(
now=now,
now=current_time,
started_at=campaign.started_at,
status=campaign.status,
window_seconds=campaign.window_seconds,
@ -358,12 +359,13 @@ async def run_campaign_loop(
if decision.action is TickAction.STOP:
return
elapsed = elapsed_seconds(now=now, started_at=campaign.started_at)
await advance_campaign(
elapsed = elapsed_seconds(now=current_time, started_at=campaign.started_at)
await _advance_campaign(
campaign_id=campaign_id,
elapsed_seconds=elapsed,
session=session,
cancel_event=cancel,
execute_child_run=execute_child_run,
)
if decision.action is TickAction.COMPLETE:
@ -388,41 +390,74 @@ async def run_campaign_loop(
session.close()
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.
"""
campaign = CampaignRepository(session).get(campaign_id)
if not campaign or campaign.status in (CampaignStatus.COMPLETED, CampaignStatus.CANCELLED, CampaignStatus.FAILED):
return None
class CampaignRuntime:
"""Own durable Campaign execution behind one lifecycle-level interface."""
started = start_campaign_lifecycle(session, campaign_id)
if started is None:
return None
def __init__(
self,
*,
session_factory: Callable[[], Session] = get_session,
now: Callable[[], datetime] = utc_now,
tick_seconds: float = DEFAULT_TICK_SECONDS,
execute_child_run: ChildRunExecutor = _execute_child_run,
) -> None:
self._session_factory = session_factory
self._now = now
self._tick_seconds = tick_seconds
self._execute_child_run = execute_child_run
self._registry = TaskRegistry()
return campaign_registry.launch(
campaign_id,
lambda cancel: run_campaign_loop(campaign_id, cancel, tick_seconds=tick_seconds),
)
def start(self, campaign_id: str) -> bool:
"""Start or resume one durable Campaign, idempotently."""
session = self._session_factory()
try:
started = start_campaign_lifecycle(session, campaign_id)
finally:
session.close()
if started is None:
return False
self._launch(campaign_id)
return True
def cancel(self, campaign_id: str) -> None:
"""Signal the process-local loop after durable cancellation commits."""
self._registry.cancel(campaign_id)
def recover(self) -> CampaignRuntimeRecovery:
"""Repair interrupted Runs and relaunch every running Campaign."""
session = self._session_factory()
try:
interrupted_runs = RunRepository(session).mark_orphans_failed()
running_ids = [
campaign.id
for campaign in CampaignRepository(session).list_all()
if campaign.status is CampaignStatus.RUNNING and campaign.id
]
finally:
session.close()
for campaign_id in running_ids:
self._launch(campaign_id)
return CampaignRuntimeRecovery(
interrupted_runs=interrupted_runs,
resumed_campaigns=len(running_ids),
)
async def shutdown(self) -> None:
"""Gracefully stop all process-local Campaign loops."""
await self._registry.shutdown_all()
def _launch(self, campaign_id: str) -> None:
self._registry.launch(
campaign_id,
lambda cancel: _run_campaign_loop(
campaign_id,
cancel,
session_factory=self._session_factory,
now=self._now,
tick_seconds=self._tick_seconds,
execute_child_run=self._execute_child_run,
),
)
def request_cancel(campaign_id: str) -> None:
"""Signal the loop (if live) to stop spawning and exit promptly."""
campaign_registry.cancel(campaign_id)
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)."""
await campaign_registry.shutdown_all()
campaign_runtime = CampaignRuntime()

View File

@ -7,27 +7,23 @@ diff——全部确定性计算ADR-0004 口径,经 ``generate_campaign_repo
"""
import json
import logging
from datetime import datetime, timezone
from typing import Any, Optional
from sqlmodel import Session
from agenteval.evaluation.analysis import ChatClient, gateway_chat_client, resolve_analysis_model
from agenteval.evaluation.analysis import ChatClient, resolve_analysis_model
from agenteval.evaluation.report import load_campaign_report
from agenteval.models import Campaign, CampaignStatus
from agenteval.storage.db import get_session, iso_utc, utc_now
from agenteval.storage.db import iso_utc, utc_now
from agenteval.storage.model_config_repository import ModelConfigRepository
from agenteval.storage.repository import (
CampaignAnalysisRepository,
CampaignPeriodComparisonRepository,
CampaignRepository,
)
from agenteval.task_registry import TaskRegistry
from agenteval.utils.llm import parse_json_from_llm_text
_logger = logging.getLogger("agenteval")
_SAME_MOMENT_EPS = 1e-3
_VALID_TRENDS = {"improving", "stable", "regressing"}
@ -377,107 +373,3 @@ async def narrate_period_comparison(
"problem_evolution": problem_evolution,
"suggestion_tracking": suggestion_tracking,
}
async def execute_campaign_comparison(
campaign_id: str,
*,
triggered_by: str,
baseline_campaign_id: Optional[str] = None,
chat_client: Optional[ChatClient] = None,
) -> None:
"""后台执行体generating → completed/failed 状态机upsert每活动一行
基线优先取显式指定手动重选缺省走自动基线 Runs 同款后台任务
约定自持 Sessiontry/finally 关闭失败落 error
"""
session = get_session()
try:
comparisons = CampaignPeriodComparisonRepository(session)
campaign = CampaignRepository(session).get(campaign_id)
if not campaign:
return
try:
baseline = validate_comparison_request(
session,
campaign,
explicit_baseline_id=baseline_campaign_id,
)
except ComparisonError as exc:
comparisons.upsert(
campaign_id,
status="failed",
triggered_by=triggered_by,
error=str(exc),
)
return
runtime = resolve_analysis_model(campaign, session)
baseline_analysis = CampaignAnalysisRepository(session).get_by_campaign(baseline.id)
current_analysis = CampaignAnalysisRepository(session).get_by_campaign(campaign_id)
comparisons.upsert(
campaign_id,
status="generating",
baseline_campaign_id=baseline.id,
model_config_id=runtime.id,
triggered_by=triggered_by,
)
try:
diff = compute_metric_diff(
load_campaign_report(session, baseline),
load_campaign_report(session, campaign),
)
valid_scenario_ids = {s["scenario_id"] for s in diff["scenarios"]}
result = await narrate_period_comparison(
baseline_analysis=baseline_analysis.get_result(),
current_analysis=current_analysis.get_result(),
metric_diff=diff,
valid_scenario_ids=valid_scenario_ids,
chat_client=chat_client or gateway_chat_client(runtime),
)
except Exception as exc:
_logger.warning("活动 %s 周期对比失败: %s", campaign_id, exc)
comparisons.upsert(
campaign_id,
status="failed",
baseline_campaign_id=baseline.id,
model_config_id=runtime.id,
error=str(exc)[:500],
triggered_by=triggered_by,
)
return
comparisons.upsert(
campaign_id,
status="completed",
baseline_campaign_id=baseline.id,
result=result,
model_config_id=runtime.id,
triggered_by=triggered_by,
)
finally:
session.close()
comparison_registry = TaskRegistry()
def start_campaign_comparison(
campaign_id: str,
*,
triggered_by: str,
baseline_campaign_id: Optional[str] = None,
) -> None:
"""以后台任务启动对比生成(状态经 campaign_period_comparisons 表观测)。
registry 持强引用防 GCshutdown 时统一收敛 id 在跑时幂等不重复派生
"""
comparison_registry.launch(
campaign_id,
lambda _cancel: execute_campaign_comparison(
campaign_id,
triggered_by=triggered_by,
baseline_campaign_id=baseline_campaign_id,
),
)

View File

@ -0,0 +1,347 @@
"""Durable runtime for Campaign intelligence jobs.
智能分析与周期对比是两个领域工作 adapter module 统一掌握它们的
持久排队进程内幂等启动重启恢复和关闭顺序数据库行是耐久权威
TaskRegistry 只保存当前进程中的任务句柄
"""
import logging
from collections.abc import Callable
from typing import Any, Optional
from sqlmodel import Session
from agenteval.storage.db import get_session
from agenteval.storage.repository import (
CampaignAnalysisRepository,
CampaignPeriodComparisonRepository,
)
from agenteval.task_registry import TaskRegistry
_registry = TaskRegistry()
_logger = logging.getLogger("agenteval")
MAX_QUEUED_RECOVERY_ATTEMPTS = 3
def _job_key(kind: str, campaign_id: str) -> str:
return f"{kind}:{campaign_id}"
def _launch_analysis(campaign_id: str, *, triggered_by: str) -> None:
async def run(_cancel) -> None:
await execute_campaign_analysis_job(campaign_id, triggered_by=triggered_by)
_registry.launch(_job_key("analysis", campaign_id), run)
def _launch_comparison(
campaign_id: str,
*,
triggered_by: str,
baseline_campaign_id: str,
) -> None:
async def run(_cancel) -> None:
await execute_campaign_comparison_job(
campaign_id,
triggered_by=triggered_by,
baseline_campaign_id=baseline_campaign_id,
)
_registry.launch(_job_key("comparison", campaign_id), run)
async def execute_campaign_analysis_job(
campaign_id: str,
*,
triggered_by: str,
chat_client: Any = None,
session_factory: Optional[Callable[[], Session]] = None,
) -> None:
"""Claim and settle one intelligent-analysis job."""
from agenteval.evaluation.analysis import (
analyze_campaign,
collect_failure_samples,
gateway_chat_client,
resolve_analysis_model,
)
from agenteval.evaluation.comparison import resolve_auto_baseline
from agenteval.evaluation.report import load_campaign_view
from agenteval.storage.repository import CampaignRepository, RunRepository
session = (session_factory or get_session)()
try:
analyses = CampaignAnalysisRepository(session)
row = analyses.get_by_campaign(campaign_id)
if row is None:
row = analyses.enqueue(campaign_id, triggered_by=triggered_by)
if not analyses.claim_queued(campaign_id).claimed:
return
effective_trigger = row.triggered_by or triggered_by
campaign = CampaignRepository(session).get(campaign_id)
if campaign is None:
analyses.upsert(
campaign_id,
status="failed",
triggered_by=effective_trigger,
error="campaign not found",
)
return
runtime = resolve_analysis_model(campaign, session)
if runtime is None:
analyses.upsert(
campaign_id,
status="failed",
triggered_by=effective_trigger,
error="未配置分析模型:请在模型配置中心将某个 chat 配置设为「分析默认」",
)
return
analyses.upsert(
campaign_id,
status="generating",
model_config_id=runtime.id,
triggered_by=effective_trigger,
)
try:
client = chat_client or gateway_chat_client(runtime)
runs = RunRepository(session).list_by_campaign(campaign_id)
view = load_campaign_view(session, campaign)
result = await analyze_campaign(
campaign=campaign,
report=view["report"],
failure_samples=collect_failure_samples(campaign_id, session),
valid_run_ids={run.id for run in runs if run.id},
chat_client=client,
exploration_summary=view["exploration"],
)
except Exception as exc:
_logger.warning("活动 %s 智能分析失败: %s", campaign_id, exc)
analyses.upsert(
campaign_id,
status="failed",
model_config_id=runtime.id,
error=str(exc)[:500],
triggered_by=effective_trigger,
)
return
analyses.upsert(
campaign_id,
status="completed",
result=result,
model_config_id=runtime.id,
triggered_by=effective_trigger,
)
try:
if campaign.time_scale != 1 or resolve_analysis_model(campaign, session) is None:
return
baseline = resolve_auto_baseline(campaign, session)
if baseline is None:
return
enqueue_campaign_comparison(
campaign.id,
triggered_by="auto",
baseline_campaign_id=baseline.id,
session=session,
)
except Exception as exc:
_logger.warning("活动 %s 自动周期对比跳过: %s", campaign.id, exc)
finally:
session.close()
async def execute_campaign_comparison_job(
campaign_id: str,
*,
triggered_by: str,
baseline_campaign_id: Optional[str] = None,
chat_client: Any = None,
session_factory: Optional[Callable[[], Session]] = None,
) -> None:
"""Claim and settle one period-comparison job."""
from agenteval.evaluation.analysis import gateway_chat_client, resolve_analysis_model
from agenteval.evaluation.comparison import (
ComparisonError,
compute_metric_diff,
narrate_period_comparison,
validate_comparison_request,
)
from agenteval.evaluation.report import load_campaign_report
from agenteval.storage.repository import CampaignRepository
session = (session_factory or get_session)()
try:
comparisons = CampaignPeriodComparisonRepository(session)
campaign = CampaignRepository(session).get(campaign_id)
if campaign is None:
return
row = comparisons.get_by_campaign(campaign_id)
if row is None:
try:
initial_baseline = validate_comparison_request(
session,
campaign,
explicit_baseline_id=baseline_campaign_id,
)
except ComparisonError as exc:
comparisons.upsert(
campaign_id,
status="failed",
triggered_by=triggered_by,
error=str(exc),
)
return
row = comparisons.enqueue(
campaign_id,
baseline_campaign_id=initial_baseline.id,
triggered_by=triggered_by,
)
if not comparisons.claim_queued(campaign_id).claimed:
return
effective_trigger = row.triggered_by or triggered_by
effective_baseline_id = baseline_campaign_id or row.baseline_campaign_id
try:
baseline = validate_comparison_request(
session,
campaign,
explicit_baseline_id=effective_baseline_id,
)
except ComparisonError as exc:
comparisons.upsert(
campaign_id,
status="failed",
baseline_campaign_id=effective_baseline_id,
triggered_by=effective_trigger,
error=str(exc),
)
return
runtime = resolve_analysis_model(campaign, session)
baseline_analysis = CampaignAnalysisRepository(session).get_by_campaign(baseline.id)
current_analysis = CampaignAnalysisRepository(session).get_by_campaign(campaign_id)
comparisons.upsert(
campaign_id,
status="generating",
baseline_campaign_id=baseline.id,
model_config_id=runtime.id,
triggered_by=effective_trigger,
)
try:
diff = compute_metric_diff(
load_campaign_report(session, baseline),
load_campaign_report(session, campaign),
)
result = await narrate_period_comparison(
baseline_analysis=baseline_analysis.get_result(),
current_analysis=current_analysis.get_result(),
metric_diff=diff,
valid_scenario_ids={item["scenario_id"] for item in diff["scenarios"]},
chat_client=chat_client or gateway_chat_client(runtime),
)
except Exception as exc:
_logger.warning("活动 %s 周期对比失败: %s", campaign_id, exc)
comparisons.upsert(
campaign_id,
status="failed",
baseline_campaign_id=baseline.id,
model_config_id=runtime.id,
error=str(exc)[:500],
triggered_by=effective_trigger,
)
return
comparisons.upsert(
campaign_id,
status="completed",
baseline_campaign_id=baseline.id,
result=result,
model_config_id=runtime.id,
triggered_by=effective_trigger,
)
finally:
session.close()
def enqueue_campaign_analysis(
campaign_id: str,
*,
triggered_by: str,
session: Optional[Session] = None,
) -> None:
"""Persist an analysis job, then launch its process-local worker."""
owns_session = session is None
active_session = session or get_session()
try:
row = CampaignAnalysisRepository(active_session).enqueue(
campaign_id,
triggered_by=triggered_by,
)
finally:
if owns_session:
active_session.close()
if row.status == "queued":
_launch_analysis(campaign_id, triggered_by=row.triggered_by or triggered_by)
def enqueue_campaign_comparison(
campaign_id: str,
*,
triggered_by: str,
baseline_campaign_id: str,
session: Optional[Session] = None,
) -> None:
"""Persist a comparison job, then launch its process-local worker."""
owns_session = session is None
active_session = session or get_session()
try:
row = CampaignPeriodComparisonRepository(active_session).enqueue(
campaign_id,
baseline_campaign_id=baseline_campaign_id,
triggered_by=triggered_by,
)
finally:
if owns_session:
active_session.close()
if row.status == "queued":
_launch_comparison(
campaign_id,
triggered_by=row.triggered_by or triggered_by,
baseline_campaign_id=row.baseline_campaign_id,
)
def recover_campaign_intelligence_jobs(session: Session) -> tuple[int, int]:
"""Fail interrupted work and relaunch every durably queued job."""
analyses = CampaignAnalysisRepository(session)
comparisons = CampaignPeriodComparisonRepository(session)
interrupted = analyses.mark_orphans_failed() + comparisons.mark_orphans_failed()
queued_analyses = analyses.prepare_queued_recovery(
MAX_QUEUED_RECOVERY_ATTEMPTS,
"服务重启恢复次数超过上限,分析任务已终止",
)
for row in queued_analyses:
_launch_analysis(row.campaign_id, triggered_by=row.triggered_by or "manual")
queued_comparisons = comparisons.prepare_queued_recovery(
MAX_QUEUED_RECOVERY_ATTEMPTS,
"服务重启恢复次数超过上限,周期对比任务已终止",
)
for row in queued_comparisons:
_launch_comparison(
row.campaign_id,
triggered_by=row.triggered_by or "manual",
baseline_campaign_id=row.baseline_campaign_id,
)
return interrupted, len(queued_analyses) + len(queued_comparisons)
def is_intelligence_job_running(kind: str, campaign_id: str) -> bool:
"""Expose process-local liveness without exposing registry internals."""
return _registry.is_running(_job_key(kind, campaign_id))
async def shutdown_campaign_intelligence_jobs() -> None:
"""Stop every live analysis and comparison worker."""
await _registry.shutdown_all()

View File

@ -230,7 +230,7 @@ def _run_window_offset(campaign: Campaign, run: EvalRun) -> float:
def summarize_campaign_progress(campaign: Campaign, runs: list[EvalRun]) -> dict[str, Any]:
"""Compact list-row progress: completed vs *planned* total, plus pass_rate.
Unlike ``campaign_progress`` (live window position), this powers the list
Unlike the detail projection's live window position, this powers the list
view. ``planned_total`` is the sum of plan-entry counts a fixed target the
campaign works toward, so the progress bar fills from 0 rather than tracking
a growing spawned count. ``overall_pass_rate`` reuses ``aggregate_runs`` so
@ -361,30 +361,13 @@ def load_campaign_report(session: Session, campaign: Campaign) -> dict[str, Any]
def load_campaign_view(session: Session, campaign: Campaign) -> dict[str, Any]:
"""活动级读模型单一出口:一次取齐报告 / 探索 / 分析 / 对比。
"""Compatibility entry for the unified Campaign read model."""
from agenteval.evaluation.campaign_read_model import CampaignReadModel
四个数据源总是被一起取markdown 导出分析执行器前端报告抽屉
统一入口消除调用方的拼装逻辑各子模块的取法保持不变此处只做组合
"""
from agenteval.evaluation.comparison import load_comparison_view
from agenteval.exploration.summary import summarize_campaign_exploration
from agenteval.storage.repository import CampaignAnalysisRepository
report = load_campaign_report(session, campaign)
exploration = summarize_campaign_exploration(session, campaign.id)
analysis_row = CampaignAnalysisRepository(session).get_by_campaign(campaign.id)
analysis = analysis_row.get_result() if analysis_row and analysis_row.status == "completed" else None
comparison_view = load_comparison_view(session, campaign)
comparison = comparison_view if comparison_view.get("status") != "none" else None
return {
"report": report,
"exploration": exploration,
"analysis": analysis,
"comparison": comparison,
}
view = CampaignReadModel(session).full_view(campaign.id)
if view is None:
raise ValueError(f"campaign not found: {campaign.id}")
return view
def save_report(run_id: str, fmt: str = "html", output_dir: Optional[Path] = None) -> Path:

View File

@ -239,6 +239,7 @@ class CampaignAnalysisDB(SQLModel, table=True):
model_config_id: Optional[str] = None
error: Optional[str] = None
triggered_by: str = "manual"
recovery_attempts: int = 0
created_at: Optional[datetime] = Field(default_factory=utc_now)
updated_at: Optional[datetime] = Field(default_factory=utc_now)
@ -266,6 +267,7 @@ class CampaignPeriodComparisonDB(SQLModel, table=True):
model_config_id: Optional[str] = None
error: Optional[str] = None
triggered_by: str = "manual"
recovery_attempts: int = 0
created_at: Optional[datetime] = Field(default_factory=utc_now)
updated_at: Optional[datetime] = Field(default_factory=utc_now)

View File

@ -46,6 +46,22 @@ M = TypeVar("M") # domain model
DB = TypeVar("DB") # persisted table row
class AsyncJobClaimStatus(str, Enum):
CLAIMED = "claimed"
NOT_FOUND = "not_found"
ALREADY_CLAIMED = "already_claimed"
NOT_QUEUED = "not_queued"
@dataclass(frozen=True)
class AsyncJobClaimResult:
status: AsyncJobClaimStatus
@property
def claimed(self) -> bool:
return self.status is AsyncJobClaimStatus.CLAIMED
class BaseRepository(Generic[M, DB]):
"""Shared CRUD skeleton for id-keyed entity repositories.
@ -425,6 +441,21 @@ class RunRepository(BaseRepository[EvalRun, EvalRunDB]):
statement = select(EvalRunDB).where(EvalRunDB.campaign_id == campaign_id).order_by(EvalRunDB.started_at)
return [self._from_db(r) for r in self.session.exec(statement).all()]
def list_by_campaigns(self, campaign_ids: list[str]) -> dict[str, list[EvalRun]]:
"""Load child Runs for multiple Campaigns in one query."""
grouped = {campaign_id: [] for campaign_id in campaign_ids}
if not campaign_ids:
return grouped
statement = (
select(EvalRunDB)
.where(EvalRunDB.campaign_id.in_(campaign_ids)) # type: ignore[union-attr]
.order_by(EvalRunDB.started_at)
)
for row in self.session.exec(statement).all():
if row.campaign_id in grouped:
grouped[row.campaign_id].append(self._from_db(row))
return grouped
def mark_orphans_failed(self) -> int:
"""Fail process-orphaned Runs while preserving recoverable child claims.
@ -764,6 +795,54 @@ class AsyncJobRepository(Generic[DB]):
statement = select(self._table).where(self._table.campaign_id == campaign_id) # type: ignore[attr-defined]
return self.session.exec(statement).first()
def list_queued(self) -> list[DB]:
statement = select(self._table).where(self._table.status == "queued") # type: ignore[attr-defined]
return list(self.session.exec(statement).all())
def claim_queued(self, campaign_id: str) -> "AsyncJobClaimResult":
"""Atomically move one queued job to generating.
The status predicate is the durable idempotency authority. Competing
workers may observe the same queued row, but only one can claim it.
"""
statement = (
sql_update(self._table)
.where(
self._table.campaign_id == campaign_id, # type: ignore[attr-defined]
self._table.status == "queued", # type: ignore[attr-defined]
)
.values(status="generating", error=None, updated_at=utc_now())
)
result = self.session.exec(statement)
self.session.commit()
self.session.expire_all()
if result.rowcount == 1:
return AsyncJobClaimResult(AsyncJobClaimStatus.CLAIMED)
row = self.get_by_campaign(campaign_id)
if row is None:
return AsyncJobClaimResult(AsyncJobClaimStatus.NOT_FOUND)
if row.status == "generating": # type: ignore[attr-defined]
return AsyncJobClaimResult(AsyncJobClaimStatus.ALREADY_CLAIMED)
return AsyncJobClaimResult(AsyncJobClaimStatus.NOT_QUEUED)
def prepare_queued_recovery(self, max_attempts: int, exhausted_error: str) -> list[DB]:
"""Increment queued recovery attempts and fail exhausted jobs."""
recoverable: list[DB] = []
rows = self.list_queued()
for row in rows:
if row.recovery_attempts >= max_attempts: # type: ignore[attr-defined]
row.status = "failed" # type: ignore[attr-defined]
row.error = exhausted_error # type: ignore[attr-defined]
else:
row.recovery_attempts += 1 # type: ignore[attr-defined]
recoverable.append(row)
row.updated_at = utc_now() # type: ignore[attr-defined]
self.session.add(row)
if rows:
self.session.commit()
self.session.expire_all()
return recoverable
def mark_orphans_failed(self, error_message: str) -> int:
"""服务启动时清理:把滞留的 generating 行标记为 failed。
@ -792,20 +871,22 @@ class CampaignAnalysisRepository(AsyncJobRepository[CampaignAnalysisDB]):
"""Persist an analysis job before launching its process-local task."""
row = self.get_by_campaign(campaign_id)
if row is None:
row = CampaignAnalysisDB(campaign_id=campaign_id)
row.status = "queued"
row.result = None
row.error = None
row.triggered_by = triggered_by
row.updated_at = utc_now()
row = CampaignAnalysisDB(campaign_id=campaign_id, status="queued", triggered_by=triggered_by)
elif row.status in {"queued", "generating"}:
return row
else:
row.status = "queued"
row.result = None
row.model_config_id = None
row.error = None
row.triggered_by = triggered_by
row.recovery_attempts = 0
row.updated_at = utc_now()
self.session.add(row)
self.session.commit()
self.session.refresh(row)
return row
def list_queued(self) -> list[CampaignAnalysisDB]:
return list(self.session.exec(select(CampaignAnalysisDB).where(CampaignAnalysisDB.status == "queued")).all())
def upsert(
self,
campaign_id: str,
@ -843,6 +924,38 @@ class CampaignPeriodComparisonRepository(AsyncJobRepository[CampaignPeriodCompar
_table = CampaignPeriodComparisonDB
def enqueue(
self,
campaign_id: str,
*,
baseline_campaign_id: str,
triggered_by: str = "manual",
) -> CampaignPeriodComparisonDB:
"""Persist a comparison job before launching its process-local task."""
row = self.get_by_campaign(campaign_id)
if row is None:
row = CampaignPeriodComparisonDB(
campaign_id=campaign_id,
baseline_campaign_id=baseline_campaign_id,
status="queued",
triggered_by=triggered_by,
)
elif row.status in {"queued", "generating"}:
return row
else:
row.baseline_campaign_id = baseline_campaign_id
row.status = "queued"
row.result = None
row.model_config_id = None
row.error = None
row.triggered_by = triggered_by
row.recovery_attempts = 0
row.updated_at = utc_now()
self.session.add(row)
self.session.commit()
self.session.refresh(row)
return row
def upsert(
self,
campaign_id: str,

View File

@ -32,28 +32,32 @@ from agenteval.web.websocket import ws_manager
@asynccontextmanager
async def lifespan(_: FastAPI):
init_db()
# 评测任务是进程内 asyncio 任务,重启后不会恢复——清理僵尸运行(尽力而为,不阻断启动)
# 恢复耐久 Campaign/智能作业,并清理无法安全重放的中断运行(尽力而为,不阻断启动)。
try:
session = get_session()
try:
from agenteval.evaluation.campaign_lifecycle import recover_campaign_runtime
from agenteval.evaluation.campaign_runner import campaign_runtime
from agenteval.evaluation.intelligence_jobs import recover_campaign_intelligence_jobs
recovery = recover_campaign_runtime(session)
interrupted_jobs, resumed_jobs = recover_campaign_intelligence_jobs(session)
recovery = campaign_runtime.recover()
if recovery.interrupted_runs:
logging.getLogger("agenteval").warning(
"启动清理:%d 个中断的运行已标记为 failed", recovery.interrupted_runs
)
if recovery.interrupted_analysis:
if interrupted_jobs:
logging.getLogger("agenteval").warning(
"启动清理:%d 条中断的分析/对比已标记为 failed", recovery.interrupted_analysis
"启动清理:%d 条中断的分析/对比已标记为 failed",
interrupted_jobs,
)
if recovery.resumed_campaigns:
logging.getLogger("agenteval").warning(
"启动恢复:%d 个进行中的评估活动已续跑", recovery.resumed_campaigns
)
if recovery.resumed_analysis:
if resumed_jobs:
logging.getLogger("agenteval").warning(
"启动恢复:%d 条排队中的活动分析已续跑", recovery.resumed_analysis
"启动恢复:%d 条排队中的活动分析/对比已续跑",
resumed_jobs,
)
finally:
session.close()
@ -61,18 +65,16 @@ async def lifespan(_: FastAPI):
logging.getLogger("agenteval").warning("启动清理失败(忽略): %s", exc)
yield
# 优雅停止所有进程内任务:先停活动调度循环,再停在跑的评测运行,
# 最后停三条 LLM 任务链(分析 / 周期对比 / judge 复核)
# 最后停活动智能作业(分析 / 周期对比)和 judge 复核
try:
from agenteval.evaluation.analysis import analysis_registry
from agenteval.evaluation.campaign_runner import shutdown_all
from agenteval.evaluation.comparison import comparison_registry
from agenteval.evaluation.campaign_runner import campaign_runtime
from agenteval.evaluation.intelligence_jobs import shutdown_campaign_intelligence_jobs
from agenteval.exploration.judge import judge_registry
from agenteval.web.routers.runs import run_registry
await shutdown_all()
await campaign_runtime.shutdown()
await run_registry.shutdown_all()
await analysis_registry.shutdown_all()
await comparison_registry.shutdown_all()
await shutdown_campaign_intelligence_jobs()
await judge_registry.shutdown_all()
except Exception as exc:
logging.getLogger("agenteval").warning("活动调度停止失败(忽略): %s", exc)
@ -104,7 +106,9 @@ app.include_router(scenarios.router, prefix="/api/scenarios", tags=["scenarios"]
app.include_router(runs.router, prefix="/api/runs", tags=["runs"], dependencies=_api_deps)
app.include_router(campaigns.router, prefix="/api/campaigns", tags=["campaigns"], dependencies=_api_deps)
app.include_router(exploration.router, prefix="/api/exploration", tags=["exploration"], dependencies=_api_deps)
app.include_router(intelligent_evals.router, prefix="/api/intelligent-evals", tags=["intelligent-evals"], dependencies=_api_deps)
app.include_router(
intelligent_evals.router, prefix="/api/intelligent-evals", tags=["intelligent-evals"], dependencies=_api_deps
)
app.include_router(reports.router, prefix="/api/reports", tags=["reports"], dependencies=_api_deps)
app.include_router(stats.router, prefix="/api/stats", tags=["stats"], dependencies=_api_deps)
app.include_router(files.router, prefix="/api/files", tags=["files"], dependencies=_api_deps)

View File

@ -1,6 +1,6 @@
"""API routes for evaluation campaigns (评估活动).
Creating a campaign starts its durable scheduler loop (``campaign_runner``),
Creating a campaign starts its durable ``CampaignRuntime`` loop,
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
@ -11,34 +11,17 @@ from fastapi import APIRouter, Body, Depends, HTTPException, Response
from pydantic import BaseModel, Field
from sqlmodel import Session
from agenteval.evaluation.analysis import enqueue_campaign_analysis, resolve_analysis_model
from agenteval.evaluation.analysis import resolve_analysis_model
from agenteval.evaluation.campaign_lifecycle import CampaignCreateError, CampaignLifecycleError
from agenteval.evaluation.campaign_lifecycle import cancel_campaign as cancel_campaign_lifecycle
from agenteval.evaluation.campaign_lifecycle import create_campaign as create_campaign_lifecycle
from agenteval.evaluation.campaign_runner import campaign_progress, request_cancel, start_campaign
from agenteval.evaluation.comparison import (
ComparisonError,
load_comparison_view,
start_campaign_comparison,
validate_comparison_request,
)
from agenteval.evaluation.report import (
build_campaign_timeline,
load_campaign_report,
load_campaign_view,
summarize_campaign_progress,
)
from agenteval.evaluation.campaign_read_model import CampaignReadModel
from agenteval.evaluation.campaign_runner import campaign_runtime
from agenteval.evaluation.comparison import ComparisonError, validate_comparison_request
from agenteval.evaluation.intelligence_jobs import enqueue_campaign_analysis, enqueue_campaign_comparison
from agenteval.evaluation.report_render import render_campaign_markdown
from agenteval.exploration.summary import summarize_campaign_exploration
from agenteval.models import CampaignPlanEntry, CampaignStatus, ExplorationBudgetConfig, ExplorationSeeds
from agenteval.storage.db import iso_utc
from agenteval.storage.repository import (
CampaignAnalysisRepository,
CampaignRepository,
RunRepository,
ScenarioRepository,
TargetRepository,
)
from agenteval.storage.repository import CampaignRepository
from agenteval.web.deps import get_db
router = APIRouter()
@ -57,15 +40,7 @@ class CreateCampaignRequest(BaseModel):
@router.get("")
async def list_campaigns(session: Session = Depends(get_db)) -> list[dict]:
repo = CampaignRepository(session)
run_repo = RunRepository(session)
result = []
for campaign in repo.list_all():
data = campaign.model_dump()
runs = run_repo.list_by_campaign(campaign.id)
data["progress"] = summarize_campaign_progress(campaign, runs)
result.append(data)
return result
return CampaignReadModel(session).list_items()
@router.post("")
@ -84,7 +59,7 @@ async def create_campaign(
analysis_model_config_id=request.analysis_model_config_id,
exploration_seeds=request.exploration_seeds,
exploration_budget=request.exploration_budget,
launch=start_campaign,
launch=campaign_runtime.start,
)
except CampaignCreateError as exc:
raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc
@ -94,7 +69,7 @@ async def create_campaign(
@router.post("/{campaign_id}/cancel")
async def cancel_campaign(campaign_id: str, session: Session = Depends(get_db)) -> dict:
try:
campaign = cancel_campaign_lifecycle(session, campaign_id, stop=request_cancel)
campaign = cancel_campaign_lifecycle(session, campaign_id, stop=campaign_runtime.cancel)
except CampaignLifecycleError as exc:
raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc
return campaign.model_dump()
@ -102,47 +77,25 @@ async def cancel_campaign(campaign_id: str, session: Session = Depends(get_db))
@router.get("/{campaign_id}/report")
async def get_campaign_report(campaign_id: str, session: Session = Depends(get_db)) -> dict:
campaign = CampaignRepository(session).get(campaign_id)
if not campaign:
report = CampaignReadModel(session).report(campaign_id)
if report is None:
raise HTTPException(status_code=404, detail="campaign not found")
report = load_campaign_report(session, campaign)
exploration = summarize_campaign_exploration(session, campaign_id)
if exploration is not None:
report["exploration"] = exploration
return report
@router.get("/{campaign_id}/report/markdown")
async def get_campaign_report_markdown(campaign_id: str, session: Session = Depends(get_db)) -> Response:
campaign = CampaignRepository(session).get(campaign_id)
if not campaign:
projection = CampaignReadModel(session).markdown_projection(campaign_id)
if projection is None:
raise HTTPException(status_code=404, detail="campaign not found")
view = load_campaign_view(session, campaign)
scenario_names = ScenarioRepository(session).name_map()
target = TargetRepository(session).get(campaign.target_id)
target_name = target.name if target else None
comparison = None
if view["comparison"] and view["comparison"].get("status") == "completed":
cmp_view = view["comparison"]
cmp = cmp_view.get("comparison") or {}
baseline = cmp.get("baseline") or {}
comparison = {
"result": cmp.get("result"),
"baseline_name": baseline.get("name"),
"baseline_completed_at": baseline.get("completed_at"),
"model_name": cmp.get("model_name"),
"updated_at": cmp.get("updated_at"),
"metric_diff": cmp_view.get("metric_diff"),
}
md = render_campaign_markdown(
view["report"],
analysis=view["analysis"],
comparison=comparison,
exploration=view["exploration"],
target_name=target_name,
scenario_names=scenario_names,
projection["report"],
analysis=projection["analysis"],
comparison=projection["comparison"],
exploration=projection["exploration"],
target_name=projection["target_name"],
scenario_names=projection["scenario_names"],
)
return Response(
content=md,
@ -153,29 +106,18 @@ async def get_campaign_report_markdown(campaign_id: str, session: Session = Depe
@router.get("/{campaign_id}/timeline")
async def get_campaign_timeline(campaign_id: str, session: Session = Depends(get_db)) -> dict:
campaign = CampaignRepository(session).get(campaign_id)
if not campaign:
timeline = CampaignReadModel(session).timeline(campaign_id)
if timeline is None:
raise HTTPException(status_code=404, detail="campaign not found")
runs = RunRepository(session).list_by_campaign(campaign_id)
scenario_names = ScenarioRepository(session).name_map()
return {"entries": build_campaign_timeline(campaign, runs, scenario_names=scenario_names)}
return timeline
@router.get("/{campaign_id}/analysis")
async def get_campaign_analysis(campaign_id: str, session: Session = Depends(get_db)) -> dict:
if not CampaignRepository(session).get(campaign_id):
analysis = CampaignReadModel(session).analysis(campaign_id)
if analysis is None:
raise HTTPException(status_code=404, detail="campaign not found")
row = CampaignAnalysisRepository(session).get_by_campaign(campaign_id)
if row is None:
return {"status": "none"}
return {
"status": row.status,
"result": row.get_result(),
"error": row.error,
"model_config_id": row.model_config_id,
"triggered_by": row.triggered_by,
"updated_at": iso_utc(row.updated_at),
}
return analysis
@router.post("/{campaign_id}/analysis")
@ -201,11 +143,10 @@ async def get_campaign_comparison(campaign_id: str, session: Session = Depends(g
``metric_diff`` 对应生效基线已有对比行时为该行记录的基线
否则为自动基线两者都没有时为 null
"""
campaign = CampaignRepository(session).get(campaign_id)
if not campaign:
comparison = CampaignReadModel(session).comparison(campaign_id)
if comparison is None:
raise HTTPException(status_code=404, detail="campaign not found")
return load_comparison_view(session, campaign)
return comparison
class GenerateComparisonRequest(BaseModel):
@ -228,17 +169,18 @@ async def trigger_campaign_comparison(
except ComparisonError as exc:
raise HTTPException(status_code=400, detail=str(exc))
start_campaign_comparison(campaign_id, triggered_by="manual", baseline_campaign_id=baseline.id)
enqueue_campaign_comparison(
campaign_id,
triggered_by="manual",
baseline_campaign_id=baseline.id,
session=session,
)
return {"status": "generating"}
@router.get("/{campaign_id}")
async def get_campaign(campaign_id: str, session: Session = Depends(get_db)) -> dict:
campaign = CampaignRepository(session).get(campaign_id)
if not campaign:
campaign = CampaignReadModel(session).detail(campaign_id)
if campaign is None:
raise HTTPException(status_code=404, detail="campaign not found")
runs = RunRepository(session).list_by_campaign(campaign_id)
data = campaign.model_dump()
data["progress"] = campaign_progress(campaign, runs)
return data
return campaign

View File

@ -51,6 +51,11 @@
- **Claim 先于执行**:调度器对每个 occurrence 先执行条件 claim再进入 `EvalEngine`。重复 tick、并发调度或重启不会创建第二个 Run取消后的 Campaign 不允许新 claim已启动 Run 可继续完成。
- **恢复边界**:持久化为 `pending` 且仍属于 running Campaign 的子 Run 可以恢复;进程中断遗留的 `running` 子 Run 统一标记为 `failed/interrupted`,禁止重放可能已经发送的外部消息。
- **生命周期 CAS**创建、启动、取消和完成均通过生命周期模块与条件更新完成。取消竞态优先于完成Campaign 终态与 running 探索会话结算在同一事务提交,任一步失败均保持活动与探索会话为 `running`,由后续调度 tick 重试。
- **分析任务耐久化**:活动分析先写入 `queued` 再启动进程内 worker重启恢复 queued 任务,遗留 `generating` 任务标记为中断失败。分析失败或重复执行不改变 Campaign 完成状态。
- **活动智能作业耐久化**:智能分析与周期对比统一由活动智能作业 module 管理。两类作业都先写入 `queued` 再启动进程内 worker并以 `queued → generating` 条件认领防止同一作业被重复执行。
- **作业恢复边界**:重启恢复 queued 智能分析与周期对比;遗留 `generating` 作业标记为中断失败,不自动重放可能已经产生费用的模型调用。失败或重复执行不改变 Campaign 完成状态。
- **进程句柄收口**:智能分析与周期对比共享一个进程内 TaskRegistry ownerWeb 生命周期只恢复和关闭该 owner不分别了解两条作业链的 registry。
- **Campaign runtime seam**Campaign 耐久执行统一由 `CampaignRuntime` module 提供 `start`、`cancel`、`recover`、`shutdown` 四个生命周期级操作。HTTP 与 Web 生命周期不再调用 tick、advance、reconcile、loop 或 TaskRegistry 阶段函数。
- **内部 adapter**:真实 UTC 时钟与 `EvalEngine` child-Run 执行器是 runtime implementation 的默认 adapter测试可注入可控时钟和本地执行器但这些 internal seam 不进入生产调用方的 interface。
- **恢复与取消**`recover` 自行清理中断 Run、枚举 running Campaign 并重建循环;每个循环内部先 reconcile child Run再 tick。`cancel` 只处理进程句柄Campaign 终态仍先由生命周期 CAS 持久化。
启动恢复顺序固定为:清理中断 Run 与 LLM 任务 → 重建 running Campaign 调度循环(其中包含 child Run reconciliation→ 重启 queued 分析任务。该顺序保证恢复动作只依据已提交的数据库事实,不依赖上一次进程的内存状态。
启动恢复顺序固定为:结算遗留 generating 智能作业并重启 queued 智能作业 → `CampaignRuntime.recover` 清理中断 Run 并重建 running Campaign 调度循环(每个循环先完成 child Run reconciliation。该顺序保证恢复动作只依据已提交的数据库事实,不依赖上一次进程的内存状态。

View File

@ -0,0 +1,35 @@
# 拒绝跨领域通用条件写入 Module
状态已采纳2026-08-11
## 背景
Campaign 与 Intelligent Evaluation 的 Repository 都使用条件更新,并都区分 `applied`、`not_found`、`conflict`。因此曾考虑抽取跨领域的 compare-and-set module以复用 SQL 模板和结果分类。
## 删除测试
假想 interface 至少需要调用方提供数据库表与主键列、状态列、一个或多个期望状态、目标字段映射、时间戳策略、JSON 序列化、更新后模型映射,以及可选的同事务副作用。
两个领域真正需要隐藏的知识并不相同:
| 维度 | Campaign | Intelligent Evaluation |
|---|---|---|
| 状态条件 | 可接受多个来源状态 | 单一期望状态与完整转换表 |
| 时间语义 | 调用方传入时间;按目标状态写开始或完成时间 | Repository 取当前时间;始终更新 `updated_at`,首次执行用 `coalesce` |
| 业务字段 | 主要写状态 | 同时写计划、反馈或报告,并负责 JSON 序列化 |
| 事务副作用 | 完成/取消时同步结算 Exploration Session | 无跨聚合结算 |
| 冲突结果 | 返回当前 Campaign 快照 | 返回冲突分类,由 lifecycle 重新表达领域错误 |
删除该假想 module 后,只会在两个 Repository 中恢复少量 SQL 更新与结果分类;状态规则、字段映射和事务副作用仍必须留在各自领域。它不能减少调用方需要掌握的知识,反而会把稳定的领域差异暴露为通用参数和回调,形成浅 interface。
## 决策
不创建跨 Campaign 与 Intelligent Evaluation 的通用条件写入 module。两侧保留领域命名的结果类型和 Repository 私有 helper生命周期 module 继续作为生产调用方的写入 interface。
只有出现第三个真实 adapter且至少三个调用面共享相同的状态条件、时间策略、字段语义、事务副作用和冲突映射时才重新评估该 seam。单纯重复 SQL 形状不足以建立新 module。
## 影响
- 保持领域事务知识的 localityCampaign 结算变化不会影响 Intelligent Evaluation。
- 接受少量 SQL 模板重复,避免通用字典、回调和类型擦除。
- 测试继续穿过各自 lifecycle interface断言持久状态和领域副作用而不是测试通用 SQL helper。

View File

@ -1,6 +1,6 @@
# volcengine-102 正式线部署指南
**状态**已部署并通过正式域名验收最后验证2026-08-09)。
**状态**已部署并通过正式域名验收最后验证2026-08-11)。
**Compose**`deploy/volcengine-102/docker-compose.yml`
**入口脚本**`scripts/deploy-volcengine-102.sh`

View File

@ -471,7 +471,7 @@ export interface CampaignTimelineEntry {
started_at: string | null
}
export type CampaignAnalysisStatus = 'none' | 'generating' | 'completed' | 'failed'
export type CampaignAnalysisStatus = 'none' | 'queued' | 'generating' | 'completed' | 'failed'
export interface CampaignAnalysisProblem {
severity: string

View File

@ -0,0 +1,153 @@
import { act, renderHook, waitFor } from '@testing-library/react'
import { afterEach, describe, expect, it, vi } from 'vitest'
import type { CampaignListItem } from '../api'
import {
type CampaignReadAdapter,
type CampaignReportSnapshot,
useCampaignReport,
} from './useCampaignReport'
const campaign = { id: 'campaign-1', status: 'running' } as CampaignListItem
const terminalCampaign = { ...campaign, status: 'completed' } as CampaignListItem
const snapshot = {
report: { campaign_id: 'campaign-1', status: 'running' },
runs: [{ id: 'run-1', campaign_id: 'campaign-1' }],
timeline: [{ run_id: 'run-1' }],
analysis: { status: 'none' },
comparison: { status: 'none' },
} as CampaignReportSnapshot
function deferred<T>() {
let resolve!: (value: T) => void
const promise = new Promise<T>((resolvePromise) => { resolve = resolvePromise })
return { promise, resolve }
}
function createAdapter(overrides: Partial<CampaignReadAdapter> = {}): CampaignReadAdapter {
return {
list: vi.fn().mockResolvedValue([]),
report: vi.fn().mockResolvedValue(snapshot),
timeline: vi.fn().mockResolvedValue([]),
generateAnalysis: vi.fn().mockResolvedValue(undefined),
generateComparison: vi.fn().mockResolvedValue(undefined),
...overrides,
}
}
afterEach(() => {
vi.useRealTimers()
})
describe('useCampaignReport', () => {
it('loads complete list and report snapshots', async () => {
const adapter = createAdapter({
list: vi.fn().mockResolvedValue([campaign]),
report: vi.fn().mockResolvedValue(snapshot),
})
const { result } = renderHook(() => useCampaignReport('campaign-1', true, adapter))
await waitFor(() => expect(result.current.report.value).toBe(snapshot))
expect(result.current.list.value).toEqual([campaign])
expect(result.current.list.phase).toBe('ready')
expect(result.current.report.phase).toBe('ready')
})
it('retains the last complete snapshot when a silent refresh fails', async () => {
const adapter = createAdapter({
list: vi.fn().mockResolvedValue([campaign]),
report: vi.fn()
.mockResolvedValueOnce(snapshot)
.mockRejectedValueOnce(new Error('network error')),
})
const { result } = renderHook(() => useCampaignReport('campaign-1', true, adapter))
await waitFor(() => expect(result.current.report.value).toBe(snapshot))
await act(async () => { await result.current.reloadReport() })
expect(result.current.report.phase).toBe('ready')
expect(result.current.report.value).toBe(snapshot)
expect(result.current.report.error).toBeNull()
})
it('ignores a late report response for the previous campaign', async () => {
const first = deferred<CampaignReportSnapshot>()
const second = deferred<CampaignReportSnapshot>()
const secondSnapshot = {
...snapshot,
report: { ...snapshot.report, campaign_id: 'campaign-2' },
} as CampaignReportSnapshot
const adapter = createAdapter({
list: vi.fn().mockResolvedValue([]),
report: vi.fn((id: string) => (id === 'campaign-1' ? first.promise : second.promise)),
})
const { result, rerender } = renderHook(
({ selectedId }) => useCampaignReport(selectedId, true, adapter),
{ initialProps: { selectedId: 'campaign-1' as string | null } },
)
rerender({ selectedId: 'campaign-2' })
expect(result.current.report.value).toBeNull()
await act(async () => { second.resolve(secondSnapshot) })
expect(result.current.report.value?.report.campaign_id).toBe('campaign-2')
await act(async () => { first.resolve(snapshot) })
expect(result.current.report.value?.report.campaign_id).toBe('campaign-2')
})
it('polls active lists and stops after a terminal snapshot', async () => {
vi.useFakeTimers()
const adapter = createAdapter({
list: vi.fn()
.mockResolvedValueOnce([campaign])
.mockResolvedValue([terminalCampaign]),
report: vi.fn().mockResolvedValue({
...snapshot,
report: { ...snapshot.report, status: 'completed' },
}),
})
renderHook(() => useCampaignReport(null, true, adapter))
await act(async () => { await Promise.resolve() })
await act(async () => { await vi.advanceTimersByTimeAsync(5000) })
await act(async () => { await vi.advanceTimersByTimeAsync(10000) })
expect(adapter.list).toHaveBeenCalledTimes(2)
})
it('refreshes the complete snapshot after analysis and comparison mutations', async () => {
const refreshed = {
...snapshot,
analysis: { status: 'queued' },
comparison: { ...snapshot.comparison, status: 'queued' },
} as CampaignReportSnapshot
const adapter = createAdapter({
list: vi.fn().mockResolvedValue([campaign]),
report: vi.fn().mockResolvedValueOnce(snapshot).mockResolvedValue(refreshed),
})
const { result } = renderHook(() => useCampaignReport('campaign-1', true, adapter))
await waitFor(() => expect(result.current.report.value).toBe(snapshot))
await act(async () => { await result.current.generateAnalysis() })
expect(adapter.generateAnalysis).toHaveBeenCalledWith('campaign-1')
expect(result.current.report.value).toBe(refreshed)
await act(async () => { await result.current.generateComparison('campaign-0') })
expect(adapter.generateComparison).toHaveBeenCalledWith('campaign-1', 'campaign-0')
expect(adapter.report).toHaveBeenCalledTimes(3)
})
it('owns expanded timeline loading', async () => {
const entries = [{ run_id: 'run-2' }] as CampaignReportSnapshot['timeline']
const adapter = createAdapter({ timeline: vi.fn().mockResolvedValue(entries) })
const { result } = renderHook(() => useCampaignReport(null, true, adapter))
act(() => { result.current.setTimelineExpanded('campaign-1', true) })
await waitFor(() => expect(result.current.timelines['campaign-1']).toBe(entries))
expect(result.current.expandedIds).toEqual(['campaign-1'])
act(() => { result.current.setTimelineExpanded('campaign-1', false) })
expect(result.current.expandedIds).toEqual([])
})
})

View File

@ -1,83 +1,278 @@
import { useState, useCallback, useEffect, type Dispatch, type SetStateAction } from 'react'
import { campaignsApi, runsApi, type CampaignReport, type Run, type CampaignTimelineEntry, type CampaignAnalysis, type CampaignComparison } from '../api'
import { useCallback, useEffect, useLayoutEffect, useReducer, useRef, useState } from 'react'
import {
campaignsApi,
runsApi,
type CampaignAnalysis,
type CampaignAnalysisStatus,
type CampaignComparison,
type CampaignListItem,
type CampaignReport,
type CampaignTimelineEntry,
type Run,
} from '../api'
import { usePolling } from './usePolling'
interface CampaignReportData {
report: CampaignReport | null
type ReadPhase = 'idle' | 'loading' | 'refreshing' | 'ready' | 'error'
interface ReadSlot<T> {
phase: ReadPhase
value: T
error: string | null
}
export interface CampaignReportSnapshot {
report: CampaignReport
runs: Run[]
timeline: CampaignTimelineEntry[]
analysis: CampaignAnalysis | null
comparison: CampaignComparison | null
loading: boolean
analysis: CampaignAnalysis
comparison: CampaignComparison
}
interface UseCampaignReportResult extends CampaignReportData {
refetch: (silent?: boolean) => Promise<void>
setAnalysis: Dispatch<SetStateAction<CampaignAnalysis | null>>
setComparison: Dispatch<SetStateAction<CampaignComparison | null>>
interface CampaignReadState {
list: ReadSlot<CampaignListItem[]>
report: ReadSlot<CampaignReportSnapshot | null> & {
selectedId: string | null
requestId: number
}
}
export function useCampaignReport(campaignId: string | null): UseCampaignReportResult {
const [data, setData] = useState<CampaignReportData>({
report: null,
runs: [],
timeline: [],
analysis: null,
comparison: null,
loading: false,
})
type CampaignReadAction =
| { type: 'list_requested'; silent?: boolean }
| { type: 'list_succeeded'; value: CampaignListItem[] }
| { type: 'list_failed'; error: string }
| { type: 'report_cleared'; requestId: number }
| { type: 'report_requested'; id: string; requestId: number; silent?: boolean }
| { type: 'report_succeeded'; id: string; requestId: number; value: CampaignReportSnapshot }
| { type: 'report_failed'; id: string; requestId: number; error: string }
const refetch = useCallback(async (silent = false) => {
if (!campaignId) return
if (!silent) {
setData({
report: null,
runs: [],
timeline: [],
analysis: null,
comparison: null,
loading: true,
})
export interface CampaignReadAdapter {
list: () => Promise<CampaignListItem[]>
report: (campaignId: string) => Promise<CampaignReportSnapshot>
timeline: (campaignId: string) => Promise<CampaignTimelineEntry[]>
generateAnalysis: (campaignId: string) => Promise<void>
generateComparison: (campaignId: string, baselineCampaignId?: string) => Promise<void>
}
export const campaignReadAdapter: CampaignReadAdapter = {
list: () => campaignsApi.list().then((response) => response.data),
report: async (campaignId) => {
const [report, runs, timeline, analysis, comparison] = await Promise.all([
campaignsApi.report(campaignId),
runsApi.list(),
campaignsApi.timeline(campaignId),
campaignsApi.getAnalysis(campaignId),
campaignsApi.getComparison(campaignId),
])
return {
report: report.data,
runs: runs.data.filter((run) => run.campaign_id === campaignId),
timeline: timeline.data.entries,
analysis: analysis.data,
comparison: comparison.data,
}
},
timeline: (campaignId) => campaignsApi.timeline(campaignId).then((response) => response.data.entries),
generateAnalysis: (campaignId) => campaignsApi.generateAnalysis(campaignId).then(() => undefined),
generateComparison: (campaignId, baselineCampaignId) => (
campaignsApi.generateComparison(campaignId, baselineCampaignId).then(() => undefined)
),
}
const initialState: CampaignReadState = {
list: { phase: 'idle', value: [], error: null },
report: {
phase: 'idle', value: null, error: null, selectedId: null, requestId: 0,
},
}
function requestSlot<T>(slot: ReadSlot<T>, silent: boolean | undefined): ReadSlot<T> {
return {
...slot,
phase: silent && slot.phase === 'ready' ? 'refreshing' : 'loading',
error: null,
}
}
function reducer(state: CampaignReadState, action: CampaignReadAction): CampaignReadState {
switch (action.type) {
case 'list_requested':
return { ...state, list: requestSlot(state.list, action.silent) }
case 'list_succeeded':
return { ...state, list: { phase: 'ready', value: action.value, error: null } }
case 'list_failed':
return {
...state,
list: state.list.value.length > 0
? { ...state.list, phase: 'ready', error: null }
: { ...state.list, phase: 'error', error: action.error },
}
case 'report_cleared':
return {
...state,
report: {
phase: 'idle', value: null, error: null, selectedId: null, requestId: action.requestId,
},
}
case 'report_requested': {
const sameSelection = state.report.selectedId === action.id
const current = sameSelection
? state.report
: { ...state.report, value: null, selectedId: action.id }
return {
...state,
report: {
...requestSlot(current, action.silent),
selectedId: action.id,
requestId: action.requestId,
},
}
}
case 'report_succeeded':
if (state.report.selectedId !== action.id || state.report.requestId !== action.requestId) return state
return {
...state,
report: {
phase: 'ready', value: action.value, error: null,
selectedId: action.id, requestId: action.requestId,
},
}
case 'report_failed':
if (state.report.selectedId !== action.id || state.report.requestId !== action.requestId) return state
return {
...state,
report: state.report.value
? { ...state.report, phase: 'ready', error: null }
: { ...state.report, phase: 'error', error: action.error },
}
}
}
function errorMessage(error: unknown): string {
return error instanceof Error ? error.message : '读取评估活动失败'
}
const isCampaignActive = (status: CampaignListItem['status'] | undefined) => (
status === 'planned' || status === 'running'
)
const isIntelligenceJobActive = (status: CampaignAnalysisStatus | undefined) => (
status === 'queued' || status === 'generating'
)
export function useCampaignReport(
selectedId: string | null,
enabled = true,
adapter: CampaignReadAdapter = campaignReadAdapter,
) {
const [state, dispatch] = useReducer(reducer, initialState)
const [expandedIds, setExpandedIds] = useState<string[]>([])
const [timelines, setTimelines] = useState<Record<string, CampaignTimelineEntry[]>>({})
const [analysisBusy, setAnalysisBusy] = useState(false)
const [comparisonBusy, setComparisonBusy] = useState(false)
const reportRequestId = useRef(0)
const selectedIdRef = useRef(selectedId)
selectedIdRef.current = selectedId
const loadList = useCallback(async (silent = false) => {
dispatch({ type: 'list_requested', silent })
try {
const [rep, runs, tl, ana, cmp] = await Promise.all([
campaignsApi.report(campaignId),
runsApi.list(),
campaignsApi.timeline(campaignId),
campaignsApi.getAnalysis(campaignId),
campaignsApi.getComparison(campaignId),
])
setData({
report: rep.data,
runs: runs.data.filter((r) => r.campaign_id === campaignId),
timeline: tl.data.entries,
analysis: ana.data,
comparison: cmp.data,
loading: false,
})
} catch {
setData((prev) => ({ ...prev, loading: false }))
dispatch({ type: 'list_succeeded', value: await adapter.list() })
} catch (error) {
dispatch({ type: 'list_failed', error: errorMessage(error) })
}
}, [campaignId])
}, [adapter])
useEffect(() => {
if (campaignId) {
void refetch(false)
const loadReport = useCallback(async (campaignId: string, silent = false) => {
const requestId = ++reportRequestId.current
dispatch({ type: 'report_requested', id: campaignId, requestId, silent })
try {
const value = await adapter.report(campaignId)
dispatch({ type: 'report_succeeded', id: campaignId, requestId, value })
} catch (error) {
dispatch({ type: 'report_failed', id: campaignId, requestId, error: errorMessage(error) })
}
}, [campaignId, refetch])
}, [adapter])
const setAnalysis: Dispatch<SetStateAction<CampaignAnalysis | null>> = (value) => {
setData((prev) => ({
...prev,
analysis: typeof value === 'function' ? value(prev.analysis) : value,
}))
const loadTimeline = useCallback(async (campaignId: string) => {
const timeline = await adapter.timeline(campaignId)
setTimelines((current) => ({ ...current, [campaignId]: timeline }))
}, [adapter])
const setTimelineExpanded = useCallback((campaignId: string, expanded: boolean) => {
setExpandedIds((current) => (
expanded
? current.includes(campaignId) ? current : [...current, campaignId]
: current.filter((id) => id !== campaignId)
))
if (expanded) void loadTimeline(campaignId)
}, [loadTimeline])
const generateAnalysis = useCallback(async () => {
const campaignId = selectedIdRef.current
if (!campaignId) return
setAnalysisBusy(true)
try {
await adapter.generateAnalysis(campaignId)
if (selectedIdRef.current === campaignId) await loadReport(campaignId, true)
} finally {
setAnalysisBusy(false)
}
}, [adapter, loadReport])
const generateComparison = useCallback(async (baselineCampaignId?: string) => {
const campaignId = selectedIdRef.current
if (!campaignId) return
setComparisonBusy(true)
try {
await adapter.generateComparison(campaignId, baselineCampaignId)
if (selectedIdRef.current === campaignId) await loadReport(campaignId, true)
} finally {
setComparisonBusy(false)
}
}, [adapter, loadReport])
useEffect(() => { void loadList() }, [loadList])
useLayoutEffect(() => {
if (selectedId == null) {
dispatch({ type: 'report_cleared', requestId: ++reportRequestId.current })
return
}
void loadReport(selectedId)
}, [loadReport, selectedId])
const listActive = state.list.value.some((campaign) => isCampaignActive(campaign.status))
const snapshot = state.report.value
const reportActive = isCampaignActive(snapshot?.report.status)
|| isIntelligenceJobActive(snapshot?.analysis.status)
|| isIntelligenceJobActive(snapshot?.comparison.status)
const activeExpandedIds = expandedIds.filter(
(id) => state.list.value.some((campaign) => campaign.id === id && isCampaignActive(campaign.status)),
)
usePolling(() => { void loadList(true) }, 5000, enabled && listActive)
usePolling(
() => { if (selectedId) void loadReport(selectedId, true) },
5000,
enabled && selectedId != null && reportActive,
)
usePolling(
() => { activeExpandedIds.forEach((id) => { void loadTimeline(id) }) },
5000,
enabled && activeExpandedIds.length > 0,
)
return {
list: state.list,
report: state.report,
reloadList: () => loadList(true),
reloadReport: () => (selectedId ? loadReport(selectedId, true) : Promise.resolve()),
expandedIds,
timelines,
setTimelineExpanded,
generateAnalysis,
generateComparison,
analysisBusy,
comparisonBusy,
}
const setComparison: Dispatch<SetStateAction<CampaignComparison | null>> = (value) => {
setData((prev) => ({
...prev,
comparison: typeof value === 'function' ? value(prev.comparison) : value,
}))
}
return { ...data, refetch, setAnalysis, setComparison }
}

View File

@ -17,7 +17,7 @@ import StatCard from '../components/StatCard'
import {
campaignsApi, targetsApi, scenariosApi, modelConfigsApi,
type CampaignListItem, type Target, type Scenario, type Run,
type CampaignTimelineEntry, type ModelConfig,
type ModelConfig,
} from '../api'
import { passRateColor } from '../utils/colors'
import { shortDateTime, toDate } from '../utils/date'
@ -27,7 +27,6 @@ import CampaignRunTimeline from '../components/CampaignRunTimeline'
import PeriodComparisonSection from '../components/PeriodComparisonSection'
import ExplorationSection from '../components/ExplorationSection'
import { useResource } from '../hooks/useResource'
import { usePolling } from '../hooks/usePolling'
import { useCampaignReport } from '../hooks/useCampaignReport'
import { useTabStore } from '../stores/tabStore'
import { colors, statusColors, statusLabels, triggerColors, triggerLabels } from '../tokens'
@ -58,8 +57,6 @@ function scaleFor(realtime: boolean, windowSeconds: number, targetValue: number,
return realtime ? 1 : deriveTimeScale(windowSeconds, targetValue * targetUnit)
}
const POLL_INTERVAL_MS = 5000
const isActiveStatus = (status: string) => status === 'planned' || status === 'running'
/** How far an active campaign's window has progressed, in window seconds. */
@ -133,8 +130,7 @@ interface PlanFormEntry {
count?: number
}
interface CampaignsListsData {
campaigns: CampaignListItem[]
interface CampaignAuxiliaryData {
targets: Target[]
scenarios: Scenario[]
modelConfigs: ModelConfig[]
@ -156,23 +152,17 @@ export default function CampaignsPage() {
const [reportOpen, setReportOpen] = useState(false)
const [reportCampaignId, setReportCampaignId] = useState<string | null>(null)
const [analysisBusy, setAnalysisBusy] = useState(false)
const [expandedIds, setExpandedIds] = useState<string[]>([])
const [timelines, setTimelines] = useState<Record<string, CampaignTimelineEntry[]>>({})
const targetName = (id: string) => targets.find((t) => t.id === id)?.name ?? id.slice(0, 8)
const { data, loading, reload } = useResource<CampaignsListsData>(
const { data } = useResource<CampaignAuxiliaryData>(
async () => {
const [c, t, s, m] = await Promise.all([
campaignsApi.list(), targetsApi.list(), scenariosApi.list(), modelConfigsApi.list(),
const [t, s, m] = await Promise.all([
targetsApi.list(), scenariosApi.list(), modelConfigsApi.list(),
])
return { campaigns: c.data, targets: t.data, scenarios: s.data, modelConfigs: m.data }
return { targets: t.data, scenarios: s.data, modelConfigs: m.data }
},
{ tabPath: '/campaigns' },
)
const campaigns = data?.campaigns ?? []
const targets = data?.targets ?? []
const scenarios = data?.scenarios ?? []
const chatConfigs = (data?.modelConfigs ?? []).filter((m) => m.enabled && m.capability === 'chat')
@ -180,9 +170,28 @@ export default function CampaignsPage() {
const scenarioNames = Object.fromEntries(scenarios.map((s) => [s.id, s.name]))
const {
report, runs: reportRuns, timeline: reportTimeline, analysis, comparison, loading: reportLoading,
refetch: fetchReport, setAnalysis, setComparison,
} = useCampaignReport(reportOpen ? reportCampaignId : null)
list: campaignList,
report: reportSlot,
reloadList,
expandedIds,
timelines,
setTimelineExpanded,
generateAnalysis,
generateComparison,
analysisBusy,
} = useCampaignReport(
reportOpen ? reportCampaignId : null,
activeKey === '/campaigns',
)
const campaigns = campaignList.value
const loading = campaignList.phase === 'loading'
const reportSnapshot = reportSlot.value
const report = reportSnapshot?.report ?? null
const reportRuns = reportSnapshot?.runs ?? []
const reportTimeline = reportSnapshot?.timeline ?? []
const analysis = reportSnapshot?.analysis ?? null
const comparison = reportSnapshot?.comparison ?? null
const reportLoading = reportSlot.phase === 'loading'
const wPlan = (Form.useWatch('plan', form) as PlanFormEntry[] | undefined) ?? []
const planMarkers: TimelineMarker[] = wPlan
@ -200,15 +209,6 @@ export default function CampaignsPage() {
}
})
// Poll the list while this tab is active and a campaign is still working —
// compressed dev-line campaigns change fast. Stop once all are terminal.
const hasActiveCampaign = campaigns.some((c) => isActiveStatus(c.status))
usePolling(
() => void reload(true),
POLL_INTERVAL_MS,
activeKey === '/campaigns' && hasActiveCampaign,
)
const openCreate = () => {
form.setFieldsValue({
name: '', target_id: undefined, window_seconds: 24 * 3600,
@ -256,7 +256,7 @@ export default function CampaignsPage() {
})
message.success('评估活动已创建并开始调度')
setCreateOpen(false)
reload()
void reloadList()
} finally {
setSubmitting(false)
}
@ -265,17 +265,11 @@ export default function CampaignsPage() {
const cancelCampaign = async (id: string) => {
await campaignsApi.cancel(id)
message.success('活动已取消')
reload()
}
const fetchTimeline = async (campaignId: string) => {
const res = await campaignsApi.timeline(campaignId)
setTimelines((prev) => ({ ...prev, [campaignId]: res.data.entries }))
void reloadList()
}
const onExpandRow = (expanded: boolean, record: CampaignListItem) => {
setExpandedIds((prev) => (expanded ? [...prev, record.id] : prev.filter((x) => x !== record.id)))
if (expanded) void fetchTimeline(record.id)
setTimelineExpanded(record.id, expanded)
}
const openReport = (campaign: CampaignListItem) => {
@ -283,66 +277,6 @@ export default function CampaignsPage() {
setReportCampaignId(campaign.id)
}
// Keep the open report drawer live while its campaign is still running.
const reportCampaignActive = campaigns.some(
(c) => c.id === reportCampaignId && isActiveStatus(c.status),
)
usePolling(
() => { if (reportCampaignId) void fetchReport(true) },
POLL_INTERVAL_MS,
activeKey === '/campaigns' && reportOpen && !!reportCampaignId && reportCampaignActive,
)
const generateAnalysis = async () => {
if (!reportCampaignId) return
setAnalysisBusy(true)
try {
await campaignsApi.generateAnalysis(reportCampaignId)
const res = await campaignsApi.getAnalysis(reportCampaignId)
setAnalysis(res.data)
} finally {
setAnalysisBusy(false)
}
}
// 分析后台任务进行时轮询,直到进入 completed/failed 终态。
usePolling(
() => {
if (!reportCampaignId) return
void campaignsApi.getAnalysis(reportCampaignId).then((res) => setAnalysis(res.data))
},
POLL_INTERVAL_MS,
activeKey === '/campaigns' && reportOpen && !!reportCampaignId && analysis?.status === 'generating',
)
const generateComparison = async (baselineCampaignId?: string) => {
if (!reportCampaignId) return
await campaignsApi.generateComparison(reportCampaignId, baselineCampaignId)
const res = await campaignsApi.getComparison(reportCampaignId)
setComparison(res.data)
}
// 对比生成中同样轮询,直到 completed/failed。
usePolling(
() => {
if (!reportCampaignId) return
void campaignsApi.getComparison(reportCampaignId).then((res) => setComparison(res.data))
},
POLL_INTERVAL_MS,
activeKey === '/campaigns' && reportOpen && !!reportCampaignId && comparison?.status === 'generating',
)
// Grow the expanded timeline of any still-running campaign as new child Runs
// spawn. Completed/cancelled campaigns are fetched once on expand.
const activeExpandedIds = expandedIds.filter(
(id) => campaigns.some((c) => c.id === id && isActiveStatus(c.status)),
)
usePolling(
() => { activeExpandedIds.forEach((id) => void fetchTimeline(id)) },
POLL_INTERVAL_MS,
activeKey === '/campaigns' && activeExpandedIds.length > 0,
)
const columns = [
{ title: '名称', dataIndex: 'name', key: 'name' },
{ title: '评测对象', key: 'target', render: (_: unknown, c: CampaignListItem) => targetName(c.target_id) },
@ -688,7 +622,7 @@ export default function CampaignsPage() {
fullHeight
extra={
<Space>
<Button icon={<ReloadOutlined />} onClick={() => reload()} />
<Button icon={<ReloadOutlined />} onClick={() => void reloadList()} />
<Button type="primary" icon={<PlusOutlined />} onClick={openCreate}></Button>
</Space>
}

View File

@ -0,0 +1,46 @@
"""add intelligence job recovery attempts
Revision ID: d4e7f9a1b2c3
Revises: c2f4a6b8d0e1
Create Date: 2026-08-11
"""
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
revision: str = "d4e7f9a1b2c3"
down_revision: Union[str, Sequence[str], None] = "c2f4a6b8d0e1"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
inspector = sa.inspect(op.get_bind())
analysis_columns = {column["name"] for column in inspector.get_columns("campaign_analyses")}
if "recovery_attempts" not in analysis_columns:
op.add_column(
"campaign_analyses",
sa.Column("recovery_attempts", sa.Integer(), nullable=False, server_default="0"),
)
comparison_columns = {
column["name"] for column in inspector.get_columns("campaign_period_comparisons")
}
if "recovery_attempts" not in comparison_columns:
op.add_column(
"campaign_period_comparisons",
sa.Column("recovery_attempts", sa.Integer(), nullable=False, server_default="0"),
)
def downgrade() -> None:
inspector = sa.inspect(op.get_bind())
comparison_columns = {
column["name"] for column in inspector.get_columns("campaign_period_comparisons")
}
if "recovery_attempts" in comparison_columns:
op.drop_column("campaign_period_comparisons", "recovery_attempts")
analysis_columns = {column["name"] for column in inspector.get_columns("campaign_analyses")}
if "recovery_attempts" in analysis_columns:
op.drop_column("campaign_analyses", "recovery_attempts")

View File

@ -4,6 +4,7 @@ Revision ID: e6c3d1a2f809
Revises: d5b8c2e4f617
Create Date: 2026-07-29
"""
from typing import Sequence, Union
import sqlalchemy as sa
@ -35,16 +36,12 @@ def upgrade() -> None:
sa.PrimaryKeyConstraint("id"),
)
with op.batch_alter_table("eval_runs") as batch_op:
batch_op.add_column(
sa.Column("campaign_id", sqlmodel.sql.sqltypes.AutoString(), nullable=True)
)
batch_op.create_foreign_key(
"fk_eval_runs_campaign_id", "campaigns", ["campaign_id"], ["id"]
)
batch_op.add_column(sa.Column("campaign_id", sqlmodel.sql.sqltypes.AutoString(), nullable=True))
batch_op.create_foreign_key("fk_eval_runs_campaign_id", "campaigns", ["campaign_id"], ["id"])
def downgrade() -> None:
"""Remove Campaign storage without relying on SQLite foreign-key names."""
with op.batch_alter_table("eval_runs") as batch_op:
batch_op.drop_constraint("fk_eval_runs_campaign_id", type_="foreignkey")
batch_op.drop_column("campaign_id")
op.drop_table("campaigns")

View File

@ -27,7 +27,7 @@ def seeded_db(db_session, monkeypatch):
from agenteval.web.routers import campaigns as campaigns_module
monkeypatch.setattr(app_module, "init_db", lambda: None)
monkeypatch.setattr(campaigns_module, "start_campaign", lambda *a, **k: None)
monkeypatch.setattr(campaigns_module.campaign_runtime, "start", lambda *a, **k: None)
monkeypatch.setattr(db_module, "get_session", lambda: db_session)
monkeypatch.setattr(repo_module, "get_session", lambda: db_session)

View File

@ -7,7 +7,8 @@ skip silently. The analysis service itself is spied, not executed.
import pytest
from agenteval.evaluation import campaign_runner
from agenteval.evaluation.campaign_runner import request_cancel, start_campaign
from agenteval.evaluation.campaign_lifecycle import cancel_campaign
from agenteval.evaluation.campaign_runner import CampaignRuntime
from agenteval.models import (
Campaign,
CampaignPlanEntry,
@ -20,7 +21,6 @@ from agenteval.models import (
Scenario,
TargetStatus,
)
from agenteval.storage.db import utc_now
from agenteval.storage.repository import CampaignRepository, ScenarioRepository, TargetRepository
from tests.unit.mock_channel import MockChannel
@ -46,15 +46,23 @@ def seeded_db(db_session, monkeypatch):
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"])],
))
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
@ -63,7 +71,8 @@ def analysis_spy(monkeypatch):
"""Spy the analysis seam: resolvable model, recorded enqueue calls."""
calls: list[tuple[str, str]] = []
monkeypatch.setattr(
campaign_runner, "enqueue_campaign_analysis",
campaign_runner,
"enqueue_campaign_analysis",
lambda cid, *, triggered_by: calls.append((cid, triggered_by)),
)
monkeypatch.setattr(campaign_runner, "resolve_analysis_model", lambda campaign, session: object())
@ -82,56 +91,60 @@ def _make_campaign(session, **overrides) -> Campaign:
return CampaignRepository(session).create(Campaign(**payload))
async def _await_task(campaign_id, timeout=5.0):
@pytest.fixture()
async def runtime(seeded_db):
value = CampaignRuntime(session_factory=lambda: seeded_db, tick_seconds=TICK)
yield value
await value.shutdown()
async def _await_terminal(session, campaign_id, timeout=5.0):
import asyncio
task = campaign_runner.campaign_registry.get(campaign_id)
if task is not None:
await asyncio.wait_for(task, timeout=timeout)
async with asyncio.timeout(timeout):
while CampaignRepository(session).get(campaign_id).status is CampaignStatus.RUNNING:
await asyncio.sleep(TICK)
async def test_realtime_completion_auto_enqueues_analysis(seeded_db, analysis_spy):
async def test_realtime_completion_auto_enqueues_analysis(seeded_db, runtime, analysis_spy):
campaign = _make_campaign(seeded_db)
start_campaign(campaign.id, seeded_db, tick_seconds=TICK)
await _await_task(campaign.id)
assert runtime.start(campaign.id)
await _await_terminal(seeded_db, campaign.id)
final = CampaignRepository(seeded_db).get(campaign.id)
assert final.status == CampaignStatus.COMPLETED
assert analysis_spy == [(campaign.id, "auto")]
async def test_accelerated_completion_does_not_enqueue(seeded_db, analysis_spy):
async def test_accelerated_completion_does_not_enqueue(seeded_db, runtime, analysis_spy):
campaign = _make_campaign(seeded_db, time_scale=1000.0) # 加速调试线
start_campaign(campaign.id, seeded_db, tick_seconds=TICK)
await _await_task(campaign.id)
assert runtime.start(campaign.id)
await _await_terminal(seeded_db, campaign.id)
final = CampaignRepository(seeded_db).get(campaign.id)
assert final.status == CampaignStatus.COMPLETED
assert analysis_spy == []
async def test_cancelled_campaign_does_not_enqueue(seeded_db, analysis_spy):
async def test_cancelled_campaign_does_not_enqueue(seeded_db, runtime, analysis_spy):
import asyncio
campaign = _make_campaign(seeded_db, window_seconds=100)
start_campaign(campaign.id, seeded_db, tick_seconds=TICK)
assert runtime.start(campaign.id)
await asyncio.sleep(0.05)
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)
cancel_campaign(seeded_db, campaign.id, stop=runtime.cancel)
assert repo.get(campaign.id).status == CampaignStatus.CANCELLED
assert analysis_spy == []
async def test_missing_analysis_model_skips_silently(seeded_db, monkeypatch, analysis_spy):
async def test_missing_analysis_model_skips_silently(seeded_db, runtime, monkeypatch, analysis_spy):
monkeypatch.setattr(campaign_runner, "resolve_analysis_model", lambda campaign, session: None)
campaign = _make_campaign(seeded_db)
start_campaign(campaign.id, seeded_db, tick_seconds=TICK)
await _await_task(campaign.id)
assert runtime.start(campaign.id)
await _await_terminal(seeded_db, campaign.id)
final = CampaignRepository(seeded_db).get(campaign.id)
assert final.status == CampaignStatus.COMPLETED # 活动完成流程不受影响

View File

@ -32,7 +32,7 @@ def seeded_db(db_session, monkeypatch):
from agenteval.web.routers import campaigns as campaigns_module
monkeypatch.setattr(app_module, "init_db", lambda: None)
monkeypatch.setattr(campaigns_module, "start_campaign", lambda *a, **k: None)
monkeypatch.setattr(campaigns_module.campaign_runtime, "start", lambda *a, **k: None)
monkeypatch.setattr(db_module, "get_session", lambda: db_session)
monkeypatch.setattr(repo_module, "get_session", lambda: db_session)
@ -309,8 +309,8 @@ async def test_post_comparison_auto_baseline_success(client, seeded_db, monkeypa
monkeypatch.setattr(
campaigns_module,
"start_campaign_comparison",
lambda cid, *, triggered_by, baseline_campaign_id=None: _complete_comparison_row(
"enqueue_campaign_comparison",
lambda cid, *, triggered_by, baseline_campaign_id, session=None: _complete_comparison_row(
seeded_db, cid, baseline_campaign_id
),
)
@ -349,8 +349,8 @@ async def test_post_comparison_manual_baseline_across_fingerprints(client, seede
monkeypatch.setattr(
campaigns_module,
"start_campaign_comparison",
lambda cid, *, triggered_by, baseline_campaign_id=None: _complete_comparison_row(
"enqueue_campaign_comparison",
lambda cid, *, triggered_by, baseline_campaign_id, session=None: _complete_comparison_row(
seeded_db, cid, baseline_campaign_id
),
)
@ -394,8 +394,8 @@ async def test_post_comparison_rerun_upserts_without_new_row(client, seeded_db,
monkeypatch.setattr(
campaigns_module,
"start_campaign_comparison",
lambda cid, *, triggered_by, baseline_campaign_id=None: _complete_comparison_row(
"enqueue_campaign_comparison",
lambda cid, *, triggered_by, baseline_campaign_id, session=None: _complete_comparison_row(
seeded_db, cid, baseline_campaign_id
),
)

View File

@ -11,8 +11,8 @@ from types import SimpleNamespace
import pytest
from agenteval.evaluation import analysis as analysis_module
from agenteval.evaluation import comparison as comparison_module
from agenteval.evaluation.analysis import execute_campaign_analysis
from agenteval.evaluation import intelligence_jobs
from agenteval.evaluation.intelligence_jobs import execute_campaign_analysis_job
from agenteval.models import Campaign, CampaignPlanEntry, CampaignStatus, EvalRun, RunStatus, RunSummary
from agenteval.storage.db import CampaignAnalysisDB, utc_now
from agenteval.storage.model_config_repository import ModelConfigRepository
@ -75,21 +75,21 @@ def _seed_current(session, *, time_scale: float = 1.0, completed_at) -> None:
def comparison_spy(monkeypatch):
calls: list[tuple[str, str]] = []
def _spy(campaign_id, *, triggered_by, baseline_campaign_id=None):
def _spy(campaign_id, *, triggered_by, baseline_campaign_id, session=None):
calls.append((campaign_id, triggered_by))
monkeypatch.setattr(comparison_module, "start_campaign_comparison", _spy)
monkeypatch.setattr(intelligence_jobs, "enqueue_campaign_comparison", _spy)
return calls
async def test_production_line_with_baseline_enqueues_comparison(db_session, monkeypatch, comparison_spy):
monkeypatch.setattr(analysis_module, "get_session", lambda: db_session)
async def test_production_line_with_baseline_enqueues_comparison(db_session, comparison_spy):
_seed_config(db_session)
_seed_baseline(db_session, T0)
_seed_current(db_session, completed_at=T0 + timedelta(hours=2))
await execute_campaign_analysis(
await execute_campaign_analysis_job(
"camp-cur", triggered_by="auto", chat_client=FakeChatClient(STAGE1, STAGE2),
session_factory=lambda: db_session,
)
analysis_row = db_session.exec(
@ -100,32 +100,31 @@ async def test_production_line_with_baseline_enqueues_comparison(db_session, mon
async def test_accelerated_line_does_not_enqueue(db_session, monkeypatch, comparison_spy):
monkeypatch.setattr(analysis_module, "get_session", lambda: db_session)
_seed_config(db_session)
_seed_baseline(db_session, T0)
_seed_current(db_session, time_scale=24.0, completed_at=T0 + timedelta(hours=2))
await execute_campaign_analysis(
await execute_campaign_analysis_job(
"camp-cur", triggered_by="auto", chat_client=FakeChatClient(STAGE1, STAGE2),
session_factory=lambda: db_session,
)
assert comparison_spy == []
async def test_no_auto_baseline_does_not_enqueue(db_session, monkeypatch, comparison_spy):
monkeypatch.setattr(analysis_module, "get_session", lambda: db_session)
_seed_config(db_session)
_seed_current(db_session, completed_at=T0 + timedelta(hours=2)) # 无历史活动
await execute_campaign_analysis(
await execute_campaign_analysis_job(
"camp-cur", triggered_by="auto", chat_client=FakeChatClient(STAGE1, STAGE2),
session_factory=lambda: db_session,
)
assert comparison_spy == []
async def test_missing_model_skips_comparison(db_session, monkeypatch, comparison_spy):
monkeypatch.setattr(analysis_module, "get_session", lambda: db_session)
_seed_config(db_session)
_seed_baseline(db_session, T0)
_seed_current(db_session, completed_at=T0 + timedelta(hours=2))
@ -139,8 +138,9 @@ async def test_missing_model_skips_comparison(db_session, monkeypatch, compariso
monkeypatch.setattr(analysis_module, "resolve_analysis_model", _resolve)
await execute_campaign_analysis(
await execute_campaign_analysis_job(
"camp-cur", triggered_by="auto", chat_client=FakeChatClient(STAGE1, STAGE2),
session_factory=lambda: db_session,
)
analysis_row = db_session.exec(
@ -151,18 +151,18 @@ async def test_missing_model_skips_comparison(db_session, monkeypatch, compariso
async def test_comparison_enqueue_error_does_not_affect_analysis(db_session, monkeypatch):
monkeypatch.setattr(analysis_module, "get_session", lambda: db_session)
_seed_config(db_session)
_seed_baseline(db_session, T0)
_seed_current(db_session, completed_at=T0 + timedelta(hours=2))
def _boom(campaign_id, *, triggered_by, baseline_campaign_id=None):
def _boom(campaign_id, *, triggered_by, baseline_campaign_id, session=None):
raise RuntimeError("enqueue failed")
monkeypatch.setattr(comparison_module, "start_campaign_comparison", _boom)
monkeypatch.setattr(intelligence_jobs, "enqueue_campaign_comparison", _boom)
await execute_campaign_analysis(
await execute_campaign_analysis_job(
"camp-cur", triggered_by="auto", chat_client=FakeChatClient(STAGE1, STAGE2),
session_factory=lambda: db_session,
)
analysis_row = db_session.exec(

View File

@ -1,6 +1,6 @@
"""Integration test for campaign Run-spawning driven by a manual clock.
Drives ``advance_campaign`` at injected clock positions against a
Drives the runtime's internal advance seam at injected clock positions against a
compressed-time-scale campaign and asserts the spawned child Runs match the
plan (count, ownership, scenario) and complete with results/summary in the DB.
No real timer is used the clock is injected, mirroring how the durable loop
@ -8,7 +8,7 @@ No real timer is used — the clock is injected, mirroring how the durable loop
"""
import pytest
from agenteval.evaluation.campaign_runner import advance_campaign, reconcile_campaign_child_runs
from agenteval.evaluation.campaign_runner import _advance_campaign, _reconcile_campaign_child_runs
from agenteval.models import (
Campaign,
CampaignPlanEntry,
@ -50,7 +50,8 @@ def seeded_db(db_session, monkeypatch):
monkeypatch.setattr(factory_module.ChannelFactory, "create", lambda target: channel)
target = EvalTarget(
id="t-1", name="mock-target",
id="t-1",
name="mock-target",
platform=PlatformType.AI_DIGITAL_EMPLOYEE,
channel_type=ChannelType.TUTU_API,
channel_config={"base_url": "http://mock", "token": "x"},
@ -59,7 +60,8 @@ def seeded_db(db_session, monkeypatch):
TargetRepository(db_session).create(target)
scenario = Scenario(
id="s-1", name="mock-scenario",
id="s-1",
name="mock-scenario",
cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])],
)
ScenarioRepository(db_session).create(scenario)
@ -68,36 +70,36 @@ def seeded_db(db_session, monkeypatch):
def _make_campaign(session) -> Campaign:
return CampaignRepository(session).create(Campaign(
name="compressed",
target_id="t-1",
status="running",
window_seconds=7200,
time_scale=3600.0, # 1 real second == 3600 window seconds
plan=[
CampaignPlanEntry(scenario_id="s-1", offset_seconds=0, count=2),
CampaignPlanEntry(scenario_id="s-1", offset_seconds=3600, count=1),
],
))
return CampaignRepository(session).create(
Campaign(
name="compressed",
target_id="t-1",
status="running",
window_seconds=7200,
time_scale=3600.0, # 1 real second == 3600 window seconds
plan=[
CampaignPlanEntry(scenario_id="s-1", offset_seconds=0, count=2),
CampaignPlanEntry(scenario_id="s-1", offset_seconds=3600, count=1),
],
)
)
async def test_advance_spawns_due_runs_matching_plan(seeded_db):
campaign = _make_campaign(seeded_db)
# t=0 → window offset 0 → only entry 0 (count 2) is due.
r0 = await advance_campaign(campaign_id=campaign.id, elapsed_seconds=0.0, session=seeded_db)
r0 = await _advance_campaign(campaign_id=campaign.id, elapsed_seconds=0.0, session=seeded_db)
assert len(r0.spawned_run_ids) == 2
assert r0.finished is False
# t=1s → window offset 3600 → entry 1 (count 1) becomes due.
r1 = await advance_campaign(campaign_id=campaign.id, elapsed_seconds=1.0, session=seeded_db)
r1 = await _advance_campaign(campaign_id=campaign.id, elapsed_seconds=1.0, session=seeded_db)
assert len(r1.spawned_run_ids) == 1
runs = RunRepository(seeded_db).list_all()
assert len(runs) == 3
assert {
(run.campaign_plan_index, run.campaign_occurrence_index) for run in runs
} == {(0, 0), (0, 1), (1, 0)}
assert {(run.campaign_plan_index, run.campaign_occurrence_index) for run in runs} == {(0, 0), (0, 1), (1, 0)}
for run in runs:
assert run.campaign_id == campaign.id
assert run.scenario_id == "s-1"
@ -109,9 +111,9 @@ async def test_advance_spawns_due_runs_matching_plan(seeded_db):
async def test_advance_is_idempotent(seeded_db):
campaign = _make_campaign(seeded_db)
await advance_campaign(campaign_id=campaign.id, elapsed_seconds=1.0, session=seeded_db)
await _advance_campaign(campaign_id=campaign.id, elapsed_seconds=1.0, session=seeded_db)
# Re-advancing to the same clock must not double-spawn.
again = await advance_campaign(campaign_id=campaign.id, elapsed_seconds=1.0, session=seeded_db)
again = await _advance_campaign(campaign_id=campaign.id, elapsed_seconds=1.0, session=seeded_db)
assert again.spawned_run_ids == []
assert len(RunRepository(seeded_db).list_all()) == 3 # entry0(2) + entry1(1)
@ -119,12 +121,12 @@ async def test_advance_is_idempotent(seeded_db):
async def test_advance_reports_finished_at_window_end(seeded_db):
campaign = _make_campaign(seeded_db)
# t=2s → offset 7200 == window end.
result = await advance_campaign(campaign_id=campaign.id, elapsed_seconds=2.0, session=seeded_db)
result = await _advance_campaign(campaign_id=campaign.id, elapsed_seconds=2.0, session=seeded_db)
assert result.finished is True
async def test_advance_missing_campaign_returns_none(seeded_db):
assert await advance_campaign(campaign_id="nope", elapsed_seconds=0.0, session=seeded_db) is None
assert await _advance_campaign(campaign_id="nope", elapsed_seconds=0.0, session=seeded_db) is None
async def test_restart_after_claim_does_not_create_second_run(seeded_db):
@ -146,7 +148,7 @@ async def test_restart_after_claim_does_not_create_second_run(seeded_db):
)
assert claim.run is not None
result = await advance_campaign(campaign_id=campaign.id, elapsed_seconds=0.0, session=seeded_db)
result = await _advance_campaign(campaign_id=campaign.id, elapsed_seconds=0.0, session=seeded_db)
runs = RunRepository(seeded_db).list_by_campaign(campaign.id)
assert result.spawned_run_ids == []
@ -176,7 +178,7 @@ async def test_partial_claim_does_not_hide_remaining_occurrences(seeded_db):
campaign.summary = CampaignSummary(scheduler=SchedulerState(spawned_indices=[0]))
CampaignRepository(seeded_db).update(campaign)
await advance_campaign(campaign_id=campaign.id, elapsed_seconds=0.0, session=seeded_db)
await _advance_campaign(campaign_id=campaign.id, elapsed_seconds=0.0, session=seeded_db)
runs = RunRepository(seeded_db).list_by_campaign(campaign.id)
assert len(runs) == 2
@ -208,11 +210,11 @@ async def test_partial_spawn_failure_is_retried_until_all_occurrences_are_claime
monkeypatch.setattr(campaign_runner, "_spawn_child_run", flaky_spawn)
await advance_campaign(campaign_id=campaign.id, elapsed_seconds=0.0, session=seeded_db)
await _advance_campaign(campaign_id=campaign.id, elapsed_seconds=0.0, session=seeded_db)
after_failure = CampaignRepository(seeded_db).get(campaign.id)
assert after_failure.summary.scheduler.spawned_indices == []
await advance_campaign(campaign_id=campaign.id, elapsed_seconds=0.0, session=seeded_db)
await _advance_campaign(campaign_id=campaign.id, elapsed_seconds=0.0, session=seeded_db)
runs = RunRepository(seeded_db).list_by_campaign(campaign.id)
recovered = CampaignRepository(seeded_db).get(campaign.id)
assert {run.campaign_occurrence_index for run in runs} == {0, 1}
@ -230,7 +232,7 @@ async def test_recovery_resumes_pending_claim_without_replacing_identity(seeded_
)
assert claim.run is not None
result = await reconcile_campaign_child_runs(campaign.id, seeded_db)
result = await _reconcile_campaign_child_runs(campaign.id, seeded_db)
recovered = RunRepository(seeded_db).get(claim.run.id)
assert result is not None
@ -253,7 +255,7 @@ async def test_recovery_fails_running_claim_without_replaying_messages(seeded_db
claim.run.status = RunStatus.RUNNING
repo.update(claim.run)
result = await reconcile_campaign_child_runs(campaign.id, seeded_db)
result = await _reconcile_campaign_child_runs(campaign.id, seeded_db)
interrupted = repo.get(claim.run.id)
assert result is not None

View File

@ -1,6 +1,6 @@
"""Integration tests for the durable campaign scheduler loop (ticket 03).
Drives ``campaign_runner.start_campaign`` with a tiny tick and a compressed
Drives ``CampaignRuntime`` 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).
@ -10,7 +10,8 @@ from datetime import timedelta
import pytest
from agenteval.evaluation import campaign_runner
from agenteval.evaluation.campaign_runner import request_cancel, start_campaign
from agenteval.evaluation.campaign_lifecycle import cancel_campaign
from agenteval.evaluation.campaign_runner import CampaignRuntime
from agenteval.models import (
Campaign,
CampaignPlanEntry,
@ -50,15 +51,23 @@ def seeded_db(db_session, monkeypatch):
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"])],
))
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
@ -77,17 +86,25 @@ def _make_campaign(session, **overrides) -> Campaign:
return CampaignRepository(session).create(Campaign(**payload))
async def _await_task(campaign_id, timeout=3.0):
@pytest.fixture()
async def runtime(seeded_db):
value = CampaignRuntime(session_factory=lambda: seeded_db, tick_seconds=TICK)
yield value
await value.shutdown()
async def _await_terminal(session, campaign_id, timeout=3.0):
import asyncio
task = campaign_runner.campaign_registry.get(campaign_id)
if task is not None:
await asyncio.wait_for(task, timeout=timeout)
async with asyncio.timeout(timeout):
while CampaignRepository(session).get(campaign_id).status is CampaignStatus.RUNNING:
await asyncio.sleep(TICK)
async def test_loop_runs_to_completion(seeded_db):
async def test_loop_runs_to_completion(seeded_db, runtime):
campaign = _make_campaign(seeded_db)
start_campaign(campaign.id, seeded_db, tick_seconds=TICK)
await _await_task(campaign.id)
assert runtime.start(campaign.id)
await _await_terminal(seeded_db, campaign.id)
final = CampaignRepository(seeded_db).get(campaign.id)
assert final.status == CampaignStatus.COMPLETED
@ -98,7 +115,42 @@ async def test_loop_runs_to_completion(seeded_db):
assert all(r.status == RunStatus.COMPLETED for r in runs)
async def test_loop_retries_completion_after_settlement_failure(seeded_db, monkeypatch):
async def test_runtime_start_is_idempotent(seeded_db, runtime):
campaign = _make_campaign(seeded_db)
assert runtime.start(campaign.id)
assert runtime.start(campaign.id)
await _await_terminal(seeded_db, campaign.id)
assert len(RunRepository(seeded_db).list_by_campaign(campaign.id)) == 2
async def test_runtime_accepts_internal_clock_and_executor_adapters(seeded_db):
started_at = utc_now()
campaign = _make_campaign(seeded_db, started_at=started_at, status=CampaignStatus.RUNNING)
executed: list[tuple[int, int]] = []
async def execute_child(_campaign, _scenario, run, session):
executed.append((run.campaign_plan_index, run.campaign_occurrence_index))
run.status = RunStatus.COMPLETED
RunRepository(session).update(run)
runtime = CampaignRuntime(
session_factory=lambda: seeded_db,
now=lambda: started_at + timedelta(seconds=1),
tick_seconds=TICK,
execute_child_run=execute_child,
)
try:
assert runtime.start(campaign.id)
await _await_terminal(seeded_db, campaign.id)
finally:
await runtime.shutdown()
assert executed == [(0, 0), (1, 0)]
async def test_loop_retries_completion_after_settlement_failure(seeded_db, runtime, monkeypatch):
campaign = _make_campaign(seeded_db)
real_complete = campaign_runner.complete_campaign
attempts = 0
@ -111,14 +163,14 @@ async def test_loop_retries_completion_after_settlement_failure(seeded_db, monke
return real_complete(*args, **kwargs)
monkeypatch.setattr(campaign_runner, "complete_campaign", flaky_complete)
start_campaign(campaign.id, seeded_db, tick_seconds=TICK)
await _await_task(campaign.id)
assert runtime.start(campaign.id)
await _await_terminal(seeded_db, campaign.id)
assert attempts == 2
assert CampaignRepository(seeded_db).get(campaign.id).status is CampaignStatus.COMPLETED
async def test_restart_recovery_does_not_respawn(seeded_db):
async def test_restart_recovery_does_not_respawn(seeded_db, runtime):
# 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)
@ -128,8 +180,9 @@ async def test_restart_recovery_does_not_respawn(seeded_db):
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)
recovery = runtime.recover()
assert recovery.resumed_campaigns == 1
await _await_terminal(seeded_db, campaign.id)
final = CampaignRepository(seeded_db).get(campaign.id)
assert final.status == CampaignStatus.COMPLETED
@ -139,7 +192,7 @@ async def test_restart_recovery_does_not_respawn(seeded_db):
assert len(runs) == 1
async def test_restart_preserves_original_window_start(seeded_db):
async def test_restart_preserves_original_window_start(seeded_db, runtime):
original = utc_now() - timedelta(seconds=5)
campaign = _make_campaign(seeded_db, window_seconds=100000, time_scale=1.0)
campaign.status = CampaignStatus.RUNNING
@ -147,35 +200,32 @@ async def test_restart_preserves_original_window_start(seeded_db):
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)
assert runtime.start(campaign.id)
cancel_campaign(seeded_db, campaign.id, stop=runtime.cancel)
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):
async def test_cancel_stops_further_spawning(seeded_db, runtime):
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,
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)
assert runtime.start(campaign.id)
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)
cancel_campaign(seeded_db, campaign.id, stop=runtime.cancel)
final = repo.get(campaign.id)
assert final.status == CampaignStatus.CANCELLED

View File

@ -33,7 +33,7 @@ def seeded_db(db_session, monkeypatch):
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)
monkeypatch.setattr(campaigns_module.campaign_runtime, "start", lambda *a, **k: None)
def _test_get_session():
return db_session

View File

@ -9,7 +9,7 @@ import asyncio
import pytest
from agenteval.evaluation import campaign_runner as runner_module
from agenteval.evaluation.campaign_runner import run_campaign_loop
from agenteval.evaluation.campaign_runner import CampaignRuntime
from agenteval.exploration.models import ExplorationSessionStatus
from agenteval.models import (
Campaign,
@ -128,10 +128,15 @@ 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)
# time_scale 3600 → 1 real second covers the whole window.
runtime = CampaignRuntime(session_factory=lambda: seeded_db, tick_seconds=0.05)
try:
assert runtime.start(campaign.id)
async with asyncio.timeout(5):
while CampaignRepository(seeded_db).get(campaign.id).status.value == "running":
await asyncio.sleep(0.05)
finally:
await runtime.shutdown()
assert CampaignRepository(seeded_db).get("c-done").status.value == "completed"
settled = ExplorationSessionRepository(seeded_db).get(session_id)

View File

@ -0,0 +1,31 @@
"""Exercise the complete Alembic downgrade/upgrade chain on a fresh schema."""
from pathlib import Path
from alembic import command
from alembic.config import Config
from sqlalchemy import create_engine, inspect
from sqlmodel import SQLModel
def test_fresh_schema_round_trips_through_base(tmp_path: Path, monkeypatch):
from agenteval.storage import db as db_module
database_url = f"sqlite:///{tmp_path / 'roundtrip.db'}"
monkeypatch.setattr(db_module, "DATABASE_URL", database_url)
engine = create_engine(database_url)
SQLModel.metadata.create_all(engine)
config = Config(str(Path(__file__).resolve().parents[2] / "alembic.ini"))
foreign_keys = inspect(engine).get_foreign_keys("eval_runs")
assert any(key["constrained_columns"] == ["campaign_id"] and key["name"] is None for key in foreign_keys)
command.stamp(config, "head")
command.downgrade(config, "base")
command.upgrade(config, "head")
inspector = inspect(engine)
assert "campaigns" in inspector.get_table_names()
assert {"campaign_id", "campaign_plan_index", "campaign_occurrence_index"} <= {
column["name"] for column in inspector.get_columns("eval_runs")
}

View File

@ -7,9 +7,9 @@ from agenteval.evaluation.analysis import (
AnalysisError,
analyze_campaign,
collect_failure_samples,
execute_campaign_analysis,
resolve_analysis_model,
)
from agenteval.evaluation.intelligence_jobs import execute_campaign_analysis_job
from agenteval.models import Campaign, CampaignPlanEntry, EvalRun, RunStatus
from agenteval.storage.db import CampaignAnalysisDB, EvalResultDB, ModelConfigDB, TurnDB
from agenteval.storage.model_config_repository import ModelConfigRepository
@ -264,17 +264,16 @@ def test_collect_failure_samples_skips_passing_runs(db_session):
# ── 后台执行状态机 ───────────────────────────────────────────────────────
async def test_execute_writes_completed_row_with_model_snapshot(db_session, monkeypatch):
from agenteval.evaluation import analysis as analysis_module
monkeypatch.setattr(analysis_module, "get_session", lambda: db_session)
async def test_execute_writes_completed_row_with_model_snapshot(db_session):
_seed_config(db_session, "mc-default", analysis_default=True)
CampaignRepository(db_session).create(_campaign())
_seed_failed_run(db_session, "run-1", "s-1", failed_turns=1)
_seed_failed_run(db_session, "run-2", "s-2", failed_turns=1)
client = FakeChatClient(STAGE1_A, STAGE1_B, STAGE2)
await execute_campaign_analysis("camp-1", triggered_by="manual", chat_client=client)
await execute_campaign_analysis_job(
"camp-1", triggered_by="manual", chat_client=client, session_factory=lambda: db_session,
)
row = db_session.exec(
select(CampaignAnalysisDB).where(CampaignAnalysisDB.campaign_id == "camp-1")
@ -285,15 +284,14 @@ async def test_execute_writes_completed_row_with_model_snapshot(db_session, monk
assert row.get_result()["overall"].startswith("活动整体不达标")
async def test_execute_records_failure(db_session, monkeypatch):
from agenteval.evaluation import analysis as analysis_module
monkeypatch.setattr(analysis_module, "get_session", lambda: db_session)
async def test_execute_records_failure(db_session):
_seed_config(db_session, "mc-default", analysis_default=True)
CampaignRepository(db_session).create(_campaign())
client = FakeChatClient("garbage")
await execute_campaign_analysis("camp-1", triggered_by="auto", chat_client=client)
await execute_campaign_analysis_job(
"camp-1", triggered_by="auto", chat_client=client, session_factory=lambda: db_session,
)
row = db_session.exec(
select(CampaignAnalysisDB).where(CampaignAnalysisDB.campaign_id == "camp-1")

View File

@ -8,10 +8,10 @@ from agenteval.evaluation.comparison import (
ComparisonError,
campaign_plan_fingerprint,
compute_metric_diff,
execute_campaign_comparison,
narrate_period_comparison,
resolve_auto_baseline,
)
from agenteval.evaluation.intelligence_jobs import execute_campaign_comparison_job
from agenteval.models import Campaign, CampaignPlanEntry, CampaignStatus, EvalRun, RunStatus, RunSummary
from agenteval.storage.db import (
CampaignAnalysisDB,
@ -382,18 +382,16 @@ def _seed_campaign_with_analysis(session, campaign: Campaign, analysis_result: d
))
async def test_execute_writes_completed_row_with_baseline_snapshot(db_session, monkeypatch):
from agenteval.evaluation import comparison as comparison_module
monkeypatch.setattr(comparison_module, "get_session", lambda: db_session)
async def test_execute_writes_completed_row_with_baseline_snapshot(db_session):
_seed_config(db_session, "mc-default")
baseline = _campaign("camp-base", completed_at=T0)
current = _campaign("camp-cur", completed_at=T0 + timedelta(hours=2))
_seed_campaign_with_analysis(db_session, baseline, _analysis("上期"))
_seed_campaign_with_analysis(db_session, current, _analysis("本期"))
await execute_campaign_comparison(
await execute_campaign_comparison_job(
"camp-cur", triggered_by="manual", chat_client=FakeChatClient(NARRATION),
session_factory=lambda: db_session,
)
row = db_session.exec(
@ -405,18 +403,16 @@ async def test_execute_writes_completed_row_with_baseline_snapshot(db_session, m
assert row.get_result()["trend"] == "improving"
async def test_execute_records_failure_on_unparseable_output(db_session, monkeypatch):
from agenteval.evaluation import comparison as comparison_module
monkeypatch.setattr(comparison_module, "get_session", lambda: db_session)
async def test_execute_records_failure_on_unparseable_output(db_session):
_seed_config(db_session, "mc-default")
baseline = _campaign("camp-base", completed_at=T0)
current = _campaign("camp-cur", completed_at=T0 + timedelta(hours=2))
_seed_campaign_with_analysis(db_session, baseline, _analysis("上期"))
_seed_campaign_with_analysis(db_session, current, _analysis("本期"))
await execute_campaign_comparison(
await execute_campaign_comparison_job(
"camp-cur", triggered_by="manual", chat_client=FakeChatClient("garbage"),
session_factory=lambda: db_session,
)
row = db_session.exec(
@ -427,18 +423,16 @@ async def test_execute_records_failure_on_unparseable_output(db_session, monkeyp
assert row.baseline_campaign_id == "camp-base"
async def test_execute_fails_without_baseline_analysis(db_session, monkeypatch):
from agenteval.evaluation import comparison as comparison_module
monkeypatch.setattr(comparison_module, "get_session", lambda: db_session)
async def test_execute_fails_without_baseline_analysis(db_session):
_seed_config(db_session, "mc-default")
baseline = _campaign("camp-base", completed_at=T0)
current = _campaign("camp-cur", completed_at=T0 + timedelta(hours=2))
CampaignRepository(db_session).create(baseline) # 基线无分析行
_seed_campaign_with_analysis(db_session, current, _analysis("本期"))
await execute_campaign_comparison(
await execute_campaign_comparison_job(
"camp-cur", triggered_by="manual", chat_client=FakeChatClient(NARRATION),
session_factory=lambda: db_session,
)
row = db_session.exec(

View File

@ -13,9 +13,9 @@ def test_create_campaign_commits_running_before_launch(db_session):
ScenarioRepository(db_session).create(Scenario(id="s-1", name="scenario", cases=[Case(id="c-1", messages=["hi"])]))
launched: list[str] = []
def launch(campaign_id, session):
def launch(campaign_id):
launched.append(campaign_id)
assert CampaignRepository(session).get(campaign_id).status is CampaignStatus.RUNNING
assert CampaignRepository(db_session).get(campaign_id).status is CampaignStatus.RUNNING
campaign = create_campaign(
db_session,
@ -36,7 +36,7 @@ def test_create_campaign_launch_failure_leaves_durable_row(db_session):
TargetRepository(db_session).create(_make_target())
ScenarioRepository(db_session).create(Scenario(id="s-1", name="scenario", cases=[Case(id="c-1", messages=["hi"])]))
def launch(_campaign_id, _session):
def launch(_campaign_id):
raise RuntimeError("scheduler unavailable")
with pytest.raises(RuntimeError):

View File

@ -0,0 +1,68 @@
"""Campaign read-model projections and batched child-Run loading."""
from datetime import datetime, timezone
from agenteval.evaluation.campaign_read_model import CampaignReadModel
from agenteval.models import Campaign, CampaignPlanEntry, EvalRun, RunStatus
from agenteval.storage.repository import CampaignRepository, RunRepository
def _campaign(session, campaign_id: str) -> Campaign:
return CampaignRepository(session).create(Campaign(
id=campaign_id,
name=campaign_id,
target_id="target-1",
window_seconds=60,
plan=[CampaignPlanEntry(scenario_id="scenario-1", offset_seconds=0)],
))
def _run(session, campaign_id: str, run_id: str) -> EvalRun:
return RunRepository(session).create(EvalRun(
id=run_id,
target_id="target-1",
scenario_id="scenario-1",
campaign_id=campaign_id,
status=RunStatus.COMPLETED,
started_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
summary={"total_cases": 1, "passed_cases": 1, "pass_rate": 1.0},
))
def test_list_items_empty(db_session):
assert CampaignReadModel(db_session).list_items() == []
def test_list_items_batches_runs_for_multiple_campaigns(db_session, monkeypatch):
first = _campaign(db_session, "campaign-1")
second = _campaign(db_session, "campaign-2")
_run(db_session, first.id, "run-1")
_run(db_session, second.id, "run-2")
def reject_per_campaign_query(*args, **kwargs):
raise AssertionError("list_items must not query child Runs per Campaign")
monkeypatch.setattr(RunRepository, "list_by_campaign", reject_per_campaign_query)
items = CampaignReadModel(db_session).list_items()
progress_by_id = {item["id"]: item["progress"] for item in items}
assert set(progress_by_id) == {first.id, second.id}
assert progress_by_id[first.id]["completed_runs"] == 1
assert progress_by_id[second.id]["completed_runs"] == 1
def test_markdown_projection_uses_json_report_projection(db_session):
campaign = _campaign(db_session, "campaign-1")
_run(db_session, campaign.id, "run-1")
read_model = CampaignReadModel(db_session)
json_report = read_model.report(campaign.id)
markdown_projection = read_model.markdown_projection(campaign.id)
assert json_report is not None
assert markdown_projection is not None
assert markdown_projection["report"] == {
key: value for key, value in json_report.items() if key != "exploration"
}
assert markdown_projection["exploration"] == json_report.get("exploration")

View File

@ -1,11 +1,20 @@
"""Tests for the unified durable Campaign startup recovery seam."""
"""Tests for the CampaignRuntime startup recovery interface."""
from agenteval.evaluation import campaign_lifecycle
import asyncio
from types import SimpleNamespace
from agenteval.evaluation import campaign_runner
from agenteval.evaluation.campaign_runner import CampaignRuntime
from agenteval.models import CampaignStatus
def test_recovery_coordinator_orders_cleanup_before_relaunch(db_session, monkeypatch):
async def test_runtime_recovery_repairs_before_relaunch(monkeypatch):
calls: list[str] = []
class FakeSession:
def close(self):
calls.append("close")
class FakeRuns:
def __init__(self, _session):
pass
@ -14,33 +23,27 @@ def test_recovery_coordinator_orders_cleanup_before_relaunch(db_session, monkeyp
calls.append("runs")
return 2
class FakeAnalysis:
class FakeCampaigns:
def __init__(self, _session):
pass
def mark_orphans_failed(self):
calls.append("analysis")
return 1
def list_all(self):
calls.append("campaigns")
return [SimpleNamespace(id="campaign-1", status=CampaignStatus.RUNNING)]
class FakeComparison(FakeAnalysis):
def mark_orphans_failed(self):
calls.append("comparison")
return 1
async def fake_loop(campaign_id, cancel, **kwargs):
calls.append(f"launch:{campaign_id}")
monkeypatch.setattr(campaign_lifecycle, "RunRepository", FakeRuns)
monkeypatch.setattr(campaign_lifecycle, "CampaignAnalysisRepository", FakeAnalysis)
monkeypatch.setattr(campaign_lifecycle, "CampaignPeriodComparisonRepository", FakeComparison)
monkeypatch.setattr(campaign_runner, "RunRepository", FakeRuns)
monkeypatch.setattr(campaign_runner, "CampaignRepository", FakeCampaigns)
monkeypatch.setattr(campaign_runner, "_run_campaign_loop", fake_loop)
runtime = CampaignRuntime(session_factory=FakeSession)
import agenteval.evaluation.analysis as analysis
import agenteval.evaluation.campaign_runner as runner
summary = runtime.recover()
await asyncio.sleep(0)
await runtime.shutdown()
monkeypatch.setattr(analysis, "resume_queued_campaign_analysis", lambda _session: calls.append("queued") or 3)
monkeypatch.setattr(runner, "resume_running_campaigns", lambda _session, **_: calls.append("campaigns") or 4)
summary = campaign_lifecycle.recover_campaign_runtime(db_session)
assert calls == ["runs", "analysis", "comparison", "campaigns", "queued"]
assert summary.interrupted_runs == 2
assert summary.interrupted_analysis == 2
assert summary.resumed_campaigns == 4
assert summary.resumed_analysis == 3
assert summary.resumed_campaigns == 1
assert calls[:3] == ["runs", "campaigns", "close"]
assert "launch:campaign-1" in calls

View File

@ -18,8 +18,8 @@ def test_start_planned_campaign_is_atomic_and_launches_after_commit(db_session):
)
observed = []
def launch(campaign_id, session):
observed.append(CampaignRepository(session).get(campaign_id).status)
def launch(campaign_id):
observed.append(CampaignRepository(db_session).get(campaign_id).status)
started = start_campaign(db_session, "campaign-1", launch=launch)

View File

@ -8,9 +8,10 @@ generating 行标记为 failed。
import asyncio
import pytest
from agenteval.evaluation import analysis, comparison
from agenteval.evaluation import intelligence_jobs
from agenteval.exploration import judge
from agenteval.storage.repository import (
AsyncJobClaimStatus,
CampaignAnalysisRepository,
CampaignPeriodComparisonRepository,
)
@ -38,19 +39,24 @@ def db_session(tmp_path):
engine.dispose()
async def test_start_analysis_registers_in_registry(monkeypatch):
monkeypatch.setattr(analysis, "execute_campaign_analysis", lambda *a, **kw: asyncio.sleep(0))
analysis.start_campaign_analysis("c-1", triggered_by="manual")
assert analysis.analysis_registry.is_running("c-1")
await analysis.analysis_registry.shutdown_all()
assert not analysis.analysis_registry.is_running("c-1")
async def test_start_analysis_registers_in_registry(db_session, monkeypatch):
monkeypatch.setattr(intelligence_jobs, "execute_campaign_analysis_job", lambda *a, **kw: asyncio.sleep(0))
intelligence_jobs.enqueue_campaign_analysis("c-1", triggered_by="manual", session=db_session)
assert intelligence_jobs.is_intelligence_job_running("analysis", "c-1")
await intelligence_jobs.shutdown_campaign_intelligence_jobs()
assert not intelligence_jobs.is_intelligence_job_running("analysis", "c-1")
async def test_start_comparison_registers_in_registry(monkeypatch):
monkeypatch.setattr(comparison, "execute_campaign_comparison", lambda *a, **kw: asyncio.sleep(0))
comparison.start_campaign_comparison("c-2", triggered_by="manual")
assert comparison.comparison_registry.is_running("c-2")
await comparison.comparison_registry.shutdown_all()
async def test_start_comparison_registers_in_registry(db_session, monkeypatch):
monkeypatch.setattr(intelligence_jobs, "execute_campaign_comparison_job", lambda *a, **kw: asyncio.sleep(0))
intelligence_jobs.enqueue_campaign_comparison(
"c-2",
triggered_by="manual",
baseline_campaign_id="c-1",
session=db_session,
)
assert intelligence_jobs.is_intelligence_job_running("comparison", "c-2")
await intelligence_jobs.shutdown_campaign_intelligence_jobs()
async def test_start_judge_registers_in_registry(monkeypatch):
@ -60,29 +66,34 @@ async def test_start_judge_registers_in_registry(monkeypatch):
await judge.judge_registry.shutdown_all()
async def test_launch_is_idempotent_for_live_id(monkeypatch):
async def test_launch_is_idempotent_for_live_id(db_session, monkeypatch):
gate = asyncio.Event()
async def hang(*args, **kwargs):
await gate.wait()
monkeypatch.setattr(analysis, "execute_campaign_analysis", hang)
analysis.start_campaign_analysis("c-dup", triggered_by="manual")
analysis.start_campaign_analysis("c-dup", triggered_by="manual")
assert len(analysis.analysis_registry._tasks) == 1
monkeypatch.setattr(intelligence_jobs, "execute_campaign_analysis_job", hang)
intelligence_jobs.enqueue_campaign_analysis("c-dup", triggered_by="manual", session=db_session)
intelligence_jobs.enqueue_campaign_analysis("c-dup", triggered_by="manual", session=db_session)
assert intelligence_jobs.is_intelligence_job_running("analysis", "c-dup")
gate.set()
await analysis.analysis_registry.shutdown_all()
await intelligence_jobs.shutdown_campaign_intelligence_jobs()
async def test_shutdown_all_cancels_hanging_task(monkeypatch):
async def test_shutdown_all_cancels_hanging_task(db_session, monkeypatch):
async def hang(*args, **kwargs):
await asyncio.Event().wait()
monkeypatch.setattr(comparison, "execute_campaign_comparison", hang)
comparison.start_campaign_comparison("c-hang", triggered_by="manual")
assert comparison.comparison_registry.is_running("c-hang")
await comparison.comparison_registry.shutdown_all()
assert not comparison.comparison_registry.is_running("c-hang")
monkeypatch.setattr(intelligence_jobs, "execute_campaign_comparison_job", hang)
intelligence_jobs.enqueue_campaign_comparison(
"c-hang",
triggered_by="manual",
baseline_campaign_id="c-base",
session=db_session,
)
assert intelligence_jobs.is_intelligence_job_running("comparison", "c-hang")
await intelligence_jobs.shutdown_campaign_intelligence_jobs()
assert not intelligence_jobs.is_intelligence_job_running("comparison", "c-hang")
def test_mark_orphans_failed_flips_generating_analysis(db_session):
@ -99,13 +110,14 @@ def test_mark_orphans_failed_flips_generating_analysis(db_session):
def test_enqueue_analysis_persists_before_launch(db_session, monkeypatch):
from agenteval.evaluation import analysis
launched = []
monkeypatch.setattr(analysis, "get_session", lambda: db_session)
monkeypatch.setattr(analysis, "start_campaign_analysis", lambda cid, *, triggered_by: launched.append((cid, triggered_by)))
monkeypatch.setattr(
intelligence_jobs,
"_launch_analysis",
lambda cid, *, triggered_by: launched.append((cid, triggered_by)),
)
analysis.enqueue_campaign_analysis("c-queued", triggered_by="auto")
intelligence_jobs.enqueue_campaign_analysis("c-queued", triggered_by="auto", session=db_session)
assert launched == [("c-queued", "auto")]
row = CampaignAnalysisRepository(db_session).get_by_campaign("c-queued")
@ -113,15 +125,36 @@ def test_enqueue_analysis_persists_before_launch(db_session, monkeypatch):
assert row.triggered_by == "auto"
def test_resume_queued_analysis_relaunches_persisted_jobs(db_session, monkeypatch):
from agenteval.evaluation import analysis
def test_only_one_session_can_claim_a_queued_analysis(db_session):
repo = CampaignAnalysisRepository(db_session)
repo.enqueue("c-claim", triggered_by="auto")
competing_session = Session(db_session.get_bind())
try:
assert repo.claim_queued("c-claim").status is AsyncJobClaimStatus.CLAIMED
assert (
CampaignAnalysisRepository(competing_session).claim_queued("c-claim").status
is AsyncJobClaimStatus.ALREADY_CLAIMED
)
finally:
competing_session.close()
row = repo.get_by_campaign("c-claim")
assert row.status == "generating"
assert row.triggered_by == "auto"
def test_recovery_relaunches_persisted_analysis_jobs(db_session, monkeypatch):
repo = CampaignAnalysisRepository(db_session)
repo.enqueue("c-queued", triggered_by="auto")
launched = []
monkeypatch.setattr(analysis, "start_campaign_analysis", lambda cid, *, triggered_by: launched.append((cid, triggered_by)))
monkeypatch.setattr(
intelligence_jobs,
"_launch_analysis",
lambda cid, *, triggered_by: launched.append((cid, triggered_by)),
)
assert analysis.resume_queued_campaign_analysis(db_session) == 1
assert intelligence_jobs.recover_campaign_intelligence_jobs(db_session) == (0, 1)
assert launched == [("c-queued", "auto")]
@ -136,3 +169,126 @@ def test_mark_orphans_failed_flips_generating_comparison(db_session):
assert repo.get_by_campaign("c-gen").status == "failed"
assert repo.get_by_campaign("c-gen").error
assert repo.get_by_campaign("c-done").status == "completed"
def test_enqueue_comparison_preserves_baseline_before_claim(db_session):
repo = CampaignPeriodComparisonRepository(db_session)
row = repo.enqueue(
"c-current",
baseline_campaign_id="c-baseline",
triggered_by="auto",
)
assert row.status == "queued"
assert row.baseline_campaign_id == "c-baseline"
assert row.triggered_by == "auto"
assert repo.claim_queued("c-current").status is AsyncJobClaimStatus.CLAIMED
assert repo.claim_queued("c-current").status is AsyncJobClaimStatus.ALREADY_CLAIMED
def test_enqueue_comparison_persists_before_launch(db_session, monkeypatch):
launched = []
monkeypatch.setattr(
intelligence_jobs,
"_launch_comparison",
lambda cid, *, triggered_by, baseline_campaign_id: launched.append(
(cid, triggered_by, baseline_campaign_id)
),
)
intelligence_jobs.enqueue_campaign_comparison(
"c-current",
baseline_campaign_id="c-baseline",
triggered_by="auto",
session=db_session,
)
assert launched == [("c-current", "auto", "c-baseline")]
row = CampaignPeriodComparisonRepository(db_session).get_by_campaign("c-current")
assert row.status == "queued"
assert row.baseline_campaign_id == "c-baseline"
def test_recovery_relaunches_persisted_comparison_jobs(db_session, monkeypatch):
CampaignPeriodComparisonRepository(db_session).enqueue(
"c-current",
baseline_campaign_id="c-baseline",
triggered_by="auto",
)
launched = []
monkeypatch.setattr(
intelligence_jobs,
"_launch_comparison",
lambda cid, *, triggered_by, baseline_campaign_id: launched.append(
(cid, triggered_by, baseline_campaign_id)
),
)
assert intelligence_jobs.recover_campaign_intelligence_jobs(db_session) == (0, 1)
assert launched == [("c-current", "auto", "c-baseline")]
def test_claim_reports_missing_and_terminal_jobs(db_session):
repo = CampaignAnalysisRepository(db_session)
assert repo.claim_queued("missing").status is AsyncJobClaimStatus.NOT_FOUND
repo.upsert("c-done", status="completed", result={"ok": True})
assert repo.claim_queued("c-done").status is AsyncJobClaimStatus.NOT_QUEUED
def test_reenqueue_does_not_move_generating_analysis_back_to_queued(db_session, monkeypatch):
repo = CampaignAnalysisRepository(db_session)
repo.upsert("c-live", status="generating", triggered_by="auto")
launched = []
monkeypatch.setattr(
intelligence_jobs,
"_launch_analysis",
lambda cid, *, triggered_by: launched.append((cid, triggered_by)),
)
intelligence_jobs.enqueue_campaign_analysis("c-live", triggered_by="manual", session=db_session)
row = repo.get_by_campaign("c-live")
assert row.status == "generating"
assert row.triggered_by == "auto"
assert launched == []
def test_reenqueue_preserves_queued_comparison_context(db_session):
repo = CampaignPeriodComparisonRepository(db_session)
repo.enqueue("c-current", baseline_campaign_id="c-original", triggered_by="auto")
row = repo.enqueue("c-current", baseline_campaign_id="c-replacement", triggered_by="manual")
assert row.status == "queued"
assert row.baseline_campaign_id == "c-original"
assert row.triggered_by == "auto"
@pytest.mark.parametrize(
("repository", "enqueue_kwargs"),
[
(CampaignAnalysisRepository, {}),
(CampaignPeriodComparisonRepository, {"baseline_campaign_id": "c-base"}),
],
)
def test_queued_recovery_is_bounded(db_session, repository, enqueue_kwargs):
repo = repository(db_session)
repo.enqueue("c-recover", triggered_by="auto", **enqueue_kwargs)
for expected_attempts in range(1, intelligence_jobs.MAX_QUEUED_RECOVERY_ATTEMPTS + 1):
recovered = repo.prepare_queued_recovery(
intelligence_jobs.MAX_QUEUED_RECOVERY_ATTEMPTS,
"recovery exhausted",
)
assert [row.campaign_id for row in recovered] == ["c-recover"]
assert repo.get_by_campaign("c-recover").recovery_attempts == expected_attempts
assert repo.prepare_queued_recovery(
intelligence_jobs.MAX_QUEUED_RECOVERY_ATTEMPTS,
"recovery exhausted",
) == []
exhausted = repo.get_by_campaign("c-recover")
assert exhausted.status == "failed"
assert exhausted.error == "recovery exhausted"