feat(campaigns): add Campaign persistence and create/query API

Introduce the 评估活动 (Campaign) aggregate above Run: a single-target,
service-cycle window driving a static plan. Adds Campaign/CampaignPlanEntry
models, CampaignDB table, nullable eval_runs.campaign_id, CampaignRepository,
Alembic migration, and POST/GET /api/campaigns with validation.

Ticket 01 of v0.6; no scheduling or child-run spawning yet (ADR-0003 v1).
This commit is contained in:
sinohqb 2026-07-30 11:56:19 +08:00
parent af7e7bf110
commit e4404f1fa2
9 changed files with 425 additions and 2 deletions

View File

@ -164,6 +164,8 @@ class EvalRun(BaseModel):
scenario_id: str
# 创建时快照的场景考纲版本ADR-0001
scenario_version: int = 1
# 归属的评估活动Campaign手动/单次运行为空
campaign_id: Optional[str] = None
status: RunStatus = RunStatus.PENDING
triggered_by: RunTrigger = RunTrigger.MANUAL
started_at: Optional[datetime] = None
@ -171,6 +173,41 @@ class EvalRun(BaseModel):
summary: Optional[dict[str, Any]] = None
class CampaignStatus(str, Enum):
"""Lifecycle of an evaluation campaign (评估活动)."""
PLANNED = "planned"
RUNNING = "running"
COMPLETED = "completed"
CANCELLED = "cancelled"
FAILED = "failed"
class CampaignPlanEntry(BaseModel):
"""One static plan item: run a scenario N times at a window offset."""
scenario_id: str
offset_seconds: int = Field(ge=0)
count: int = Field(default=1, ge=1)
class Campaign(BaseModel):
"""An evaluation campaign: a service-cycle window over a single target,
driving many child Runs from a static plan (ADR-0003)."""
id: Optional[str] = None
name: str
target_id: str
window_seconds: int = Field(gt=0)
time_scale: float = Field(default=1.0, gt=0)
plan: list[CampaignPlanEntry] = Field(min_length=1)
status: CampaignStatus = CampaignStatus.PLANNED
started_at: Optional[datetime] = None
completed_at: Optional[datetime] = None
created_at: Optional[datetime] = None
summary: Optional[dict[str, Any]] = None
class Turn(BaseModel):
"""A single turn in a conversation during evaluation."""

View File

@ -160,6 +160,36 @@ class ScenarioModelBindingDB(SQLModel, table=True):
model_config_id: str = Field(foreign_key="model_configs.id", index=True)
class CampaignDB(SQLModel, table=True):
"""Database table for evaluation campaigns (评估活动)."""
__tablename__ = "campaigns"
id: Optional[str] = Field(default_factory=new_uuid, primary_key=True)
name: str
target_id: Optional[str] = Field(default=None, foreign_key="eval_targets.id")
window_seconds: int
time_scale: float = 1.0
plan: str = "[]"
status: str = "planned"
started_at: Optional[datetime] = None
completed_at: Optional[datetime] = None
created_at: Optional[datetime] = Field(default_factory=utc_now)
summary: Optional[str] = None
def get_plan(self) -> list[dict[str, Any]]:
return json.loads(self.plan)
def set_plan(self, plan: list[dict[str, Any]]) -> None:
self.plan = json.dumps(plan, ensure_ascii=False)
def get_summary(self) -> Optional[dict[str, Any]]:
return json.loads(self.summary) if self.summary else None
def set_summary(self, summary: dict[str, Any]) -> None:
self.summary = json.dumps(summary, ensure_ascii=False)
class EvalRunDB(SQLModel, table=True):
"""Database table for evaluation runs."""
@ -169,6 +199,7 @@ class EvalRunDB(SQLModel, table=True):
target_id: Optional[str] = Field(default=None, foreign_key="eval_targets.id")
scenario_id: Optional[str] = Field(default=None, foreign_key="scenarios.id")
scenario_version: int = Field(default=1)
campaign_id: Optional[str] = Field(default=None, foreign_key="campaigns.id")
status: str = "pending"
triggered_by: str = Field(default="manual")
started_at: Optional[datetime] = Field(default_factory=utc_now)

View File

@ -4,9 +4,10 @@ from typing import Optional
from sqlmodel import Session, select
from agenteval.models import Case, EvalResult, EvalRun, EvalTarget, Scenario
from agenteval.models import Campaign, Case, EvalResult, EvalRun, EvalTarget, Scenario
from agenteval.services.model_configs import ModelConfigService
from agenteval.storage.db import (
CampaignDB,
EvalResultDB,
EvalRunDB,
EvalTargetDB,
@ -84,6 +85,7 @@ def _run_to_db(run: EvalRun) -> EvalRunDB:
target_id=run.target_id,
scenario_id=run.scenario_id,
scenario_version=run.scenario_version,
campaign_id=run.campaign_id,
status=run.status.value,
triggered_by=run.triggered_by.value,
started_at=run.started_at,
@ -100,6 +102,7 @@ def _run_from_db(db: EvalRunDB) -> EvalRun:
target_id=db.target_id,
scenario_id=db.scenario_id,
scenario_version=db.scenario_version or 1,
campaign_id=db.campaign_id,
status=db.status,
triggered_by=db.triggered_by or "manual",
started_at=db.started_at,
@ -108,6 +111,40 @@ def _run_from_db(db: EvalRunDB) -> EvalRun:
)
def _campaign_to_db(campaign: Campaign) -> CampaignDB:
db = CampaignDB(
id=campaign.id,
name=campaign.name,
target_id=campaign.target_id,
window_seconds=campaign.window_seconds,
time_scale=campaign.time_scale,
status=campaign.status.value,
started_at=campaign.started_at,
completed_at=campaign.completed_at,
created_at=campaign.created_at,
)
db.set_plan([entry.model_dump(mode="json") for entry in campaign.plan])
if campaign.summary:
db.set_summary(campaign.summary)
return db
def _campaign_from_db(db: CampaignDB) -> Campaign:
return Campaign(
id=db.id,
name=db.name,
target_id=db.target_id,
window_seconds=db.window_seconds,
time_scale=db.time_scale,
plan=db.get_plan(),
status=db.status,
started_at=db.started_at,
completed_at=db.completed_at,
created_at=db.created_at,
summary=db.get_summary(),
)
def _result_to_db(result: EvalResult) -> EvalResultDB:
return EvalResultDB(
id=result.id,
@ -303,6 +340,7 @@ class RunRepository:
existing.target_id = run.target_id
existing.scenario_id = run.scenario_id
existing.scenario_version = run.scenario_version
existing.campaign_id = run.campaign_id
existing.status = run.status.value
existing.triggered_by = run.triggered_by.value
existing.completed_at = run.completed_at
@ -331,6 +369,28 @@ class RunRepository:
return True
class CampaignRepository:
"""Repository for evaluation campaigns (评估活动)."""
def __init__(self, session: Optional[Session] = None):
self.session = session or get_session()
def list_all(self) -> list[Campaign]:
statement = select(CampaignDB).order_by(CampaignDB.created_at.desc())
return [_campaign_from_db(r) for r in self.session.exec(statement).all()]
def get(self, campaign_id: str) -> Optional[Campaign]:
db = self.session.get(CampaignDB, campaign_id)
return _campaign_from_db(db) if db else None
def create(self, campaign: Campaign) -> Campaign:
db = _campaign_to_db(campaign)
self.session.add(db)
self.session.commit()
self.session.refresh(db)
return _campaign_from_db(db)
class ResultRepository:
"""Repository for evaluation results."""

View File

@ -13,7 +13,7 @@ from agenteval.storage.db import get_session, init_db
from agenteval.storage.repository import RunRepository
from agenteval.version import get_build_info, get_version
from agenteval.web.deps import require_api_key
from agenteval.web.routers import auth, files, model_configs, proxy, reports, runs, scenarios, stats, targets
from agenteval.web.routers import auth, campaigns, files, model_configs, proxy, reports, runs, scenarios, stats, targets
from agenteval.web.websocket import ws_manager
@ -58,6 +58,7 @@ app.include_router(auth.router, prefix="/api/auth", tags=["auth"])
app.include_router(targets.router, prefix="/api/targets", tags=["targets"], dependencies=_api_deps)
app.include_router(scenarios.router, prefix="/api/scenarios", tags=["scenarios"], dependencies=_api_deps)
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(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

@ -0,0 +1,63 @@
"""API routes for evaluation campaigns (评估活动).
This ticket covers persistence and create/query only no scheduling or child
Run spawning. Those arrive in later tickets.
"""
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field
from sqlmodel import Session
from agenteval.models import Campaign, CampaignPlanEntry
from agenteval.storage.repository import CampaignRepository, ScenarioRepository, TargetRepository
from agenteval.web.deps import get_db
router = APIRouter()
class CreateCampaignRequest(BaseModel):
name: str
target_id: str
window_seconds: int = Field(gt=0)
time_scale: float = Field(default=1.0, gt=0)
plan: list[CampaignPlanEntry] = Field(min_length=1)
@router.get("")
async def list_campaigns(session: Session = Depends(get_db)) -> list[dict]:
return [c.model_dump() for c in CampaignRepository(session).list_all()]
@router.post("")
async def create_campaign(
request: CreateCampaignRequest,
session: Session = Depends(get_db),
) -> dict:
if not TargetRepository(session).get(request.target_id):
raise HTTPException(status_code=404, detail="target not found")
scenario_repo = ScenarioRepository(session)
for entry in request.plan:
if not scenario_repo.get(entry.scenario_id):
raise HTTPException(
status_code=404,
detail=f"scenario not found: {entry.scenario_id}",
)
campaign = Campaign(
name=request.name,
target_id=request.target_id,
window_seconds=request.window_seconds,
time_scale=request.time_scale,
plan=request.plan,
)
campaign = CampaignRepository(session).create(campaign)
return campaign.model_dump()
@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:
raise HTTPException(status_code=404, detail="campaign not found")
return campaign.model_dump()

View File

@ -11,6 +11,7 @@ from alembic import context
# Import the DB module so its SQLModel table classes register in metadata.
from agenteval.storage.db import DATABASE_URL # noqa: F401
from agenteval.storage.db import ( # noqa: F401
CampaignDB,
EvalResultDB,
EvalRunDB,
EvalTargetDB,

View File

@ -0,0 +1,50 @@
"""add campaigns table and eval_runs.campaign_id
Revision ID: e6c3d1a2f809
Revises: d5b8c2e4f617
Create Date: 2026-07-29
"""
from typing import Sequence, Union
import sqlalchemy as sa
import sqlmodel # noqa: F401
from alembic import op
# revision identifiers, used by Alembic.
revision: str = "e6c3d1a2f809"
down_revision: Union[str, Sequence[str], None] = "d5b8c2e4f617"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
"campaigns",
sa.Column("id", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column("name", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column("target_id", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
sa.Column("window_seconds", sa.Integer(), nullable=False),
sa.Column("time_scale", sa.Float(), nullable=False),
sa.Column("plan", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column("status", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column("started_at", sa.DateTime(), nullable=True),
sa.Column("completed_at", sa.DateTime(), nullable=True),
sa.Column("created_at", sa.DateTime(), nullable=True),
sa.Column("summary", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
sa.ForeignKeyConstraint(["target_id"], ["eval_targets.id"]),
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"]
)
def downgrade() -> None:
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

@ -30,6 +30,7 @@ def db_session(tmp_db_path: Path) -> Session:
"""
# Import DB models so their table=True declarations register in metadata.
from agenteval.storage.db import ( # noqa: F401
CampaignDB,
EvalResultDB,
EvalRunDB,
EvalTargetDB,

View File

@ -0,0 +1,179 @@
"""Integration tests for the /api/campaigns endpoints.
Uses ``httpx.AsyncClient`` with ``app=`` to drive the FastAPI app in-process
(no real server). No scheduling or child-run spawning happens in this ticket
these tests cover persistence, create/query, validation, and that adding the
nullable ``campaign_id`` column does not disturb existing Run behavior.
"""
import pytest
from httpx import ASGITransport, AsyncClient
from agenteval.models import (
Case, CaseType, ChannelType, EvalRun, EvalTarget, PlatformType,
Scenario, TargetStatus,
)
from agenteval.storage.repository import RunRepository, ScenarioRepository, TargetRepository
from agenteval.web.app import app
@pytest.fixture()
def seeded_db(db_session, monkeypatch):
"""Patch the global get_session to the test session and seed target + scenario."""
from agenteval.storage import db as db_module
from agenteval.storage import repository as repo_module
from agenteval.web import app as app_module
monkeypatch.setattr(app_module, "init_db", lambda: None)
def _test_get_session():
return db_session
monkeypatch.setattr(db_module, "get_session", _test_get_session)
monkeypatch.setattr(repo_module, "get_session", _test_get_session)
from agenteval.web.deps import get_db
def _test_get_db():
try:
yield db_session
finally:
pass
app.dependency_overrides[get_db] = _test_get_db
target = EvalTarget(
id="t-1", name="mock-target",
platform=PlatformType.AI_DIGITAL_EMPLOYEE,
channel_type=ChannelType.TUTU_API,
channel_config={"base_url": "http://mock", "token": "x"},
status=TargetStatus.ACTIVE,
)
TargetRepository(db_session).create(target)
scenario = Scenario(
id="s-1", name="mock-scenario",
cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])],
)
ScenarioRepository(db_session).create(scenario)
yield db_session
app.dependency_overrides.clear()
@pytest.fixture()
async def client():
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as c:
yield c
def _valid_payload(**overrides) -> dict:
payload = {
"name": "24h-cycle",
"target_id": "t-1",
"window_seconds": 86400,
"time_scale": 1.0,
"plan": [
{"scenario_id": "s-1", "offset_seconds": 0, "count": 2},
{"scenario_id": "s-1", "offset_seconds": 3600, "count": 1},
],
}
payload.update(overrides)
return payload
# ── list / create / get ──────────────────────────────────────────────────
async def test_list_campaigns_empty(client, seeded_db):
resp = await client.get("/api/campaigns")
assert resp.status_code == 200
assert resp.json() == []
async def test_create_campaign_then_get(client, seeded_db):
resp = await client.post("/api/campaigns", json=_valid_payload())
assert resp.status_code == 200
body = resp.json()
campaign_id = body["id"]
assert body["status"] == "planned"
assert body["name"] == "24h-cycle"
assert body["target_id"] == "t-1"
assert body["window_seconds"] == 86400
assert body["time_scale"] == 1.0
assert len(body["plan"]) == 2
got = (await client.get(f"/api/campaigns/{campaign_id}")).json()
assert got["id"] == campaign_id
assert got["status"] == "planned"
assert got["plan"][0]["scenario_id"] == "s-1"
assert got["plan"][0]["offset_seconds"] == 0
assert got["plan"][0]["count"] == 2
assert got["plan"][1]["offset_seconds"] == 3600
async def test_create_campaign_default_time_scale(client, seeded_db):
payload = _valid_payload()
del payload["time_scale"]
resp = await client.post("/api/campaigns", json=payload)
assert resp.status_code == 200
assert resp.json()["time_scale"] == 1.0
async def test_list_campaigns_after_create(client, seeded_db):
await client.post("/api/campaigns", json=_valid_payload())
listing = (await client.get("/api/campaigns")).json()
assert len(listing) == 1
assert listing[0]["name"] == "24h-cycle"
async def test_get_campaign_not_found(client, seeded_db):
resp = await client.get("/api/campaigns/does-not-exist")
assert resp.status_code == 404
# ── validation ─────────────────────────────────────────────────────────────
async def test_create_campaign_empty_plan_rejected(client, seeded_db):
resp = await client.post("/api/campaigns", json=_valid_payload(plan=[]))
assert resp.status_code == 422
async def test_create_campaign_missing_target_404(client, seeded_db):
resp = await client.post("/api/campaigns", json=_valid_payload(target_id="nope"))
assert resp.status_code == 404
async def test_create_campaign_missing_scenario_404(client, seeded_db):
payload = _valid_payload(plan=[{"scenario_id": "nope", "offset_seconds": 0, "count": 1}])
resp = await client.post("/api/campaigns", json=payload)
assert resp.status_code == 404
async def test_create_campaign_bad_window_rejected(client, seeded_db):
resp = await client.post("/api/campaigns", json=_valid_payload(window_seconds=0))
assert resp.status_code == 422
async def test_create_campaign_bad_time_scale_rejected(client, seeded_db):
resp = await client.post("/api/campaigns", json=_valid_payload(time_scale=0))
assert resp.status_code == 422
# ── campaign_id column does not break existing Run behavior ─────────────────
async def test_run_campaign_id_defaults_none(seeded_db):
repo = RunRepository(seeded_db)
run = repo.create(EvalRun(target_id="t-1", scenario_id="s-1"))
assert run.campaign_id is None
fetched = repo.get(run.id)
assert fetched.campaign_id is None
async def test_run_can_belong_to_campaign(seeded_db):
repo = RunRepository(seeded_db)
run = repo.create(EvalRun(target_id="t-1", scenario_id="s-1", campaign_id="camp-1"))
assert run.campaign_id == "camp-1"
fetched = repo.get(run.id)
assert fetched.campaign_id == "camp-1"