fix(architecture): enforce lifecycle consistency

backend/agenteval/channels/__init__.py

@@ -2,6 +2,7 @@
 
 from agenteval.channels.base import (
     ChannelHealth,
+    ChannelTransportError,
     EvalChannel,
     ExchangeOutcome,
     ExchangeStatus,
@@ -15,6 +16,7 @@ from agenteval.channels.tutu import TutuApiChannel
 __all__ = [
     "ChannelFactory",
     "ChannelHealth",
+    "ChannelTransportError",
     "EvalChannel",
     "ExchangeOutcome",
     "ExchangeStatus",

backend/agenteval/channels/base.py

@@ -41,6 +41,14 @@ class ExchangeStatus(str, Enum):
     POLL_FAILED = "poll_failed"
 
 
+class ChannelTransportError(Exception):
+    """Expected operational transport failure raised by a channel adapter.
+
+    Configuration and programming errors use their original exception types
+    and deliberately cross the exchange interface for diagnosis.
+    """
+
+
 def normalize_reply_text(content: Any) -> str:
     """Convert a channel reply payload into stable plain text.
 
@@ -192,7 +200,7 @@ class EvalChannel(ABC):
         started_at = time.monotonic()
         try:
             send_result = await self._send(content, **kwargs)
-        except Exception as exc:
+        except ChannelTransportError as exc:
             return ExchangeOutcome.send_failed(str(exc))
         if not send_result.ok:
             return ExchangeOutcome.send_failed(send_result.error or "channel send failed")
@@ -210,7 +218,7 @@ class EvalChannel(ABC):
                 timeout=timeout,
                 poll_interval=poll_interval,
             )
-        except Exception as exc:
+        except ChannelTransportError as exc:
             return ExchangeOutcome.poll_failed(
                 str(exc),
                 correlation_id=correlation_id,

backend/agenteval/channels/http.py

@@ -29,7 +29,7 @@ from typing import Any, Optional
 
 import httpx
 
-from agenteval.channels.base import ChannelHealth, EvalChannel, Reply, SendResult
+from agenteval.channels.base import ChannelHealth, ChannelTransportError, EvalChannel, Reply, SendResult
 
 
 def _get_path(data: Any, path: str) -> Any:
@@ -90,7 +90,7 @@ class HttpChannel(EvalChannel):
             data = resp.json()
             msg_id = str(_get_path(data, self.msg_id_path) or uuid.uuid4())
             return SendResult(ok=True, question_msg_id=msg_id, raw_response=data)
-        except Exception as exc:
+        except (httpx.HTTPError, json.JSONDecodeError) as exc:
             return SendResult(ok=False, error=str(exc))
 
     async def _poll_reply(
@@ -124,8 +124,8 @@ class HttpChannel(EvalChannel):
                         content=str(reply_text),
                         raw_message=raw,
                     )
-            except Exception:
+            except (httpx.HTTPError, json.JSONDecodeError) as exc:
-                pass
+                raise ChannelTransportError(str(exc)) from exc
 
             await asyncio.sleep(interval)
 

backend/agenteval/channels/openclaw.py

@@ -12,12 +12,13 @@ Configuration keys (in channel_config, all optional — defaults come from setti
 """
 
 import asyncio
+import json
 import uuid
 from typing import Any, Optional
 
 import httpx
 
-from agenteval.channels.base import ChannelHealth, EvalChannel, Reply, SendResult
+from agenteval.channels.base import ChannelHealth, ChannelTransportError, EvalChannel, Reply, SendResult
 from agenteval.config import get_settings
 
 
@@ -67,7 +68,7 @@ class OpenClawChannel(EvalChannel):
             # Extract the assistant message ID from the response
             msg_id = data.get("id") or str(uuid.uuid4())
             return SendResult(ok=True, question_msg_id=msg_id, raw_response=data)
-        except Exception as exc:
+        except (httpx.HTTPError, json.JSONDecodeError) as exc:
             return SendResult(ok=False, error=str(exc))
 
     async def _poll_reply(
@@ -100,8 +101,10 @@ class OpenClawChannel(EvalChannel):
                             content=reply_content,
                             raw_message={"text": reply_content, "_raw": data},
                         )
-            except Exception:
+                elif resp.status_code != 404:
-                pass
+                    raise ChannelTransportError(f"HTTP {resp.status_code}: {resp.text[:500]}")
+            except (httpx.HTTPError, json.JSONDecodeError) as exc:
+                raise ChannelTransportError(str(exc)) from exc
 
             await asyncio.sleep(interval)
 

backend/agenteval/channels/tutu.py

@@ -7,7 +7,7 @@ from typing import Any, Optional
 
 import httpx
 
-from agenteval.channels.base import ChannelHealth, EvalChannel, Reply, SendResult
+from agenteval.channels.base import ChannelHealth, ChannelTransportError, EvalChannel, Reply, SendResult
 
 
 class TutuApiChannel(EvalChannel):
@@ -87,7 +87,7 @@ class TutuApiChannel(EvalChannel):
             # The reply references the sent message via metadata.questionMsgId == sent msgId.
             question_msg_id = data.get("msgId")
             return SendResult(ok=True, question_msg_id=question_msg_id, raw_response=data)
-        except Exception as exc:
+        except (httpx.HTTPError, json.JSONDecodeError) as exc:
             return SendResult(ok=False, error=f"发送异常: {exc}")
 
     async def _poll_reply(
@@ -129,8 +129,8 @@ class TutuApiChannel(EvalChannel):
                                 msg_time=msg.get("msgTime"),
                                 raw_message=msg,
                             )
-            except Exception:
+            except (httpx.HTTPError, json.JSONDecodeError) as exc:
-                pass
+                raise ChannelTransportError(f"轮询异常: {exc}") from exc
 
             await asyncio.sleep(poll_interval)
 

backend/agenteval/evaluation/campaign_lifecycle.py

@@ -18,7 +18,7 @@ from agenteval.storage.repository import (
     CampaignAnalysisRepository,
     CampaignPeriodComparisonRepository,
     CampaignRepository,
-    ExplorationSessionRepository,
+    CampaignWriteStatus,
     RunRepository,
     ScenarioRepository,
     TargetRepository,
@@ -34,8 +34,8 @@ class CampaignCreateError(Exception):
 
 
 @dataclass(frozen=True)
-class CampaignCancelError(Exception):
+class CampaignLifecycleError(Exception):
-    """Cancellation failure translated by the HTTP adapter."""
+    """Campaign transition failure translated by the HTTP adapter."""
 
     status_code: int
     detail: str
@@ -67,8 +67,9 @@ def start_campaign(
     ):
         return None
     if campaign.status is CampaignStatus.PLANNED:
-        status, campaign = repo.start_if_planned(campaign_id, utc_now())
+        result = repo.start_if_planned(campaign_id, utc_now())
-        if status != "applied" or campaign is None:
+        campaign = result.campaign
+        if not result.applied or campaign is None:
             return None
     if launch is not None and campaign.id:
         launch(campaign.id, session)
@@ -138,13 +139,13 @@ def cancel_campaign(
 ) -> Campaign:
     """Cancel, settle exploration, then signal the process-local scheduler."""
     repo = CampaignRepository(session)
-    status, campaign = repo.cancel_if_active(campaign_id, utc_now())
+    result = repo.cancel_if_active(campaign_id, utc_now())
-    if status == "not_found":
+    campaign = result.campaign
-        raise CampaignCancelError(404, "campaign not found")
+    if result.status is CampaignWriteStatus.NOT_FOUND:
-    if status == "conflict" or campaign is None:
+        raise CampaignLifecycleError(404, "campaign not found")
-        raise CampaignCancelError(400, "campaign is not in a cancellable state")
+    if result.status is CampaignWriteStatus.CONFLICT or campaign is None:
+        raise CampaignLifecycleError(400, "campaign is not in a cancellable state")
 
-    ExplorationSessionRepository(session).expire_running_sessions(campaign_id)
     if stop is not None:
         stop(campaign_id)
     return campaign
@@ -178,23 +179,18 @@ def recover_campaign_runtime(session: Session, *, tick_seconds: float = 1.0) ->
 def complete_campaign(
     session: Session,
     campaign_id: str,
-    *,
-    settle: Optional[Callable[[str, Session], object]] = None,
 ) -> Campaign:
-    """Complete only a still-running Campaign, then settle exploration.
+    """Atomically complete a running Campaign and settle exploration.
 
     A competing cancellation wins because the status predicate is evaluated in
-    the database. Settlement is deliberately performed after the CAS commit;
+    the database. Settlement and the terminal status share one transaction, so
-    a failed settlement can be retried without reverting the terminal status.
+    a failure leaves the Campaign running for a later tick to retry.
     """
     repo = CampaignRepository(session)
-    status, campaign = repo.complete_if_running(campaign_id, utc_now())
+    result = repo.complete_if_running(campaign_id, utc_now())
-    if status == "not_found":
+    campaign = result.campaign
-        raise CampaignCancelError(404, "campaign not found")
+    if result.status is CampaignWriteStatus.NOT_FOUND:
-    if status == "conflict" or campaign is None:
+        raise CampaignLifecycleError(404, "campaign not found")
-        raise CampaignCancelError(409, "campaign is not running")
+    if result.status is CampaignWriteStatus.CONFLICT or campaign is None:
-    if settle is not None:
+        raise CampaignLifecycleError(409, "campaign is not running")
-        settle(campaign_id, session)
-    else:
-        ExplorationSessionRepository(session).expire_running_sessions(campaign_id)
     return campaign

backend/agenteval/evaluation/campaign_runner.py

@@ -21,7 +21,8 @@ from typing import Optional
 
 from sqlmodel import Session
 
-from agenteval.evaluation.analysis import enqueue_campaign_analysis, resolve_analysis_model
+from agenteval.evaluation.analysis import enqueue_campaign_analysis as start_campaign_analysis
+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 (
@@ -63,11 +64,6 @@ _logger = logging.getLogger("agenteval")
 campaign_registry = TaskRegistry()
 
 
-def start_campaign_analysis(campaign_id: str, *, triggered_by: str) -> None:
-    """Compatibility seam for the auto-analysis launcher."""
-    enqueue_campaign_analysis(campaign_id, triggered_by=triggered_by)
-
-
 @dataclass
 class AdvanceResult:
     """Outcome of advancing a campaign's clock once."""
@@ -85,20 +81,31 @@ class CampaignRecoveryResult:
 
 
 def _spawned_indices(campaign: Campaign, session: Optional[Session] = None) -> set[int]:
-    """Return plan entries with durable child identities, plus legacy progress."""
+    """Return plan entries completed by durable identities or legacy progress.
-    spawned = set(campaign.summary.scheduler.spawned_indices) if campaign.summary else set()
+
-    if campaign.id and session is not None:
+    Once any durable occurrence exists for an entry, identities replace the
+    legacy summary for that entry. This repairs summaries written by older
+    runners after only part of a multi-occurrence entry was claimed.
+    """
+    legacy = set(campaign.summary.scheduler.spawned_indices) if campaign.summary else set()
+    if not campaign.id or session is None:
+        return legacy
+
     occurrences: dict[int, set[int]] = {}
     for run in RunRepository(session).list_by_campaign(campaign.id):
         if run.campaign_plan_index is None or run.campaign_occurrence_index is None:
             continue
         occurrences.setdefault(run.campaign_plan_index, set()).add(run.campaign_occurrence_index)
-        spawned.update(
+
+    return {
         plan_index
         for plan_index, entry in enumerate(campaign.plan)
-            if set(range(entry.count)).issubset(occurrences.get(plan_index, set()))
+        if (
+            plan_index in occurrences
+            and set(range(entry.count)).issubset(occurrences[plan_index])
         )
-    return spawned
+        or (plan_index not in occurrences and plan_index in legacy)
+    }
 
 
 def current_window_offset(campaign: Campaign) -> float:
@@ -181,9 +188,8 @@ async def advance_campaign(
 
     ``elapsed_seconds`` is real wall-clock time since the window started; it is
     mapped to a window offset via ``time_scale``. A due entry whose spawning
-    raises is skipped and recorded (so one bad entry never wedges the loop),
+    raises is recorded but remains due until every occurrence has a durable
-    but still marked spawned to avoid infinite retries. Returns ``None`` if the
+    child-Run identity. Returns ``None`` if the campaign does not exist.
-    campaign does not exist.
     """
     repo = CampaignRepository(session)
     campaign = repo.get(campaign_id)
@@ -225,7 +231,6 @@ async def advance_campaign(
         if claim_rejected:
             break
         spawned = _spawned_indices(campaign, session)
-        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. Mutate the existing summary so
@@ -255,19 +260,6 @@ def _auto_start_analysis(campaign: Campaign, session: Session) -> None:
         _logger.warning("活动 %s 自动分析触发失败（已跳过）: %s", campaign.id, exc)
 
 
-def _settle_exploration(campaign_id: str, session: Session) -> None:
-    """Expire dangling exploration sessions on completion.
-
-    Never blocks completion: a settlement failure is logged and skipped, the
-    same non-blocking semantics as ``_auto_start_analysis``.
-    """
-    try:
-        from agenteval.storage.repository import ExplorationSessionRepository
-        ExplorationSessionRepository(session).expire_running_sessions(campaign_id)
-    except Exception as exc:
-        _logger.warning("活动 %s 探索会话结算失败（已跳过）: %s", campaign_id, exc)
-
-
 async def reconcile_campaign_child_runs(
     campaign_id: str,
     session: Session,

backend/agenteval/intelligent_eval/lifecycle.py

@@ -120,7 +120,7 @@ def create_eval(
         raise IntelligentEvalNotFoundError(f"target {target_id} not found")
 
     repo = IntelligentEvalRepository(session)
-    ev = repo.create(
+    ev = repo._create(
         IntelligentEval(
             name=name,
             target_id=target_id,
@@ -281,7 +281,7 @@ async def conduct_turn(session: Session, *, eval_id: str, session_id: str, conte
         outcome.latency_ms if outcome.latency_ms is not None else int((received_at - sent_at).total_seconds() * 1000)
     )
     reply_text = outcome.reply_text or ""
-    message_repo.create(
+    message_repo._create(
         IntelligentEvalMessage(
             session_id=obj.id,
             role="assistant",

backend/agenteval/intelligent_eval/read_model.py

@@ -164,14 +164,11 @@ class IntelligentEvalReadModel:
         sessions_by_eval = self._sessions.list_by_evals([_require_id(item) for item in evals])
         return [project_list_item(item, sessions_by_eval[item.id]) for item in evals if item.id is not None]
 
-    def detail(self, eval_obj: IntelligentEval) -> IntelligentEvalDetail:
-        return project_detail(eval_obj, self._sessions.list_by_eval(_require_id(eval_obj)))
-
     def detail_by_id(self, eval_id: str) -> IntelligentEvalDetail | None:
-        """Read one evaluation and its summaries from the same session snapshot."""
+        """Read one evaluation and its summaries with one snapshot query."""
 
-        eval_obj = self._evals.get(eval_id)
+        snapshot = self._evals.get_with_sessions(eval_id)
-        return self.detail(eval_obj) if eval_obj is not None else None
+        return project_detail(*snapshot) if snapshot is not None else None
 
     def sessions_by_eval(self, eval_id: str) -> list[IntelligentEvalSession] | None:
         """Return session summaries, or ``None`` when the parent is unknown."""

backend/agenteval/intelligent_eval/repository.py

@@ -105,7 +105,31 @@ class IntelligentEvalRepository:
         db = self.session.get(IntelligentEvalDB, eval_id)
         return self._from_db(db) if db else None
 
-    def create(self, obj: IntelligentEval) -> IntelligentEval:
+    def get_with_sessions(
+        self,
+        eval_id: str,
+    ) -> tuple[IntelligentEval, list[IntelligentEvalSession]] | None:
+        """Load an evaluation and all session summaries in one DB snapshot."""
+
+        statement = (
+            select(IntelligentEvalDB, IntelligentEvalSessionDB)
+            .join(
+                IntelligentEvalSessionDB,
+                IntelligentEvalSessionDB.eval_id == IntelligentEvalDB.id,
+                isouter=True,
+            )
+            .where(IntelligentEvalDB.id == eval_id)
+            .order_by(IntelligentEvalSessionDB.created_at.asc())
+        )
+        rows = self.session.exec(statement).all()
+        if not rows:
+            return None
+        evaluation = self._from_db(rows[0][0])
+        session_repo = IntelligentEvalSessionRepository(self.session)
+        sessions = [session_repo._from_db(session_db) for _, session_db in rows if session_db is not None]
+        return evaluation, sessions
+
+    def _create(self, obj: IntelligentEval) -> IntelligentEval:
         db = self._to_db(obj)
         self.session.add(db)
         self.session.commit()
@@ -210,15 +234,6 @@ class IntelligentEvalRepository:
             },
         )
 
-    def delete(self, eval_id: str) -> bool:
-        db = self.session.get(IntelligentEvalDB, eval_id)
-        if not db:
-            return False
-        self.session.delete(db)
-        self.session.commit()
-        return True
-
-
 class IntelligentEvalSessionRepository:
     """CRUD for intelligent eval sessions."""
 
@@ -272,7 +287,7 @@ class IntelligentEvalSessionRepository:
         db = self.session.get(IntelligentEvalSessionDB, session_id)
         return self._from_db(db) if db else None
 
-    def create(self, obj: IntelligentEvalSession) -> IntelligentEvalSession:
+    def _create(self, obj: IntelligentEvalSession) -> IntelligentEvalSession:
         db = IntelligentEvalSessionDB(
             id=obj.id,
             eval_id=obj.eval_id,
@@ -341,23 +356,6 @@ class IntelligentEvalSessionRepository:
             return CompareAndSetStatus.NOT_FOUND, None
         return CompareAndSetStatus.APPLIED, self._from_db(db)
 
-    def close(
-        self,
-        session_id: str,
-        verdict: dict,
-        status: IntelligentEvalSessionStatus = IntelligentEvalSessionStatus.COMPLETED,
-    ) -> Optional[IntelligentEvalSession]:
-        db = self.session.get(IntelligentEvalSessionDB, session_id)
-        if not db:
-            return None
-        db.status = status.value
-        db.set_verdict(verdict)
-        db.closed_at = utc_now()
-        self.session.add(db)
-        self.session.commit()
-        self.session.refresh(db)
-        return self._from_db(db)
-
     def _close_if_running(
         self,
         session_id: str,
@@ -394,14 +392,6 @@ class IntelligentEvalSessionRepository:
             return CompareAndSetStatus.NOT_FOUND, None
         return CompareAndSetStatus.APPLIED, self._from_db(db)
 
-    def increment_turns(self, session_id: str) -> None:
-        db = self.session.get(IntelligentEvalSessionDB, session_id)
-        if db:
-            db.turn_count += 1
-            self.session.add(db)
-            self.session.commit()
-
-
 class IntelligentEvalMessageRepository:
     """CRUD for intelligent eval session messages."""
 
@@ -426,7 +416,7 @@ class IntelligentEvalMessageRepository:
         )
         return [self._from_db(r) for r in self.session.exec(statement).all()]
 
-    def create(self, obj: IntelligentEvalMessage) -> IntelligentEvalMessage:
+    def _create(self, obj: IntelligentEvalMessage) -> IntelligentEvalMessage:
         db = IntelligentEvalMessageDB(
             id=obj.id,
             session_id=obj.session_id,

backend/agenteval/storage/repository.py

@@ -531,6 +531,26 @@ class RunRepository(BaseRepository[EvalRun, EvalRunDB]):
         return [_result_from_db(r) for r in self.session.exec(statement).all()]
 
 
+class CampaignWriteStatus(str, Enum):
+    """Outcome of a conditional Campaign lifecycle write."""
+
+    APPLIED = "applied"
+    NOT_FOUND = "not_found"
+    CONFLICT = "conflict"
+
+
+@dataclass(frozen=True)
+class CampaignWriteResult:
+    """Typed result returned by Campaign compare-and-set operations."""
+
+    status: CampaignWriteStatus
+    campaign: Optional[Campaign] = None
+
+    @property
+    def applied(self) -> bool:
+        return self.status is CampaignWriteStatus.APPLIED
+
+
 class CampaignRepository(BaseRepository[Campaign, CampaignDB]):
     """Repository for evaluation campaigns (评估活动)."""
 
@@ -626,56 +646,96 @@ class CampaignRepository(BaseRepository[Campaign, CampaignDB]):
         self.session.refresh(db)
         return self._from_db(db)
 
-    def cancel_if_active(self, campaign_id: str, at: datetime) -> tuple[str, Optional[Campaign]]:
+    def _compare_and_set_status(
-        """Compare-and-set a planned/running Campaign to cancelled."""
+        self,
+        campaign_id: str,
+        *,
+        expected_statuses: tuple[CampaignStatus, ...],
+        target_status: CampaignStatus,
+        at: datetime,
+        settle_exploration: bool = False,
+    ) -> CampaignWriteResult:
+        """Apply one conditional Campaign transition and optional settlement.
+
+        Final Campaign status and expiration of running exploration sessions
+        share one transaction. A settlement failure therefore rolls the
+        Campaign back to its prior runnable state.
+        """
+
+        values: dict[str, object] = {"status": target_status.value}
+        if target_status is CampaignStatus.RUNNING:
+            values["started_at"] = at
+        if target_status in (CampaignStatus.CANCELLED, CampaignStatus.COMPLETED, CampaignStatus.FAILED):
+            values["completed_at"] = at
+
         statement = (
             sql_update(CampaignDB)
             .where(
                 CampaignDB.id == campaign_id,
-                CampaignDB.status.in_([CampaignStatus.PLANNED.value, CampaignStatus.RUNNING.value]),
+                CampaignDB.status.in_([status.value for status in expected_statuses]),
             )
-            .values(status=CampaignStatus.CANCELLED.value, completed_at=at)
+            .values(**values)
         )
+        try:
             result = self.session.exec(statement)
-        if result.rowcount == 1:
+            if result.rowcount != 1:
-            self.session.commit()
-            db = self.session.get(CampaignDB, campaign_id)
-            return "applied", self._from_db(db) if db else None
                 self.session.rollback()
                 db = self.session.get(CampaignDB, campaign_id)
-        return ("not_found", None) if db is None else ("conflict", self._from_db(db))
+                status = CampaignWriteStatus.NOT_FOUND if db is None else CampaignWriteStatus.CONFLICT
+                return CampaignWriteResult(status=status, campaign=self._from_db(db) if db else None)
 
-    def complete_if_running(self, campaign_id: str, at: datetime) -> tuple[str, Optional[Campaign]]:
+            if settle_exploration:
-        """Compare-and-set a running Campaign to completed."""
+                self.session.exec(
-        statement = (
+                    sql_update(ExplorationSessionDB)
-            sql_update(CampaignDB)
+                    .where(
-            .where(CampaignDB.id == campaign_id, CampaignDB.status == CampaignStatus.RUNNING.value)
+                        ExplorationSessionDB.campaign_id == campaign_id,
-            .values(status=CampaignStatus.COMPLETED.value, completed_at=at)
+                        ExplorationSessionDB.status == ExplorationSessionStatus.RUNNING.value,
+                    )
+                    .values(status=ExplorationSessionStatus.EXPIRED.value, closed_at=at)
                 )
-        result = self.session.exec(statement)
-        if result.rowcount == 1:
             self.session.commit()
-            db = self.session.get(CampaignDB, campaign_id)
+        except Exception:
-            return "applied", self._from_db(db) if db else None
             self.session.rollback()
-        db = self.session.get(CampaignDB, campaign_id)
+            raise
-        return ("not_found", None) if db is None else ("conflict", self._from_db(db))
 
-    def start_if_planned(self, campaign_id: str, at: datetime) -> tuple[str, Optional[Campaign]]:
+        self.session.expire_all()
+        db = self.session.get(CampaignDB, campaign_id)
+        return CampaignWriteResult(
+            status=CampaignWriteStatus.APPLIED,
+            campaign=self._from_db(db) if db else None,
+        )
+
+    def cancel_if_active(self, campaign_id: str, at: datetime) -> CampaignWriteResult:
+        """Atomically cancel and expire running exploration sessions."""
+
+        return self._compare_and_set_status(
+            campaign_id,
+            expected_statuses=(CampaignStatus.PLANNED, CampaignStatus.RUNNING),
+            target_status=CampaignStatus.CANCELLED,
+            at=at,
+            settle_exploration=True,
+        )
+
+    def complete_if_running(self, campaign_id: str, at: datetime) -> CampaignWriteResult:
+        """Atomically complete and expire running exploration sessions."""
+
+        return self._compare_and_set_status(
+            campaign_id,
+            expected_statuses=(CampaignStatus.RUNNING,),
+            target_status=CampaignStatus.COMPLETED,
+            at=at,
+            settle_exploration=True,
+        )
+
+    def start_if_planned(self, campaign_id: str, at: datetime) -> CampaignWriteResult:
         """Atomically stamp a planned Campaign as running."""
-        statement = (
+
-            sql_update(CampaignDB)
+        return self._compare_and_set_status(
-            .where(CampaignDB.id == campaign_id, CampaignDB.status == CampaignStatus.PLANNED.value)
+            campaign_id,
-            .values(status=CampaignStatus.RUNNING.value, started_at=at)
+            expected_statuses=(CampaignStatus.PLANNED,),
+            target_status=CampaignStatus.RUNNING,
+            at=at,
         )
-        result = self.session.exec(statement)
-        if result.rowcount == 1:
-            self.session.commit()
-            db = self.session.get(CampaignDB, campaign_id)
-            return "applied", self._from_db(db) if db else None
-        self.session.rollback()
-        db = self.session.get(CampaignDB, campaign_id)
-        return ("not_found", None) if db is None else ("conflict", self._from_db(db))
 
     def save_scheduler_state(self, campaign_id: str, summary: CampaignSummary) -> None:
         """窄口径调度进度持久化：只写 summary 列，不覆写并发的状态 / 水位变更。"""

backend/agenteval/web/routers/campaigns.py

@@ -11,8 +11,9 @@ 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 enqueue_campaign_analysis as start_campaign_analysis
-from agenteval.evaluation.campaign_lifecycle import CampaignCancelError, CampaignCreateError
+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
@@ -44,11 +45,6 @@ from agenteval.web.deps import get_db
 router = APIRouter()
 
 
-def start_campaign_analysis(campaign_id: str, *, triggered_by: str) -> None:
-    """Compatibility seam for tests and adapter-level task injection."""
-    enqueue_campaign_analysis(campaign_id, triggered_by=triggered_by)
-
-
 class CreateCampaignRequest(BaseModel):
     name: str
     target_id: str
@@ -100,7 +96,7 @@ async def create_campaign(
 async def cancel_campaign(campaign_id: str, session: Session = Depends(get_db)) -> dict:
     try:
         campaign = cancel_campaign_lifecycle(session, campaign_id, stop=request_cancel)
-    except CampaignCancelError as exc:
+    except CampaignLifecycleError as exc:
         raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc
     return campaign.model_dump()
 

backend/agenteval/web/routers/intelligent_evals.py

@@ -68,11 +68,10 @@ def _translate(exc: Exception) -> HTTPException:
     return HTTPException(status_code=409, detail=exc.reason)
 
 
-def _eval_response(ev, session: Session, *, detail: bool = False) -> dict:
+def _eval_response(ev, session: Session) -> dict:
     """Serialize one stable intelligent-evaluation read projection."""
 
-    reader = IntelligentEvalReadModel(session)
+    projection = IntelligentEvalReadModel(session).list_item(ev)
-    projection = reader.detail(ev) if detail else reader.list_item(ev)
     return projection.model_dump(mode="json")
 
 

docs/adr/0003-campaign-phased-static-then-adaptive.md

@@ -50,7 +50,7 @@
 - **子 Run 身份持久化**：新建 child Run 保存 `campaign_id`、计划条目索引和 occurrence 索引；三者由 Campaign 范围内唯一索引约束。历史 Run 保持空值，不回填。
 - **Claim 先于执行**：调度器对每个 occurrence 先执行条件 claim，再进入 `EvalEngine`。重复 tick、并发调度或重启不会创建第二个 Run；取消后的 Campaign 不允许新 claim，已启动 Run 可继续完成。
 - **恢复边界**：持久化为 `pending` 且仍属于 running Campaign 的子 Run 可以恢复；进程中断遗留的 `running` 子 Run 统一标记为 `failed/interrupted`，禁止重放可能已经发送的外部消息。
-- **生命周期 CAS**：创建、启动、取消和完成均通过生命周期模块与条件更新完成。取消竞态优先于完成；完成提交后再结算 running 探索会话，结算失败不回滚活动终态，可由后续恢复/重试处理。
+- **生命周期 CAS**：创建、启动、取消和完成均通过生命周期模块与条件更新完成。取消竞态优先于完成；Campaign 终态与 running 探索会话结算在同一事务提交，任一步失败均保持活动与探索会话为 `running`，由后续调度 tick 重试。
 - **分析任务耐久化**：活动分析先写入 `queued` 再启动进程内 worker；重启恢复 queued 任务，遗留 `generating` 任务标记为中断失败。分析失败或重复执行不改变 Campaign 完成状态。
 
 启动恢复顺序固定为：清理中断 Run 与 LLM 任务 → 重建 running Campaign 调度循环（其中包含 child Run reconciliation）→ 重启 queued 分析任务。该顺序保证恢复动作只依据已提交的数据库事实，不依赖上一次进程的内存状态。

frontend/web/src/read/useIntelligentEvalRead.ts

@@ -6,6 +6,7 @@ import {
   type IntelligentEvalReadAdapter,
   type IntelligentEvalReadState,
 } from './intelligentEval'
+import { usePolling } from '../hooks/usePolling'
 
 const ACTIVE_STATUSES = new Set(['planning', 'pending_approval', 'executing'])
 
@@ -55,17 +56,14 @@ export function useIntelligentEvalRead(
   const listActive = state.list.value.some((item) => isActive(item.status))
   const detailActive = isActive(state.detail.value?.status)
 
-  useEffect(() => {
+  usePolling(() => { void loadList(true) }, 5000, listActive)
-    if (!listActive) return
+  usePolling(
-    const timer = window.setInterval(() => { void loadList(true) }, 5000)
+    () => {
-    return () => window.clearInterval(timer)
+      if (selectedId != null) void loadDetail(selectedId, true)
-  }, [listActive, loadList])
+    },
-
+    5000,
-  useEffect(() => {
+    selectedId != null && detailActive,
-    if (selectedId == null || !detailActive) return
+  )
-    const timer = window.setInterval(() => { void loadDetail(selectedId, true) }, 5000)
-    return () => window.clearInterval(timer)
-  }, [detailActive, loadDetail, selectedId])
 
   const reloadList = useCallback(() => loadList(true), [loadList])
   const reloadDetail = useCallback(

tests/integration/test_campaign_runner.py

@@ -12,6 +12,7 @@ from agenteval.evaluation.campaign_runner import advance_campaign, reconcile_cam
 from agenteval.models import (
     Campaign,
     CampaignPlanEntry,
+    CampaignSummary,
     Case,
     CaseType,
     ChannelType,
@@ -20,6 +21,7 @@ from agenteval.models import (
     RunStatus,
     RunTrigger,
     Scenario,
+    SchedulerState,
     TargetStatus,
 )
 from agenteval.storage.repository import CampaignRepository, RunRepository, ScenarioRepository, TargetRepository
@@ -171,6 +173,8 @@ async def test_partial_claim_does_not_hide_remaining_occurrences(seeded_db):
         occurrence_index=0,
     )
     assert first_claim.run is not None
+    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)
 
@@ -180,6 +184,41 @@ async def test_partial_claim_does_not_hide_remaining_occurrences(seeded_db):
     assert all(run.status is RunStatus.COMPLETED for run in runs)
 
 
+async def test_partial_spawn_failure_is_retried_until_all_occurrences_are_claimed(seeded_db, monkeypatch):
+    from agenteval.evaluation import campaign_runner
+
+    campaign = CampaignRepository(seeded_db).create(
+        Campaign(
+            name="partial-failure",
+            target_id="t-1",
+            status="running",
+            window_seconds=60,
+            plan=[CampaignPlanEntry(scenario_id="s-1", offset_seconds=0, count=2)],
+        )
+    )
+    real_spawn = campaign_runner._spawn_child_run
+    failed_once = False
+
+    async def flaky_spawn(*args, occurrence_index, **kwargs):
+        nonlocal failed_once
+        if occurrence_index == 1 and not failed_once:
+            failed_once = True
+            raise RuntimeError("transient spawn failure")
+        return await real_spawn(*args, occurrence_index=occurrence_index, **kwargs)
+
+    monkeypatch.setattr(campaign_runner, "_spawn_child_run", flaky_spawn)
+
+    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)
+    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}
+    assert recovered.summary.scheduler.spawned_indices == [0]
+
+
 async def test_recovery_resumes_pending_claim_without_replacing_identity(seeded_db):
     campaign = _make_campaign(seeded_db)
     claim = RunRepository(seeded_db).claim_campaign_run(

tests/integration/test_campaign_scheduler_loop.py

@@ -98,6 +98,26 @@ 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):
+    campaign = _make_campaign(seeded_db)
+    real_complete = campaign_runner.complete_campaign
+    attempts = 0
+
+    def flaky_complete(*args, **kwargs):
+        nonlocal attempts
+        attempts += 1
+        if attempts == 1:
+            raise RuntimeError("settlement failed")
+        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 attempts == 2
+    assert CampaignRepository(seeded_db).get(campaign.id).status is CampaignStatus.COMPLETED
+
+
 async def test_restart_recovery_does_not_respawn(seeded_db):
     # 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.

tests/unit/test_campaign_cancel_lifecycle.py

@@ -1,9 +1,11 @@
 """Tests for atomic Campaign cancellation and settlement ordering."""
 
 import pytest
-from agenteval.evaluation.campaign_lifecycle import CampaignCancelError, cancel_campaign
+from agenteval.evaluation.campaign_lifecycle import CampaignLifecycleError, cancel_campaign
+from agenteval.exploration.models import ExplorationSession, ExplorationSessionStatus
 from agenteval.models import Campaign, CampaignPlanEntry, CampaignStatus
-from agenteval.storage.repository import CampaignRepository
+from agenteval.storage.repository import CampaignRepository, ExplorationSessionRepository
+from sqlalchemy import event
 
 
 def _campaign(session, status=CampaignStatus.RUNNING):
@@ -34,12 +36,12 @@ def test_cancel_is_atomic_and_stop_happens_after_durable_state(db_session):
 
 
 def test_cancel_distinguishes_missing_and_terminal_campaign(db_session):
-    with pytest.raises(CampaignCancelError) as missing:
+    with pytest.raises(CampaignLifecycleError) as missing:
         cancel_campaign(db_session, "missing")
     assert missing.value.status_code == 404
 
     _campaign(db_session, CampaignStatus.COMPLETED)
-    with pytest.raises(CampaignCancelError) as terminal:
+    with pytest.raises(CampaignLifecycleError) as terminal:
         cancel_campaign(db_session, "campaign-1")
     assert terminal.value.status_code == 400
 
@@ -47,6 +49,29 @@ def test_cancel_distinguishes_missing_and_terminal_campaign(db_session):
 def test_cancel_is_idempotency_guarded_by_status(db_session):
     _campaign(db_session)
     cancel_campaign(db_session, "campaign-1")
-    with pytest.raises(CampaignCancelError) as again:
+    with pytest.raises(CampaignLifecycleError) as again:
         cancel_campaign(db_session, "campaign-1")
     assert again.value.status_code == 400
+
+
+def test_cancel_settlement_failure_rolls_back_campaign_and_session(db_session):
+    _campaign(db_session)
+    session_obj = ExplorationSessionRepository(db_session).create(
+        ExplorationSession(campaign_id="campaign-1", target_id="t-1")
+    )
+    engine = db_session.get_bind()
+
+    def fail_exploration_update(_connection, _cursor, statement, _parameters, _context, _executemany):
+        if statement.lstrip().upper().startswith("UPDATE EXPLORATION_SESSIONS"):
+            raise RuntimeError("settlement failed")
+
+    event.listen(engine, "before_cursor_execute", fail_exploration_update)
+    try:
+        with pytest.raises(RuntimeError, match="settlement failed"):
+            cancel_campaign(db_session, "campaign-1")
+    finally:
+        event.remove(engine, "before_cursor_execute", fail_exploration_update)
+        db_session.rollback()
+
+    assert CampaignRepository(db_session).get("campaign-1").status is CampaignStatus.RUNNING
+    assert ExplorationSessionRepository(db_session).get(session_obj.id).status is ExplorationSessionStatus.RUNNING

tests/unit/test_campaign_complete_lifecycle.py

@@ -1,9 +1,11 @@
 """Tests for idempotent Campaign completion and settlement ordering."""
 
 import pytest
-from agenteval.evaluation.campaign_lifecycle import CampaignCancelError, complete_campaign
+from agenteval.evaluation.campaign_lifecycle import CampaignLifecycleError, complete_campaign
+from agenteval.exploration.models import ExplorationSession, ExplorationSessionStatus
 from agenteval.models import Campaign, CampaignPlanEntry, CampaignStatus
-from agenteval.storage.repository import CampaignRepository
+from agenteval.storage.repository import CampaignRepository, ExplorationSessionRepository
+from sqlalchemy import event
 
 
 def _campaign(session, status=CampaignStatus.RUNNING):
@@ -19,30 +21,52 @@ def _campaign(session, status=CampaignStatus.RUNNING):
     )
 
 
-def test_complete_uses_cas_and_settles_after_commit(db_session):
+def test_complete_atomically_settles_running_exploration(db_session):
     _campaign(db_session)
-    observed = []
+    session_obj = ExplorationSessionRepository(db_session).create(
+        ExplorationSession(campaign_id="campaign-1", target_id="t-1")
+    )
 
-    def settle(campaign_id, session):
+    completed = complete_campaign(db_session, "campaign-1")
-        observed.append(CampaignRepository(session).get(campaign_id).status)
-
-    completed = complete_campaign(db_session, "campaign-1", settle=settle)
 
     assert completed.status is CampaignStatus.COMPLETED
     assert completed.completed_at is not None
-    assert observed == [CampaignStatus.COMPLETED]
+    assert ExplorationSessionRepository(db_session).get(session_obj.id).status is ExplorationSessionStatus.EXPIRED
 
 
 def test_complete_is_idempotency_guarded(db_session):
     _campaign(db_session)
     complete_campaign(db_session, "campaign-1")
-    with pytest.raises(CampaignCancelError) as again:
+    with pytest.raises(CampaignLifecycleError) as again:
         complete_campaign(db_session, "campaign-1")
     assert again.value.status_code == 409
 
 
 def test_cancel_wins_completion_race(db_session):
     _campaign(db_session, CampaignStatus.CANCELLED)
-    with pytest.raises(CampaignCancelError) as conflict:
+    with pytest.raises(CampaignLifecycleError) as conflict:
         complete_campaign(db_session, "campaign-1")
     assert conflict.value.status_code == 409
+
+
+def test_settlement_failure_keeps_campaign_running_and_session_open(db_session):
+    _campaign(db_session)
+    session_obj = ExplorationSessionRepository(db_session).create(
+        ExplorationSession(campaign_id="campaign-1", target_id="t-1")
+    )
+    engine = db_session.get_bind()
+
+    def fail_exploration_update(_connection, _cursor, statement, _parameters, _context, _executemany):
+        if statement.lstrip().upper().startswith("UPDATE EXPLORATION_SESSIONS"):
+            raise RuntimeError("settlement failed")
+
+    event.listen(engine, "before_cursor_execute", fail_exploration_update)
+    try:
+        with pytest.raises(RuntimeError, match="settlement failed"):
+            complete_campaign(db_session, "campaign-1")
+    finally:
+        event.remove(engine, "before_cursor_execute", fail_exploration_update)
+        db_session.rollback()
+
+    assert CampaignRepository(db_session).get("campaign-1").status is CampaignStatus.RUNNING
+    assert ExplorationSessionRepository(db_session).get(session_obj.id).status is ExplorationSessionStatus.RUNNING

tests/unit/test_channel_contract.py

@@ -3,6 +3,7 @@
 import pytest
 from agenteval.channels.base import (
     ChannelHealth,
+    ChannelTransportError,
     EvalChannel,
     ExchangeOutcome,
     ExchangeStatus,
@@ -133,7 +134,14 @@ async def test_exchange_distinguishes_timeout_and_poll_failure():
     assert timeout.status is ExchangeStatus.REPLY_TIMEOUT
     assert timeout.correlation_id == "question-7"
 
-    failed_channel = ContractChannel(poll_error=RuntimeError("upstream unavailable"))
+    failed_channel = ContractChannel(poll_error=ChannelTransportError("upstream unavailable"))
     failed = await failed_channel.exchange("问题")
     assert failed.status is ExchangeStatus.POLL_FAILED
     assert failed.reason == "upstream unavailable"
+
+
+async def test_exchange_does_not_mask_programming_or_configuration_errors():
+    channel = ContractChannel(poll_error=ValueError("invalid adapter configuration"))
+
+    with pytest.raises(ValueError, match="invalid adapter configuration"):
+        await channel.exchange("问题")

tests/unit/test_engine.py

@@ -7,7 +7,7 @@ timeouts, and concurrent case execution via the semaphore.
 import asyncio
 from typing import Any
 
-from agenteval.channels.base import SendResult
+from agenteval.channels.base import ChannelTransportError, SendResult
 from agenteval.evaluation.engine import EvalEngine, TimeoutConfig
 from agenteval.models import (
     Case,
@@ -288,7 +288,7 @@ async def test_poll_failure_keeps_sent_turn_and_fails_case(db_session):
         name="poll-failure",
         cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])],
     )
-    channel = MockChannel(raise_on_poll=RuntimeError("upstream unavailable"))
+    channel = MockChannel(raise_on_poll=ChannelTransportError("upstream unavailable"))
     engine = _build_engine(scenario, channel, session=db_session)
 
     run = await engine.run()

tests/unit/test_http_channel_and_rules.py

@@ -2,6 +2,9 @@
 
 from unittest.mock import AsyncMock, MagicMock, patch
 
+import httpx
+import pytest
+from agenteval.channels.base import ExchangeStatus
 from agenteval.channels.http import HttpChannel, _get_path
 from agenteval.channels.openclaw import OpenClawChannel
 from agenteval.evaluation.rules.keyword import KeywordMatchRule
@@ -74,12 +77,34 @@ async def test_http_send_extracts_msg_id():
 
 async def test_http_send_failure():
     ch = _make_channel()
-    with patch.object(ch._client, "post", new=AsyncMock(side_effect=Exception("timeout"))):
+    with patch.object(ch._client, "post", new=AsyncMock(side_effect=httpx.ConnectError("timeout"))):
         result = await ch._send("hi")
     assert result.ok is False
     assert "timeout" in result.error
 
 
+async def test_http_exchange_classifies_poll_transport_failure():
+    ch = _make_channel()
+    sent = MagicMock()
+    sent.raise_for_status = MagicMock()
+    sent.json = MagicMock(return_value={"id": "msg-1"})
+    with (
+        patch.object(ch._client, "post", new=AsyncMock(return_value=sent)),
+        patch.object(ch._client, "get", new=AsyncMock(side_effect=httpx.ConnectError("offline"))),
+    ):
+        outcome = await ch.exchange("hi")
+
+    assert outcome.status is ExchangeStatus.POLL_FAILED
+    assert "offline" in outcome.reason
+
+
+async def test_http_exchange_does_not_mask_adapter_programming_errors():
+    ch = _make_channel()
+    with patch.object(ch._client, "post", new=AsyncMock(side_effect=RuntimeError("adapter bug"))):
+        with pytest.raises(RuntimeError, match="adapter bug"):
+            await ch.exchange("hi")
+
+
 async def test_http_poll_reply_found():
     ch = _make_channel(reply_path="answer")
     call_count = {"n": 0}
@@ -250,7 +275,7 @@ async def test_openclaw_send_ok():
 
 async def test_openclaw_send_failure():
     ch = _make_openclaw_channel()
-    with patch.object(ch._client, "post", new=AsyncMock(side_effect=Exception("timeout"))):
+    with patch.object(ch._client, "post", new=AsyncMock(side_effect=httpx.ConnectError("timeout"))):
         result = await ch._send("hi")
     assert result.ok is False
 

tests/unit/test_intelligent_eval_model.py

@@ -1,315 +1,167 @@
-"""Smoke tests for intelligent eval data model (ticket 01)."""
+"""Lifecycle contract tests for intelligent evaluations."""
 
 import pytest
-from agenteval.intelligent_eval.models import (
+from agenteval.channels.base import ExchangeOutcome, SendResult
-    IntelligentEval,
+from agenteval.intelligent_eval import lifecycle
-    IntelligentEvalMessage,
+from agenteval.intelligent_eval.lifecycle import (
-    IntelligentEvalSession,
+    IntelligentEvalChannelError,
-    IntelligentEvalSessionStatus,
+    IntelligentEvalNotFoundError,
-    IntelligentEvalStatus,
+    IntelligentEvalTransitionError,
 )
-from agenteval.intelligent_eval.repository import (
+from agenteval.intelligent_eval.models import IntelligentEvalSessionStatus, IntelligentEvalStatus
-    CompareAndSetStatus,
+from agenteval.models import ChannelType, EvalTarget, PlatformType, TargetStatus
-    IntelligentEvalMessageRepository,
+from agenteval.storage.repository import TargetRepository
-    IntelligentEvalRepository,
-    IntelligentEvalSessionRepository,
-)
-from agenteval.storage.db import EvalTargetDB
-from sqlalchemy.pool import StaticPool
-from sqlmodel import Session, SQLModel, create_engine
 
 
-@pytest.fixture
+@pytest.fixture()
-def db_session():
+def eval_session(db_session):
-    engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+    TargetRepository(db_session).create(
-    SQLModel.metadata.create_all(engine)
+        EvalTarget(
-    session = Session(engine)
+            id="target-1",
-    target = EvalTargetDB(id="t1", name="test-target")
+            name="测试对象",
-    session.add(target)
+            platform=PlatformType.AI_DIGITAL_EMPLOYEE,
-    session.commit()
+            channel_type=ChannelType.TUTU_API,
-    yield session
+            channel_config={"base_url": "http://mock", "token": "token"},
-    session.close()
+            status=TargetStatus.ACTIVE,
-
-
-class TestIntelligentEvalRepository:
-    def test_create_and_get(self, db_session):
-        repo = IntelligentEvalRepository(db_session)
-        ev = repo.create(
-            IntelligentEval(
-                name="test-eval",
-                target_id="t1",
-                goal="evaluate customer service",
-                seeds={"personas": [], "goals": []},
-                intent="test intent",
-                role_description="impatient customer",
-                time_window_hours=24,
         )
     )
-        assert ev.id is not None
+    return db_session
-        assert ev.status == IntelligentEvalStatus.DRAFT
-        assert ev.name == "test-eval"
-        assert ev.time_window_hours == 24
 
-        fetched = repo.get(ev.id)
-        assert fetched is not None
-        assert fetched.goal == "evaluate customer service"
-        assert fetched.seeds == {"personas": [], "goals": []}
 
-    def test_list_all(self, db_session):
+def _create(session, *, name: str = "智能评估"):
-        repo = IntelligentEvalRepository(db_session)
+    return lifecycle.create_eval(
-        repo.create(IntelligentEval(name="eval-1", target_id="t1"))
+        session,
-        repo.create(IntelligentEval(name="eval-2", target_id="t1"))
+        name=name,
-        all_evals = repo.list_all()
+        target_id="target-1",
-        assert len(all_evals) == 2
+        goal="验证退货流程",
-
+        seeds={"personas": ["老客户"]},
-    def test_status_transitions(self, db_session):
+        intent="流程覆盖",
-        repo = IntelligentEvalRepository(db_session)
+        role_description="模拟用户",
-        ev = repo.create(IntelligentEval(name="eval", target_id="t1"))
-        assert ev.status == IntelligentEvalStatus.DRAFT
-
-        ev = repo._compare_and_set_status(
-            ev.id, expected_status=IntelligentEvalStatus.DRAFT, new_status=IntelligentEvalStatus.PLANNING
-        ).evaluation
-        assert ev.status == IntelligentEvalStatus.PLANNING
-
-        ev = repo._submit_plan_if_planning(ev.id, {"dimensions": ["test"]}).evaluation
-        assert ev.status == IntelligentEvalStatus.PENDING_APPROVAL
-
-        ev = repo._compare_and_set_status(
-            ev.id,
-            expected_status=IntelligentEvalStatus.PENDING_APPROVAL,
-            new_status=IntelligentEvalStatus.EXECUTING,
-        ).evaluation
-        assert ev.status == IntelligentEvalStatus.EXECUTING
-        assert ev.started_at is not None
-
-        ev = repo._submit_report_if_executing(ev.id, {"summary": "done"}).evaluation
-        assert ev.status == IntelligentEvalStatus.COMPLETED
-        assert ev.completed_at is not None
-
-    def test_compare_and_set_status_reports_conflict_without_overwrite(self, db_session):
-        repo = IntelligentEvalRepository(db_session)
-        ev = repo.create(IntelligentEval(name="eval", target_id="t1"))
-
-        applied = repo._compare_and_set_status(
-            ev.id,
-            expected_status=IntelligentEvalStatus.DRAFT,
-            new_status=IntelligentEvalStatus.PLANNING,
     )
-        assert applied.status is CompareAndSetStatus.APPLIED
-        assert applied.evaluation is not None
-        assert applied.evaluation.status is IntelligentEvalStatus.PLANNING
 
-        conflict = repo._compare_and_set_status(
-            ev.id,
-            expected_status=IntelligentEvalStatus.DRAFT,
-            new_status=IntelligentEvalStatus.EXECUTING,
-        )
-        assert conflict.status is CompareAndSetStatus.CONFLICT
-        assert repo.get(ev.id).status is IntelligentEvalStatus.PLANNING
 
-    def test_compare_and_set_status_reports_missing(self, db_session):
+def _start(session, *, name: str = "智能评估"):
-        result = IntelligentEvalRepository(db_session)._compare_and_set_status(
+    evaluation = _create(session, name=name)
-            "missing",
+    lifecycle.submit_plan(session, evaluation.id, {"dimensions": ["退货"]})
-            expected_status=IntelligentEvalStatus.DRAFT,
+    return lifecycle.approve(session, evaluation.id)
-            new_status=IntelligentEvalStatus.PLANNING,
-        )
-        assert result.status is CompareAndSetStatus.NOT_FOUND
 
-    def test_compare_and_set_status_rolls_back_failed_transaction(self, db_session, monkeypatch):
-        repo = IntelligentEvalRepository(db_session)
-        ev = repo.create(IntelligentEval(name="eval", target_id="t1"))
 
-        def fail_commit():
+def test_lifecycle_owns_complete_evaluation_state_machine(eval_session) -> None:
-            raise RuntimeError("commit failed")
+    evaluation = _create(eval_session)
+    assert evaluation.status is IntelligentEvalStatus.PLANNING
+
+    pending = lifecycle.submit_plan(eval_session, evaluation.id, {"dimensions": ["退货"]})
+    assert pending.status is IntelligentEvalStatus.PENDING_APPROVAL
+    assert pending.plan == {"dimensions": ["退货"]}
+
+    executing = lifecycle.approve(eval_session, evaluation.id)
+    assert executing.status is IntelligentEvalStatus.EXECUTING
+    assert executing.started_at is not None
+
+    completed = lifecycle.submit_report(eval_session, evaluation.id, {"summary": "done"})
+    assert completed.status is IntelligentEvalStatus.COMPLETED
+    assert completed.report == {"summary": "done"}
+    assert completed.completed_at is not None
+
+
+def test_lifecycle_rejects_invalid_or_repeated_transitions(eval_session) -> None:
+    evaluation = _create(eval_session)
+
+    with pytest.raises(IntelligentEvalTransitionError):
+        lifecycle.approve(eval_session, evaluation.id)
+
+    lifecycle.submit_plan(eval_session, evaluation.id, {"dimensions": ["退货"]})
+    with pytest.raises(IntelligentEvalTransitionError):
+        lifecycle.submit_plan(eval_session, evaluation.id, {"dimensions": ["重复"]})
+
+
+def test_plan_transaction_failure_leaves_public_snapshot_unchanged(eval_session, monkeypatch) -> None:
+    evaluation = _create(eval_session)
+    monkeypatch.setattr(eval_session, "commit", lambda: (_ for _ in ()).throw(RuntimeError("commit failed")))
 
-        monkeypatch.setattr(db_session, "commit", fail_commit)
     with pytest.raises(RuntimeError, match="commit failed"):
-            repo._compare_and_set_status(
+        lifecycle.submit_plan(eval_session, evaluation.id, {"dimensions": ["退货"]})
-                ev.id,
-                expected_status=IntelligentEvalStatus.DRAFT,
-                new_status=IntelligentEvalStatus.PLANNING,
-            )
 
     monkeypatch.undo()
-        assert repo.get(ev.id).status is IntelligentEvalStatus.DRAFT
+    unchanged = lifecycle.get_eval(eval_session, evaluation.id)
-
-    def test_conditional_plan_write_updates_plan_and_status_together(self, db_session):
-        repo = IntelligentEvalRepository(db_session)
-        ev = repo.create(IntelligentEval(name="eval", target_id="t1", status=IntelligentEvalStatus.PLANNING))
-
-        result = repo._submit_plan_if_planning(ev.id, {"dimensions": ["退货"]})
-
-        assert result.status is CompareAndSetStatus.APPLIED
-        updated = repo.get(ev.id)
-        assert updated.status is IntelligentEvalStatus.PENDING_APPROVAL
-        assert updated.plan == {"dimensions": ["退货"]}
-        assert updated.plan_feedback is None
-
-    def test_conditional_plan_write_failure_leaves_state_unchanged(self, db_session, monkeypatch):
-        repo = IntelligentEvalRepository(db_session)
-        ev = repo.create(IntelligentEval(name="eval", target_id="t1", status=IntelligentEvalStatus.PLANNING))
-
-        monkeypatch.setattr(db_session, "commit", lambda: (_ for _ in ()).throw(RuntimeError("commit failed")))
-        with pytest.raises(RuntimeError, match="commit failed"):
-            repo._submit_plan_if_planning(ev.id, {"dimensions": ["退货"]})
-
-        monkeypatch.undo()
-        unchanged = repo.get(ev.id)
     assert unchanged.status is IntelligentEvalStatus.PLANNING
     assert unchanged.plan is None
 
-    def test_plan_and_report_json(self, db_session):
-        repo = IntelligentEvalRepository(db_session)
-        ev = repo.create(IntelligentEval(name="eval", target_id="t1", status=IntelligentEvalStatus.PLANNING))
 
-        plan = {"dimensions": ["退货"], "virtual_users": [], "estimated_sessions": 3}
+def test_session_ownership_and_close_invariants_are_lifecycle_rules(eval_session) -> None:
-        ev = repo._submit_plan_if_planning(ev.id, plan).evaluation
+    first = _start(eval_session, name="first")
-        assert ev.plan == plan
+    second = _start(eval_session, name="second")
-
+    session_obj = lifecycle.open_session(
-        ev = repo._compare_and_set_status(
+        eval_session,
-            ev.id,
+        eval_id=first.id,
-            expected_status=IntelligentEvalStatus.PENDING_APPROVAL,
+        persona={"name": "老客户"},
-            new_status=IntelligentEvalStatus.EXECUTING,
+        goal="完成退货",
-        ).evaluation
+        dimension="退货",
-        report = {"summary": "good", "findings": []}
-        ev = repo._submit_report_if_executing(ev.id, report).evaluation
-        assert ev.report == report
-
-
-class TestIntelligentEvalSessionRepository:
-    def test_create_and_list(self, db_session):
-        eval_repo = IntelligentEvalRepository(db_session)
-        ev = eval_repo.create(IntelligentEval(name="eval", target_id="t1"))
-
-        sess_repo = IntelligentEvalSessionRepository(db_session)
-        sess = sess_repo.create(
-            IntelligentEvalSession(
-                eval_id=ev.id,
-                target_id="t1",
-                persona={"name": "user1", "patience": "low"},
-                goal="complete return",
-                dimension="退货流程",
     )
+
+    with pytest.raises(IntelligentEvalNotFoundError):
+        lifecycle.close_session(
+            eval_session,
+            eval_id=second.id,
+            session_id=session_obj.id,
+            verdict={"goal_achieved": False},
         )
-        assert sess.id is not None
-        assert sess.status == IntelligentEvalSessionStatus.RUNNING
-        assert sess.persona == {"name": "user1", "patience": "low"}
 
-        sessions = sess_repo.list_by_eval(ev.id)
+    closed = lifecycle.close_session(
-        assert len(sessions) == 1
+        eval_session,
-
+        eval_id=first.id,
-    def test_list_by_evals_batches_and_preserves_empty_groups(self, db_session):
+        session_id=session_obj.id,
-        eval_repo = IntelligentEvalRepository(db_session)
+        verdict={"goal_achieved": True},
-        first = eval_repo.create(IntelligentEval(name="first", target_id="t1"))
+    )
-        second = eval_repo.create(IntelligentEval(name="second", target_id="t1"))
-
-        sess_repo = IntelligentEvalSessionRepository(db_session)
-        sess_repo.create(IntelligentEvalSession(eval_id=first.id, target_id="t1", goal="first"))
-
-        grouped = sess_repo.list_by_evals([first.id, second.id, "missing"])
-
-        assert [item.goal for item in grouped[first.id]] == ["first"]
-        assert grouped[second.id] == []
-        assert grouped["missing"] == []
-
-    def test_close_session(self, db_session):
-        eval_repo = IntelligentEvalRepository(db_session)
-        ev = eval_repo.create(IntelligentEval(name="eval", target_id="t1"))
-
-        sess_repo = IntelligentEvalSessionRepository(db_session)
-        sess = sess_repo.create(IntelligentEvalSession(eval_id=ev.id, target_id="t1"))
-
-        verdict = {"goal_achieved": True, "issues": []}
-        closed = sess_repo.close(sess.id, verdict)
-        assert closed.status == IntelligentEvalSessionStatus.COMPLETED
-        assert closed.verdict == verdict
-        assert closed.closed_at is not None
-
-    def test_increment_turns(self, db_session):
-        eval_repo = IntelligentEvalRepository(db_session)
-        ev = eval_repo.create(IntelligentEval(name="eval", target_id="t1"))
-
-        sess_repo = IntelligentEvalSessionRepository(db_session)
-        sess = sess_repo.create(IntelligentEvalSession(eval_id=ev.id, target_id="t1"))
-        assert sess.turn_count == 0
-
-        sess_repo.increment_turns(sess.id)
-        sess_repo.increment_turns(sess.id)
-        updated = sess_repo.get(sess.id)
-        assert updated.turn_count == 2
-
-    def test_close_if_running_is_conditional(self, db_session):
-        eval_repo = IntelligentEvalRepository(db_session)
-        ev = eval_repo.create(IntelligentEval(name="eval", target_id="t1"))
-
-        sess_repo = IntelligentEvalSessionRepository(db_session)
-        sess = sess_repo.create(IntelligentEvalSession(eval_id=ev.id, target_id="t1"))
-        status, closed = sess_repo._close_if_running(sess.id, {"goal_achieved": True})
-
-        assert status is CompareAndSetStatus.APPLIED
-        assert closed is not None
     assert closed.status is IntelligentEvalSessionStatus.COMPLETED
+    assert closed.verdict == {"goal_achieved": True}
 
-        status, closed = sess_repo._close_if_running(sess.id, {"goal_achieved": False})
-        assert status is CompareAndSetStatus.CONFLICT
-        assert closed is None
 
-    def test_create_if_executing_uses_parent_target_and_rejects_other_states(self, db_session):
+class _ReplyChannel:
-        eval_repo = IntelligentEvalRepository(db_session)
+    async def exchange(self, _content, *, on_sent, **_kwargs):
-        ev = eval_repo.create(IntelligentEval(name="eval", target_id="t1"))
+        await on_sent(SendResult(ok=True, question_msg_id="message-1"))
-        sess_repo = IntelligentEvalSessionRepository(db_session)
+        return ExchangeOutcome.succeeded(correlation_id="message-1", reply="答复", latency_ms=12)
 
-        status, created = sess_repo._create_if_executing(
+
-            IntelligentEvalSession(eval_id=ev.id, target_id="wrong", goal="goal")
+class _TimeoutChannel:
+    async def exchange(self, _content, *, on_sent, **_kwargs):
+        await on_sent(SendResult(ok=True, question_msg_id="message-1"))
+        return ExchangeOutcome.reply_timeout(correlation_id="message-1", latency_ms=30_000)
+
+
+async def test_turn_ledger_is_observable_only_through_lifecycle(eval_session, monkeypatch) -> None:
+    evaluation = _start(eval_session)
+    session_obj = lifecycle.open_session(eval_session, eval_id=evaluation.id, persona={}, goal="完成退货")
+    monkeypatch.setattr(lifecycle.ChannelFactory, "create", lambda _target: _ReplyChannel())
+
+    result = await lifecycle.conduct_turn(
+        eval_session,
+        eval_id=evaluation.id,
+        session_id=session_obj.id,
+        content="你好",
     )
-        assert status is CompareAndSetStatus.CONFLICT
-        assert created is None
 
-        eval_repo._compare_and_set_status(
+    messages = lifecycle.list_messages(eval_session, eval_id=evaluation.id, session_id=session_obj.id)
-            ev.id,
+    sessions = lifecycle.list_sessions(eval_session, evaluation.id)
-            expected_status=IntelligentEvalStatus.DRAFT,
+    assert result == {"reply": "答复", "latency_ms": 12, "turn_count": 1}
-            new_status=IntelligentEvalStatus.EXECUTING,
+    assert [(item.role, item.content) for item in messages] == [("user", "你好"), ("assistant", "答复")]
+    assert sessions[0].turn_count == 1
+
+
+async def test_timeout_preserves_sent_message_and_turn_count(eval_session, monkeypatch) -> None:
+    evaluation = _start(eval_session)
+    session_obj = lifecycle.open_session(eval_session, eval_id=evaluation.id, persona={}, goal="完成退货")
+    monkeypatch.setattr(lifecycle.ChannelFactory, "create", lambda _target: _TimeoutChannel())
+
+    with pytest.raises(IntelligentEvalChannelError, match="超时"):
+        await lifecycle.conduct_turn(
+            eval_session,
+            eval_id=evaluation.id,
+            session_id=session_obj.id,
+            content="你好",
         )
-        status, created = sess_repo._create_if_executing(
-            IntelligentEvalSession(eval_id=ev.id, target_id="wrong", goal="goal")
-        )
-        assert status is CompareAndSetStatus.APPLIED
-        assert created is not None
-        assert created.target_id == "t1"
 
-    def test_user_message_and_turn_count_share_one_transaction(self, db_session):
+    messages = lifecycle.list_messages(eval_session, eval_id=evaluation.id, session_id=session_obj.id)
-        eval_repo = IntelligentEvalRepository(db_session)
+    sessions = lifecycle.list_sessions(eval_session, evaluation.id)
-        ev = eval_repo.create(IntelligentEval(name="eval", target_id="t1"))
+    assert [(item.role, item.content) for item in messages] == [("user", "你好")]
-        eval_repo._compare_and_set_status(
+    assert sessions[0].turn_count == 1
-            ev.id,
-            expected_status=IntelligentEvalStatus.DRAFT,
-            new_status=IntelligentEvalStatus.EXECUTING,
-        )
-        sess_repo = IntelligentEvalSessionRepository(db_session)
-        _, sess = sess_repo._create_if_executing(IntelligentEvalSession(eval_id=ev.id, target_id="t1"))
-
-        message_repo = IntelligentEvalMessageRepository(db_session)
-        message = IntelligentEvalMessage(session_id=sess.id, content="hello")
-        assert message_repo._create_user_and_increment(message) is CompareAndSetStatus.APPLIED
-        assert sess_repo.get(sess.id).turn_count == 1
-        assert message_repo.list_by_session(sess.id)[0].content == "hello"
-
-
-class TestIntelligentEvalMessageRepository:
-    def test_create_and_list(self, db_session):
-        eval_repo = IntelligentEvalRepository(db_session)
-        ev = eval_repo.create(IntelligentEval(name="eval", target_id="t1"))
-
-        sess_repo = IntelligentEvalSessionRepository(db_session)
-        sess = sess_repo.create(IntelligentEvalSession(eval_id=ev.id, target_id="t1"))
-
-        msg_repo = IntelligentEvalMessageRepository(db_session)
-        msg_repo.create(IntelligentEvalMessage(session_id=sess.id, role="user", content="hello"))
-        msg_repo.create(IntelligentEvalMessage(session_id=sess.id, role="assistant", content="hi", latency_ms=120))
-
-        messages = msg_repo.list_by_session(sess.id)
-        assert len(messages) == 2
-        assert messages[0].role == "user"
-        assert messages[1].role == "assistant"
-        assert messages[1].latency_ms == 120

tests/unit/test_intelligent_eval_read_model.py

@@ -8,13 +8,16 @@ from agenteval.intelligent_eval.models import (
     IntelligentEvalSessionStatus,
     IntelligentEvalStatus,
 )
-from agenteval.intelligent_eval.read_model import project_detail, project_list_item
+from agenteval.intelligent_eval.read_model import IntelligentEvalReadModel, project_detail, project_list_item
+from agenteval.intelligent_eval.repository import IntelligentEvalRepository, IntelligentEvalSessionRepository
+from sqlalchemy import event
+from sqlmodel import Session
 
 
-def _evaluation() -> IntelligentEval:
+def _evaluation(eval_id: str = "eval-1") -> IntelligentEval:
     now = datetime(2026, 1, 1, tzinfo=timezone.utc)
     return IntelligentEval(
-        id="eval-1",
+        id=eval_id,
         name="客服评估",
         target_id="target-1",
         status=IntelligentEvalStatus.EXECUTING,
@@ -29,10 +32,14 @@ def _evaluation() -> IntelligentEval:
     )
 
 
-def _session(session_id: str, status: IntelligentEvalSessionStatus) -> IntelligentEvalSession:
+def _session(
+    session_id: str,
+    status: IntelligentEvalSessionStatus,
+    eval_id: str = "eval-1",
+) -> IntelligentEvalSession:
     return IntelligentEvalSession(
         id=session_id,
-        eval_id="eval-1",
+        eval_id=eval_id,
         target_id="target-1",
         persona={"name": session_id},
         goal="完成退货",
@@ -63,3 +70,56 @@ def test_detail_projection_contains_session_metadata_without_messages() -> None:
     assert "messages" not in payload
     assert "content" not in payload["sessions"][0]
 
+
+def _count_queries(session: Session, action) -> int:
+    count = 0
+
+    def before_cursor_execute(*_args) -> None:
+        nonlocal count
+        count += 1
+
+    engine = session.get_bind()
+    event.listen(engine, "before_cursor_execute", before_cursor_execute)
+    try:
+        action()
+    finally:
+        event.remove(engine, "before_cursor_execute", before_cursor_execute)
+    return count
+
+
+def test_empty_list_projection_does_not_query_sessions(db_session: Session) -> None:
+    reader = IntelligentEvalReadModel(db_session)
+
+    query_count = _count_queries(db_session, lambda: reader.list_items([]))
+
+    assert query_count == 0
+
+
+def test_list_projection_loads_all_sessions_with_one_query(db_session: Session) -> None:
+    sessions = IntelligentEvalSessionRepository(db_session)
+    sessions._create(_session("s-1", IntelligentEvalSessionStatus.RUNNING, "eval-1"))
+    sessions._create(_session("s-2", IntelligentEvalSessionStatus.COMPLETED, "eval-2"))
+    reader = IntelligentEvalReadModel(db_session)
+    evaluations = [_evaluation("eval-1"), _evaluation("eval-2")]
+    result: list = []
+
+    query_count = _count_queries(db_session, lambda: result.extend(reader.list_items(evaluations)))
+
+    assert query_count == 1
+    assert [(item.id, item.session_count) for item in result] == [("eval-1", 1), ("eval-2", 1)]
+
+
+def test_detail_projection_uses_one_snapshot_query(db_session: Session) -> None:
+    evaluations = IntelligentEvalRepository(db_session)
+    sessions = IntelligentEvalSessionRepository(db_session)
+    evaluations._create(_evaluation())
+    sessions._create(_session("s-1", IntelligentEvalSessionStatus.COMPLETED))
+    reader = IntelligentEvalReadModel(db_session)
+    result: list = []
+
+    query_count = _count_queries(db_session, lambda: result.append(reader.detail_by_id("eval-1")))
+
+    assert query_count == 1
+    assert result[0] is not None
+    assert result[0].id == "eval-1"
+    assert [item.id for item in result[0].sessions] == ["s-1"]