From 38849d46f1e1b4d99f1fd6722357c479b9d5acf7 Mon Sep 17 00:00:00 2001 From: sinohqb Date: Tue, 4 Aug 2026 03:39:42 +0800 Subject: [PATCH] refactor(repository): narrow atomic updates for patrol/cancel/scheduler writes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 全行 update 逼着调用方承诺「持有最新完整快照」，巡检水位、活动取消、 调度循环三方并发写入会互相覆盖（后写者抹掉先写者的 summary/status）。 CampaignRepository 新增三个窄口径原子操作：touch_patrol_watermark、 mark_cancelled、save_scheduler_state，各自只写自己的列；三处调用点迁移。 --- .../agenteval/evaluation/campaign_runner.py | 2 +- backend/agenteval/storage/repository.py | 35 +++++- backend/agenteval/web/routers/campaigns.py | 4 +- backend/agenteval/web/routers/exploration.py | 3 +- tests/unit/test_campaign_narrow_updates.py | 102 ++++++++++++++++++ 5 files changed, 139 insertions(+), 7 deletions(-) create mode 100644 tests/unit/test_campaign_narrow_updates.py diff --git a/backend/agenteval/evaluation/campaign_runner.py b/backend/agenteval/evaluation/campaign_runner.py index 2339060..d99b2a8 100644 --- a/backend/agenteval/evaluation/campaign_runner.py +++ b/backend/agenteval/evaluation/campaign_runner.py @@ -180,7 +180,7 @@ async def advance_campaign( summary = campaign.summary or CampaignSummary() summary.scheduler = SchedulerState(spawned_indices=sorted(spawned), errors=errors) campaign.summary = summary - repo.update(campaign) + repo.save_scheduler_state(campaign_id, summary) return result diff --git a/backend/agenteval/storage/repository.py b/backend/agenteval/storage/repository.py index ed1fe83..a9fb23e 100644 --- a/backend/agenteval/storage/repository.py +++ b/backend/agenteval/storage/repository.py @@ -1,11 +1,12 @@ """Repository layer for database access.""" +from datetime import datetime from typing import Generic, Optional, TypeVar from sqlmodel import Session, select from agenteval.exploration.models import ExplorationMessage, ExplorationSession -from agenteval.models import Campaign, Case, EvalResult, EvalRun, EvalTarget, Scenario +from agenteval.models import Campaign, CampaignStatus, CampaignSummary, Case, EvalResult, EvalRun, EvalTarget, Scenario from agenteval.services.model_configs import ModelConfigService from agenteval.storage.db import ( CampaignAnalysisDB, @@ -411,6 +412,38 @@ class CampaignRepository(BaseRepository[Campaign, CampaignDB]): self.session.refresh(existing) return self._from_db(existing) + def touch_patrol_watermark(self, campaign_id: str, at: datetime) -> None: + """窄口径原子更新：只写巡检水位，不覆写并发的 status / summary 变更。""" + db = self.session.get(CampaignDB, campaign_id) + if not db: + return + db.last_patrolled_at = at + self.session.add(db) + self.session.commit() + self.session.refresh(db) + + def mark_cancelled(self, campaign_id: str, at: datetime) -> Optional[Campaign]: + """窄口径终态迁移：只写 status + completed_at，不抹掉水位与调度进度。""" + db = self.session.get(CampaignDB, campaign_id) + if not db: + return None + db.status = CampaignStatus.CANCELLED.value + db.completed_at = at + self.session.add(db) + self.session.commit() + self.session.refresh(db) + return self._from_db(db) + + def save_scheduler_state(self, campaign_id: str, summary: CampaignSummary) -> None: + """窄口径调度进度持久化：只写 summary 列，不覆写并发的状态 / 水位变更。""" + db = self.session.get(CampaignDB, campaign_id) + if not db: + return + db.set_summary(summary.model_dump(mode="json")) + self.session.add(db) + self.session.commit() + self.session.refresh(db) + class CampaignAnalysisRepository: """Repository for campaign analysis rows (one per campaign, upserted).""" diff --git a/backend/agenteval/web/routers/campaigns.py b/backend/agenteval/web/routers/campaigns.py index f60a3ce..e788004 100644 --- a/backend/agenteval/web/routers/campaigns.py +++ b/backend/agenteval/web/routers/campaigns.py @@ -117,9 +117,7 @@ async def cancel_campaign(campaign_id: str, session: Session = Depends(get_db)) if campaign.status not in (CampaignStatus.PLANNED, CampaignStatus.RUNNING): raise HTTPException(status_code=400, detail="campaign is not in a cancellable state") - campaign.status = CampaignStatus.CANCELLED - campaign.completed_at = utc_now() - repo.update(campaign) + campaign = repo.mark_cancelled(campaign_id, utc_now()) or campaign request_cancel(campaign_id) settle_campaign_sessions(campaign_id, session) return campaign.model_dump() diff --git a/backend/agenteval/web/routers/exploration.py b/backend/agenteval/web/routers/exploration.py index 1cc7b4d..f799ccb 100644 --- a/backend/agenteval/web/routers/exploration.py +++ b/backend/agenteval/web/routers/exploration.py @@ -185,8 +185,7 @@ async def patrol(session: Session = Depends(get_db)) -> dict: # 水位取构建响应之后的时刻：查询与持久化之间完成的结果不会在下次重复上报。 watermark_at = utc_now() for campaign in patrolled_campaigns: - campaign.last_patrolled_at = watermark_at - campaign_repo.update(campaign) + campaign_repo.touch_patrol_watermark(campaign.id, watermark_at) return {"patrolled_at": iso_utc(watermark_at), "campaigns": entries} diff --git a/tests/unit/test_campaign_narrow_updates.py b/tests/unit/test_campaign_narrow_updates.py new file mode 100644 index 0000000..3e97aab --- /dev/null +++ b/tests/unit/test_campaign_narrow_updates.py @@ -0,0 +1,102 @@ +"""CampaignRepository 窄口径原子更新直测（架构保养候选 1）。 + +全行 update 逼着调用方承诺「持有最新完整快照」，巡检水位 / 取消 / +调度循环并发写入会互相覆盖。三个窄方法各自只动自己的列： +touch_patrol_watermark（水位）、mark_cancelled（终态迁移）、 +save_scheduler_state（调度进度）。 +""" + +from datetime import datetime, timezone + +import pytest +from agenteval.models import Campaign, CampaignPlanEntry, CampaignStatus, CampaignSummary, SchedulerState +from agenteval.storage.db import as_utc +from agenteval.storage.repository import CampaignRepository +from sqlmodel import Session, SQLModel, create_engine + +T0 = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc) +T1 = datetime(2026, 1, 1, 1, 0, 0, tzinfo=timezone.utc) + + +@pytest.fixture() +def db_session(tmp_path): + from agenteval.storage.db import ( # noqa: F401 + CampaignDB, + EvalResultDB, + EvalRunDB, + EvalTargetDB, + FileCategoryDB, + FileRecordDB, + ScenarioDB, + TurnDB, + ) + engine = create_engine( + f"sqlite:///{tmp_path / 'narrow.db'}", + connect_args={"check_same_thread": False}, + ) + SQLModel.metadata.create_all(engine) + session = Session(engine) + try: + yield session + finally: + session.close() + engine.dispose() + + +def _seed(db_session, status=CampaignStatus.RUNNING) -> Campaign: + campaign = CampaignRepository(db_session).create(Campaign( + name="cycle", target_id="t-1", window_seconds=3600, + plan=[CampaignPlanEntry(scenario_id="s-a", offset_seconds=0, count=1)], + status=status, started_at=T0, + summary=CampaignSummary(scheduler=SchedulerState(spawned_indices=[0])), + )) + return campaign + + +def test_touch_patrol_watermark_only_touches_watermark(db_session): + campaign = _seed(db_session) + CampaignRepository(db_session).touch_patrol_watermark(campaign.id, T1) + + fresh = CampaignRepository(db_session).get(campaign.id) + assert as_utc(fresh.last_patrolled_at) == T1 + # 其余列原样：窄口径不覆写并发写入的 status / summary + assert fresh.status == CampaignStatus.RUNNING + assert fresh.summary.scheduler.spawned_indices == [0] + assert fresh.name == "cycle" and as_utc(fresh.started_at) == T0 + + +def test_mark_cancelled_only_touches_status_and_completed_at(db_session): + campaign = _seed(db_session) + CampaignRepository(db_session).touch_patrol_watermark(campaign.id, T1) + + updated = CampaignRepository(db_session).mark_cancelled(campaign.id, T1) + assert updated.status == CampaignStatus.CANCELLED + assert as_utc(updated.completed_at) == T1 + + fresh = CampaignRepository(db_session).get(campaign.id) + assert fresh.status == CampaignStatus.CANCELLED + assert as_utc(fresh.completed_at) == T1 + # 水位与调度进度不被终态迁移抹掉 + assert as_utc(fresh.last_patrolled_at) == T1 + assert fresh.summary.scheduler.spawned_indices == [0] + + +def test_save_scheduler_state_only_touches_summary(db_session): + campaign = _seed(db_session) + CampaignRepository(db_session).touch_patrol_watermark(campaign.id, T1) + + new_summary = CampaignSummary(scheduler=SchedulerState(spawned_indices=[0, 1, 2])) + CampaignRepository(db_session).save_scheduler_state(campaign.id, new_summary) + + fresh = CampaignRepository(db_session).get(campaign.id) + assert fresh.summary.scheduler.spawned_indices == [0, 1, 2] + # 状态与水位不被调度进度覆写 + assert fresh.status == CampaignStatus.RUNNING + assert as_utc(fresh.last_patrolled_at) == T1 + + +def test_narrow_updates_ignore_missing_campaign(db_session): + repo = CampaignRepository(db_session) + repo.touch_patrol_watermark("missing", T1) + repo.save_scheduler_state("missing", CampaignSummary()) + assert repo.mark_cancelled("missing", T1) is None