refactor(repository): narrow atomic updates for patrol/cancel/scheduler writes

全行 update 逼着调用方承诺「持有最新完整快照」，巡检水位、活动取消、
调度循环三方并发写入会互相覆盖（后写者抹掉先写者的 summary/status）。
CampaignRepository 新增三个窄口径原子操作：touch_patrol_watermark、
mark_cancelled、save_scheduler_state，各自只写自己的列；三处调用点迁移。

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
 

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)."""

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()

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}
 

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