diff --git a/backend/agenteval/evaluation/campaign_runner.py b/backend/agenteval/evaluation/campaign_runner.py index 13b9362..46f02f2 100644 --- a/backend/agenteval/evaluation/campaign_runner.py +++ b/backend/agenteval/evaluation/campaign_runner.py @@ -29,7 +29,15 @@ from agenteval.evaluation.campaign_scheduler import ( resolve_finalize, ) from agenteval.evaluation.engine import EvalEngine -from agenteval.models import Campaign, CampaignStatus, EvalRun, RunStatus, RunTrigger +from agenteval.models import ( + Campaign, + CampaignStatus, + CampaignSummary, + EvalRun, + RunStatus, + RunTrigger, + SchedulerState, +) from agenteval.storage.db import get_session, utc_now from agenteval.storage.repository import ( CampaignRepository, @@ -39,9 +47,6 @@ from agenteval.storage.repository import ( ) from agenteval.task_registry import TaskRegistry -_SPAWNED_KEY = "spawned_indices" -_ERRORS_KEY = "errors" - # Real wall-clock seconds between scheduler ticks. time_scale compresses the # *window*, not the tick cadence — production 24h campaigns still tick slowly. DEFAULT_TICK_SECONDS = 1.0 @@ -62,8 +67,9 @@ class AdvanceResult: def _spawned_indices(campaign: Campaign) -> set[int]: - scheduler = (campaign.summary or {}).get("scheduler", {}) - return set(scheduler.get(_SPAWNED_KEY, [])) + if campaign.summary is None: + return set() + return set(campaign.summary.scheduler.spawned_indices) def current_window_offset(campaign: Campaign) -> float: @@ -152,7 +158,7 @@ async def advance_campaign( ) result = AdvanceResult(finished=decision.finished) - errors: list[dict] = list((campaign.summary or {}).get("scheduler", {}).get(_ERRORS_KEY, [])) + errors: list[dict] = list(campaign.summary.scheduler.errors) if campaign.summary else [] for due in decision.due: # Cancellation stops further spawning; runs already in flight finish. if cancel_event is not None and cancel_event.is_set(): @@ -167,12 +173,10 @@ async def advance_campaign( spawned.add(due.index) # Persist progress per entry: a failure partway through a multi-entry # advance must never lose which entries already spawned, since restart - # recovery reads this back from the DB. - scheduler_state: dict = {_SPAWNED_KEY: sorted(spawned)} - if errors: - scheduler_state[_ERRORS_KEY] = errors - summary = dict(campaign.summary or {}) - summary["scheduler"] = scheduler_state + # recovery reads this back from the DB. Mutate the existing summary so + # any unknown top-level keys survive the read-modify-write. + summary = campaign.summary or CampaignSummary() + summary.scheduler = SchedulerState(spawned_indices=sorted(spawned), errors=errors) campaign.summary = summary repo.update(campaign) diff --git a/backend/agenteval/models.py b/backend/agenteval/models.py index 8c4b35a..8142ddd 100644 --- a/backend/agenteval/models.py +++ b/backend/agenteval/models.py @@ -246,10 +246,37 @@ class CampaignPlanEntry(BaseModel): count: int = Field(default=1, ge=1) +class SchedulerState(BaseModel): + """Durable scheduler progress for a campaign — restart-safe (ADR-0003). + + ``spawned_indices`` are the plan entries already派生 into child Runs; + ``errors`` records entries whose spawn failed (marked spawned to avoid + infinite retry). + """ + + spawned_indices: list[int] = Field(default_factory=list) + errors: list[dict[str, Any]] = Field(default_factory=list) + + +class CampaignSummary(BaseModel): + """Typed value of ``Campaign.summary`` — mirrors RunSummary's treatment. + + Unknown top-level keys are preserved (extra=allow) so summaries written by + older versions keep parsing and survive read-modify-write. + """ + + model_config = {"extra": "allow"} + + scheduler: SchedulerState = Field(default_factory=SchedulerState) + + class Campaign(BaseModel): """An evaluation campaign: a service-cycle window over a single target, driving many child Runs from a static plan (ADR-0003).""" + # summary 以属性赋值写入（scheduler loop）；赋值时即校验成 CampaignSummary + model_config = {"validate_assignment": True} + id: Optional[str] = None name: str target_id: str @@ -260,7 +287,7 @@ class Campaign(BaseModel): started_at: Optional[datetime] = None completed_at: Optional[datetime] = None created_at: Optional[datetime] = None - summary: Optional[dict[str, Any]] = None + summary: Optional[CampaignSummary] = None class Turn(BaseModel): diff --git a/backend/agenteval/storage/db.py b/backend/agenteval/storage/db.py index 91c2732..2090bd4 100644 --- a/backend/agenteval/storage/db.py +++ b/backend/agenteval/storage/db.py @@ -44,6 +44,16 @@ def new_uuid() -> str: return str(uuid.uuid4()) +def _json_dumps(value: Any) -> str: + """Serialize a JSON column value. ensure_ascii=False keeps CJK readable + in the stored text — the single serialization口径 for all JSON columns.""" + return json.dumps(value, ensure_ascii=False) + + +def _json_loads(raw: str) -> Any: + return json.loads(raw) + + class EvalTargetDB(SQLModel, table=True): """Database table for evaluation targets.""" @@ -65,10 +75,10 @@ class EvalTargetDB(SQLModel, table=True): ) def get_config(self) -> dict[str, Any]: - return json.loads(self.channel_config) + return _json_loads(self.channel_config) def set_config(self, config: dict[str, Any]) -> None: - self.channel_config = json.dumps(config, ensure_ascii=False) + self.channel_config = _json_dumps(config) class ScenarioDB(SQLModel, table=True): @@ -92,22 +102,22 @@ class ScenarioDB(SQLModel, table=True): ) def get_tags(self) -> list[str]: - return json.loads(self.tags) + return _json_loads(self.tags) def set_tags(self, tags: list[str]) -> None: - self.tags = json.dumps(tags, ensure_ascii=False) + self.tags = _json_dumps(tags) def get_cases(self) -> list[dict[str, Any]]: - return json.loads(self.cases) + return _json_loads(self.cases) def set_cases(self, cases: list[dict[str, Any]]) -> None: - self.cases = json.dumps(cases, ensure_ascii=False) + self.cases = _json_dumps(cases) def get_llm_config(self) -> Optional[dict[str, Any]]: - return json.loads(self.llm_config) if self.llm_config else None + return _json_loads(self.llm_config) if self.llm_config else None def set_llm_config(self, config: Optional[dict[str, Any]]) -> None: - self.llm_config = json.dumps(config, ensure_ascii=False) if config else None + self.llm_config = _json_dumps(config) if config else None class ModelConfigDB(SQLModel, table=True): @@ -140,14 +150,14 @@ class ModelConfigDB(SQLModel, table=True): updated_at: Optional[datetime] = Field(default_factory=utc_now) def get_input_modalities(self) -> list[str]: - return json.loads(self.input_modalities) + return _json_loads(self.input_modalities) def get_output_modalities(self) -> list[str]: - return json.loads(self.output_modalities) + return _json_loads(self.output_modalities) def set_modalities(self, input_modalities: list[str], output_modalities: list[str]) -> None: - self.input_modalities = json.dumps(input_modalities, ensure_ascii=True) - self.output_modalities = json.dumps(output_modalities, ensure_ascii=True) + self.input_modalities = _json_dumps(input_modalities) + self.output_modalities = _json_dumps(output_modalities) class ScenarioModelBindingDB(SQLModel, table=True): @@ -178,16 +188,16 @@ class CampaignDB(SQLModel, table=True): summary: Optional[str] = None def get_plan(self) -> list[dict[str, Any]]: - return json.loads(self.plan) + return _json_loads(self.plan) def set_plan(self, plan: list[dict[str, Any]]) -> None: - self.plan = json.dumps(plan, ensure_ascii=False) + self.plan = _json_dumps(plan) def get_summary(self) -> Optional[dict[str, Any]]: - return json.loads(self.summary) if self.summary else None + 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) + self.summary = _json_dumps(summary) class EvalRunDB(SQLModel, table=True): @@ -218,10 +228,10 @@ class EvalRunDB(SQLModel, table=True): ) def get_summary(self) -> Optional[dict[str, Any]]: - return json.loads(self.summary) if self.summary else None + 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) + self.summary = _json_dumps(summary) class TurnDB(SQLModel, table=True): @@ -243,16 +253,16 @@ class TurnDB(SQLModel, table=True): run: Optional[EvalRunDB] = Relationship(back_populates="turns") def get_sent_message(self) -> dict[str, Any]: - return json.loads(self.sent_message) + return _json_loads(self.sent_message) def set_sent_message(self, message: dict[str, Any]) -> None: - self.sent_message = json.dumps(message, ensure_ascii=False) + self.sent_message = _json_dumps(message) def get_reply(self) -> Optional[dict[str, Any]]: - return json.loads(self.reply) if self.reply else None + return _json_loads(self.reply) if self.reply else None def set_reply(self, reply: Optional[dict[str, Any]]) -> None: - self.reply = json.dumps(reply, ensure_ascii=False) if reply else None + self.reply = _json_dumps(reply) if reply else None class EvalResultDB(SQLModel, table=True): diff --git a/backend/agenteval/storage/repository.py b/backend/agenteval/storage/repository.py index 183d1cc..1626965 100644 --- a/backend/agenteval/storage/repository.py +++ b/backend/agenteval/storage/repository.py @@ -353,7 +353,7 @@ class CampaignRepository(BaseRepository[Campaign, CampaignDB]): ) db.set_plan([entry.model_dump(mode="json") for entry in campaign.plan]) if campaign.summary: - db.set_summary(campaign.summary) + db.set_summary(campaign.summary.model_dump(mode="json")) return db def _from_db(self, db: CampaignDB) -> Campaign: @@ -384,7 +384,7 @@ class CampaignRepository(BaseRepository[Campaign, CampaignDB]): existing.started_at = campaign.started_at existing.completed_at = campaign.completed_at if campaign.summary is not None: - existing.set_summary(campaign.summary) + existing.set_summary(campaign.summary.model_dump(mode="json")) self.session.add(existing) self.session.commit() self.session.refresh(existing) diff --git a/tests/unit/test_campaign_summary.py b/tests/unit/test_campaign_summary.py new file mode 100644 index 0000000..fdd1b47 --- /dev/null +++ b/tests/unit/test_campaign_summary.py @@ -0,0 +1,74 @@ +"""CampaignSummary VO — typed campaign scheduler state, restart-safe (ADR-0003). + +Mirrors RunSummary's typed treatment: Campaign.summary is no longer a bare +dict. Legacy dict summaries coerce; unknown top-level keys survive +(extra=allow) so older records keep parsing. +""" + +from agenteval.models import ( + Campaign, + CampaignPlanEntry, + CampaignSummary, + SchedulerState, +) +from agenteval.storage.repository import CampaignRepository, TargetRepository +from tests.unit.test_repository import _make_target + + +def _campaign(**summary_kw) -> Campaign: + kw = dict( + name="c", + target_id="t-1", + window_seconds=3600, + plan=[CampaignPlanEntry(scenario_id="s-1", offset_seconds=0, count=1)], + ) + kw.update(summary_kw) + return Campaign(**kw) + + +def test_summary_defaults_are_empty(): + s = CampaignSummary() + assert s.scheduler.spawned_indices == [] + assert s.scheduler.errors == [] + + +def test_scheduler_state_holds_progress(): + s = SchedulerState(spawned_indices=[0, 2], errors=[{"index": 1, "error": "boom"}]) + assert s.spawned_indices == [0, 2] + assert s.errors[0]["error"] == "boom" + + +def test_legacy_dict_summary_coerces(): + c = _campaign(summary={"scheduler": {"spawned_indices": [0, 1], "errors": [{"index": 2, "error": "x"}]}}) + assert isinstance(c.summary, CampaignSummary) + assert c.summary.scheduler.spawned_indices == [0, 1] + assert c.summary.scheduler.errors[0]["error"] == "x" + + +def test_assigning_dict_coerces_to_vo(): + c = _campaign() + c.summary = {"scheduler": {"spawned_indices": [3]}} + assert isinstance(c.summary, CampaignSummary) + assert c.summary.scheduler.spawned_indices == [3] + + +def test_unknown_top_level_keys_survive(): + c = _campaign(summary={"scheduler": {}, "future_axis": {"availability": 0.9}}) + assert isinstance(c.summary, CampaignSummary) + assert c.summary.model_extra["future_axis"] == {"availability": 0.9} + + +def test_summary_survives_repository_round_trip(db_session): + TargetRepository(db_session).create(_make_target()) + repo = CampaignRepository(db_session) + repo.create( + _campaign( + id="cp-1", + summary=CampaignSummary(scheduler=SchedulerState(spawned_indices=[0, 1], errors=[{"index": 2}])), + ) + ) + fetched = repo.get("cp-1") + assert fetched is not None + assert isinstance(fetched.summary, CampaignSummary) + assert fetched.summary.scheduler.spawned_indices == [0, 1] + assert fetched.summary.scheduler.errors == [{"index": 2}]