From 62485684ca3c1b1da0b95fc4a838c1b8ef19976e Mon Sep 17 00:00:00 2001 From: sinohqb Date: Fri, 7 Aug 2026 10:59:27 +0800 Subject: [PATCH] fix(architecture): enforce lifecycle consistency --- backend/agenteval/channels/__init__.py | 2 + backend/agenteval/channels/base.py | 12 +- backend/agenteval/channels/http.py | 8 +- backend/agenteval/channels/openclaw.py | 11 +- backend/agenteval/channels/tutu.py | 8 +- .../evaluation/campaign_lifecycle.py | 46 +- .../agenteval/evaluation/campaign_runner.py | 66 ++- .../agenteval/intelligent_eval/lifecycle.py | 4 +- .../agenteval/intelligent_eval/read_model.py | 9 +- .../agenteval/intelligent_eval/repository.py | 64 ++- backend/agenteval/storage/repository.py | 138 ++++-- backend/agenteval/web/routers/campaigns.py | 12 +- .../web/routers/intelligent_evals.py | 5 +- ...03-campaign-phased-static-then-adaptive.md | 2 +- .../web/src/read/useIntelligentEvalRead.ts | 20 +- tests/integration/test_campaign_runner.py | 39 ++ .../test_campaign_scheduler_loop.py | 20 + tests/unit/test_campaign_cancel_lifecycle.py | 35 +- .../unit/test_campaign_complete_lifecycle.py | 46 +- tests/unit/test_channel_contract.py | 10 +- tests/unit/test_engine.py | 4 +- tests/unit/test_http_channel_and_rules.py | 29 +- tests/unit/test_intelligent_eval_model.py | 444 ++++++------------ .../unit/test_intelligent_eval_read_model.py | 70 ++- 24 files changed, 599 insertions(+), 505 deletions(-) diff --git a/backend/agenteval/channels/__init__.py b/backend/agenteval/channels/__init__.py index a72ac00..fc48fea 100644 --- a/backend/agenteval/channels/__init__.py +++ b/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", diff --git a/backend/agenteval/channels/base.py b/backend/agenteval/channels/base.py index cc81456..cb58355 100644 --- a/backend/agenteval/channels/base.py +++ b/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, diff --git a/backend/agenteval/channels/http.py b/backend/agenteval/channels/http.py index 168f3f2..ac592ca 100644 --- a/backend/agenteval/channels/http.py +++ b/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: - pass + except (httpx.HTTPError, json.JSONDecodeError) as exc: + raise ChannelTransportError(str(exc)) from exc await asyncio.sleep(interval) diff --git a/backend/agenteval/channels/openclaw.py b/backend/agenteval/channels/openclaw.py index ec442a5..4c41223 100644 --- a/backend/agenteval/channels/openclaw.py +++ b/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: - pass + elif resp.status_code != 404: + 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) diff --git a/backend/agenteval/channels/tutu.py b/backend/agenteval/channels/tutu.py index ab0c47b..001df08 100644 --- a/backend/agenteval/channels/tutu.py +++ b/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: - pass + except (httpx.HTTPError, json.JSONDecodeError) as exc: + raise ChannelTransportError(f"轮询异常: {exc}") from exc await asyncio.sleep(poll_interval) diff --git a/backend/agenteval/evaluation/campaign_lifecycle.py b/backend/agenteval/evaluation/campaign_lifecycle.py index 1ca7440..fe4ad38 100644 --- a/backend/agenteval/evaluation/campaign_lifecycle.py +++ b/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): - """Cancellation failure translated by the HTTP adapter.""" +class CampaignLifecycleError(Exception): + """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()) - if status != "applied" or campaign is None: + result = repo.start_if_planned(campaign_id, utc_now()) + 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()) - if status == "not_found": - raise CampaignCancelError(404, "campaign not found") - if status == "conflict" or campaign is None: - raise CampaignCancelError(400, "campaign is not in a cancellable state") + result = repo.cancel_if_active(campaign_id, utc_now()) + campaign = result.campaign + if result.status is CampaignWriteStatus.NOT_FOUND: + raise CampaignLifecycleError(404, "campaign not found") + 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; - a failed settlement can be retried without reverting the terminal status. + the database. Settlement and the terminal status share one transaction, so + 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()) - if status == "not_found": - raise CampaignCancelError(404, "campaign not found") - if status == "conflict" or campaign is None: - raise CampaignCancelError(409, "campaign is not running") - if settle is not None: - settle(campaign_id, session) - else: - ExplorationSessionRepository(session).expire_running_sessions(campaign_id) + result = repo.complete_if_running(campaign_id, utc_now()) + campaign = result.campaign + if result.status is CampaignWriteStatus.NOT_FOUND: + raise CampaignLifecycleError(404, "campaign not found") + if result.status is CampaignWriteStatus.CONFLICT or campaign is None: + raise CampaignLifecycleError(409, "campaign is not running") return campaign diff --git a/backend/agenteval/evaluation/campaign_runner.py b/backend/agenteval/evaluation/campaign_runner.py index ec43d0d..60b985e 100644 --- a/backend/agenteval/evaluation/campaign_runner.py +++ b/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.""" - spawned = set(campaign.summary.scheduler.spawned_indices) if campaign.summary else set() - if campaign.id and session is not None: - 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( - plan_index - for plan_index, entry in enumerate(campaign.plan) - if set(range(entry.count)).issubset(occurrences.get(plan_index, set())) + """Return plan entries completed by durable identities or legacy progress. + + 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) + + return { + plan_index + for plan_index, entry in enumerate(campaign.plan) + 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), - but still marked spawned to avoid infinite retries. Returns ``None`` if the - campaign does not exist. + raises is recorded but remains due until every occurrence has a durable + child-Run identity. Returns ``None`` if the 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, @@ -385,7 +377,7 @@ async def run_campaign_loop( _logger.warning("活动 %s 完成迁移失败: %s", campaign_id, exc) else: _auto_start_analysis(completed, session) - return + return try: await asyncio.wait_for(cancel.wait(), timeout=tick_seconds) diff --git a/backend/agenteval/intelligent_eval/lifecycle.py b/backend/agenteval/intelligent_eval/lifecycle.py index 5390792..558ffd2 100644 --- a/backend/agenteval/intelligent_eval/lifecycle.py +++ b/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", diff --git a/backend/agenteval/intelligent_eval/read_model.py b/backend/agenteval/intelligent_eval/read_model.py index 53a156f..5f8b950 100644 --- a/backend/agenteval/intelligent_eval/read_model.py +++ b/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) - return self.detail(eval_obj) if eval_obj is not None else None + snapshot = self._evals.get_with_sessions(eval_id) + 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.""" diff --git a/backend/agenteval/intelligent_eval/repository.py b/backend/agenteval/intelligent_eval/repository.py index 1dbed94..5b86bfe 100644 --- a/backend/agenteval/intelligent_eval/repository.py +++ b/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, diff --git a/backend/agenteval/storage/repository.py b/backend/agenteval/storage/repository.py index 729162a..9e24fd7 100644 --- a/backend/agenteval/storage/repository.py +++ b/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]]: - """Compare-and-set a planned/running Campaign to cancelled.""" + def _compare_and_set_status( + 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) ) - 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)) + try: + result = self.session.exec(statement) + if result.rowcount != 1: + self.session.rollback() + db = self.session.get(CampaignDB, campaign_id) + 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]]: - """Compare-and-set a running Campaign to completed.""" - statement = ( - sql_update(CampaignDB) - .where(CampaignDB.id == campaign_id, CampaignDB.status == CampaignStatus.RUNNING.value) - .values(status=CampaignStatus.COMPLETED.value, completed_at=at) + if settle_exploration: + self.session.exec( + sql_update(ExplorationSessionDB) + .where( + ExplorationSessionDB.campaign_id == campaign_id, + ExplorationSessionDB.status == ExplorationSessionStatus.RUNNING.value, + ) + .values(status=ExplorationSessionStatus.EXPIRED.value, closed_at=at) + ) + self.session.commit() + except Exception: + self.session.rollback() + raise + + 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, ) - 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 start_if_planned(self, campaign_id: str, at: datetime) -> tuple[str, Optional[Campaign]]: + 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) - .where(CampaignDB.id == campaign_id, CampaignDB.status == CampaignStatus.PLANNED.value) - .values(status=CampaignStatus.RUNNING.value, started_at=at) + + return self._compare_and_set_status( + campaign_id, + 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 列，不覆写并发的状态 / 水位变更。""" diff --git a/backend/agenteval/web/routers/campaigns.py b/backend/agenteval/web/routers/campaigns.py index 751e355..5c11adc 100644 --- a/backend/agenteval/web/routers/campaigns.py +++ b/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.campaign_lifecycle import CampaignCancelError, CampaignCreateError +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 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() diff --git a/backend/agenteval/web/routers/intelligent_evals.py b/backend/agenteval/web/routers/intelligent_evals.py index a83b954..763c96c 100644 --- a/backend/agenteval/web/routers/intelligent_evals.py +++ b/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 = reader.detail(ev) if detail else reader.list_item(ev) + projection = IntelligentEvalReadModel(session).list_item(ev) return projection.model_dump(mode="json") diff --git a/docs/adr/0003-campaign-phased-static-then-adaptive.md b/docs/adr/0003-campaign-phased-static-then-adaptive.md index 7b0e50a..492e7b8 100644 --- a/docs/adr/0003-campaign-phased-static-then-adaptive.md +++ b/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 分析任务。该顺序保证恢复动作只依据已提交的数据库事实，不依赖上一次进程的内存状态。 diff --git a/frontend/web/src/read/useIntelligentEvalRead.ts b/frontend/web/src/read/useIntelligentEvalRead.ts index 69e32cb..c6ab4ba 100644 --- a/frontend/web/src/read/useIntelligentEvalRead.ts +++ b/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(() => { - if (!listActive) return - const timer = window.setInterval(() => { void loadList(true) }, 5000) - return () => window.clearInterval(timer) - }, [listActive, loadList]) - - useEffect(() => { - if (selectedId == null || !detailActive) return - const timer = window.setInterval(() => { void loadDetail(selectedId, true) }, 5000) - return () => window.clearInterval(timer) - }, [detailActive, loadDetail, selectedId]) + usePolling(() => { void loadList(true) }, 5000, listActive) + usePolling( + () => { + if (selectedId != null) void loadDetail(selectedId, true) + }, + 5000, + selectedId != null && detailActive, + ) const reloadList = useCallback(() => loadList(true), [loadList]) const reloadDetail = useCallback( diff --git a/tests/integration/test_campaign_runner.py b/tests/integration/test_campaign_runner.py index c05ad29..66857bf 100644 --- a/tests/integration/test_campaign_runner.py +++ b/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( diff --git a/tests/integration/test_campaign_scheduler_loop.py b/tests/integration/test_campaign_scheduler_loop.py index f83662a..b09b9c5 100644 --- a/tests/integration/test_campaign_scheduler_loop.py +++ b/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. diff --git a/tests/unit/test_campaign_cancel_lifecycle.py b/tests/unit/test_campaign_cancel_lifecycle.py index 4c7069a..09c1c1c 100644 --- a/tests/unit/test_campaign_cancel_lifecycle.py +++ b/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 diff --git a/tests/unit/test_campaign_complete_lifecycle.py b/tests/unit/test_campaign_complete_lifecycle.py index fb6797a..9ff39c4 100644 --- a/tests/unit/test_campaign_complete_lifecycle.py +++ b/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): - observed.append(CampaignRepository(session).get(campaign_id).status) - - completed = complete_campaign(db_session, "campaign-1", settle=settle) + completed = complete_campaign(db_session, "campaign-1") 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 diff --git a/tests/unit/test_channel_contract.py b/tests/unit/test_channel_contract.py index d720f1b..d66f552 100644 --- a/tests/unit/test_channel_contract.py +++ b/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("问题") diff --git a/tests/unit/test_engine.py b/tests/unit/test_engine.py index 9d46fa3..4a9b13c 100644 --- a/tests/unit/test_engine.py +++ b/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() diff --git a/tests/unit/test_http_channel_and_rules.py b/tests/unit/test_http_channel_and_rules.py index 092352f..a064ad1 100644 --- a/tests/unit/test_http_channel_and_rules.py +++ b/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 diff --git a/tests/unit/test_intelligent_eval_model.py b/tests/unit/test_intelligent_eval_model.py index 9cd446c..4c6f3a6 100644 --- a/tests/unit/test_intelligent_eval_model.py +++ b/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 ( - IntelligentEval, - IntelligentEvalMessage, - IntelligentEvalSession, - IntelligentEvalSessionStatus, - IntelligentEvalStatus, +from agenteval.channels.base import ExchangeOutcome, SendResult +from agenteval.intelligent_eval import lifecycle +from agenteval.intelligent_eval.lifecycle import ( + IntelligentEvalChannelError, + IntelligentEvalNotFoundError, + IntelligentEvalTransitionError, ) -from agenteval.intelligent_eval.repository import ( - CompareAndSetStatus, - IntelligentEvalMessageRepository, - IntelligentEvalRepository, - IntelligentEvalSessionRepository, -) -from agenteval.storage.db import EvalTargetDB -from sqlalchemy.pool import StaticPool -from sqlmodel import Session, SQLModel, create_engine +from agenteval.intelligent_eval.models import IntelligentEvalSessionStatus, IntelligentEvalStatus +from agenteval.models import ChannelType, EvalTarget, PlatformType, TargetStatus +from agenteval.storage.repository import TargetRepository -@pytest.fixture -def db_session(): - engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) - SQLModel.metadata.create_all(engine) - session = Session(engine) - target = EvalTargetDB(id="t1", name="test-target") - session.add(target) - session.commit() - yield session - session.close() - - -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, - ) +@pytest.fixture() +def eval_session(db_session): + TargetRepository(db_session).create( + EvalTarget( + id="target-1", + name="测试对象", + platform=PlatformType.AI_DIGITAL_EMPLOYEE, + channel_type=ChannelType.TUTU_API, + channel_config={"base_url": "http://mock", "token": "token"}, + status=TargetStatus.ACTIVE, ) - assert ev.id is not None - assert ev.status == IntelligentEvalStatus.DRAFT - assert ev.name == "test-eval" - assert ev.time_window_hours == 24 + ) + return db_session - 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): - repo = IntelligentEvalRepository(db_session) - repo.create(IntelligentEval(name="eval-1", target_id="t1")) - repo.create(IntelligentEval(name="eval-2", target_id="t1")) - all_evals = repo.list_all() - assert len(all_evals) == 2 +def _create(session, *, name: str = "智能评估"): + return lifecycle.create_eval( + session, + name=name, + target_id="target-1", + goal="验证退货流程", + seeds={"personas": ["老客户"]}, + intent="流程覆盖", + role_description="模拟用户", + ) - def test_status_transitions(self, db_session): - repo = IntelligentEvalRepository(db_session) - 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 +def _start(session, *, name: str = "智能评估"): + evaluation = _create(session, name=name) + lifecycle.submit_plan(session, evaluation.id, {"dimensions": ["退货"]}) + return lifecycle.approve(session, evaluation.id) - 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 +def test_lifecycle_owns_complete_evaluation_state_machine(eval_session) -> None: + evaluation = _create(eval_session) + assert evaluation.status is IntelligentEvalStatus.PLANNING - ev = repo._submit_report_if_executing(ev.id, {"summary": "done"}).evaluation - assert ev.status == IntelligentEvalStatus.COMPLETED - assert ev.completed_at is not None + pending = lifecycle.submit_plan(eval_session, evaluation.id, {"dimensions": ["退货"]}) + assert pending.status is IntelligentEvalStatus.PENDING_APPROVAL + assert pending.plan == {"dimensions": ["退货"]} - 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")) + executing = lifecycle.approve(eval_session, evaluation.id) + assert executing.status is IntelligentEvalStatus.EXECUTING + assert executing.started_at is not None - applied = repo._compare_and_set_status( - ev.id, - expected_status=IntelligentEvalStatus.DRAFT, - new_status=IntelligentEvalStatus.PLANNING, + 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"))) + + with pytest.raises(RuntimeError, match="commit failed"): + lifecycle.submit_plan(eval_session, evaluation.id, {"dimensions": ["退货"]}) + + monkeypatch.undo() + unchanged = lifecycle.get_eval(eval_session, evaluation.id) + assert unchanged.status is IntelligentEvalStatus.PLANNING + assert unchanged.plan is None + + +def test_session_ownership_and_close_invariants_are_lifecycle_rules(eval_session) -> None: + first = _start(eval_session, name="first") + second = _start(eval_session, name="second") + session_obj = lifecycle.open_session( + eval_session, + eval_id=first.id, + persona={"name": "老客户"}, + goal="完成退货", + dimension="退货", + ) + + with pytest.raises(IntelligentEvalNotFoundError): + lifecycle.close_session( + eval_session, + eval_id=second.id, + session_id=session_obj.id, + verdict={"goal_achieved": False}, ) - 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, + closed = lifecycle.close_session( + eval_session, + eval_id=first.id, + session_id=session_obj.id, + verdict={"goal_achieved": True}, + ) + assert closed.status is IntelligentEvalSessionStatus.COMPLETED + assert closed.verdict == {"goal_achieved": True} + + +class _ReplyChannel: + async def exchange(self, _content, *, on_sent, **_kwargs): + await on_sent(SendResult(ok=True, question_msg_id="message-1")) + return ExchangeOutcome.succeeded(correlation_id="message-1", reply="答复", latency_ms=12) + + +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="你好", + ) + + messages = lifecycle.list_messages(eval_session, eval_id=evaluation.id, session_id=session_obj.id) + sessions = lifecycle.list_sessions(eval_session, evaluation.id) + assert result == {"reply": "答复", "latency_ms": 12, "turn_count": 1} + 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="你好", ) - 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): - result = IntelligentEvalRepository(db_session)._compare_and_set_status( - "missing", - expected_status=IntelligentEvalStatus.DRAFT, - 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(): - raise RuntimeError("commit failed") - - monkeypatch.setattr(db_session, "commit", fail_commit) - with pytest.raises(RuntimeError, match="commit failed"): - repo._compare_and_set_status( - ev.id, - expected_status=IntelligentEvalStatus.DRAFT, - new_status=IntelligentEvalStatus.PLANNING, - ) - - monkeypatch.undo() - assert repo.get(ev.id).status is IntelligentEvalStatus.DRAFT - - 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} - ev = repo._submit_plan_if_planning(ev.id, plan).evaluation - assert ev.plan == plan - - ev = repo._compare_and_set_status( - ev.id, - expected_status=IntelligentEvalStatus.PENDING_APPROVAL, - new_status=IntelligentEvalStatus.EXECUTING, - ).evaluation - 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="退货流程", - ) - ) - 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) - assert len(sessions) == 1 - - def test_list_by_evals_batches_and_preserves_empty_groups(self, db_session): - eval_repo = IntelligentEvalRepository(db_session) - 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 - - 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): - eval_repo = IntelligentEvalRepository(db_session) - ev = eval_repo.create(IntelligentEval(name="eval", target_id="t1")) - sess_repo = IntelligentEvalSessionRepository(db_session) - - status, created = sess_repo._create_if_executing( - IntelligentEvalSession(eval_id=ev.id, target_id="wrong", goal="goal") - ) - assert status is CompareAndSetStatus.CONFLICT - assert created is None - - eval_repo._compare_and_set_status( - ev.id, - expected_status=IntelligentEvalStatus.DRAFT, - new_status=IntelligentEvalStatus.EXECUTING, - ) - 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): - eval_repo = IntelligentEvalRepository(db_session) - ev = eval_repo.create(IntelligentEval(name="eval", target_id="t1")) - eval_repo._compare_and_set_status( - 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 + messages = lifecycle.list_messages(eval_session, eval_id=evaluation.id, session_id=session_obj.id) + sessions = lifecycle.list_sessions(eval_session, evaluation.id) + assert [(item.role, item.content) for item in messages] == [("user", "你好")] + assert sessions[0].turn_count == 1 diff --git a/tests/unit/test_intelligent_eval_read_model.py b/tests/unit/test_intelligent_eval_read_model.py index 9c9ec93..015f107 100644 --- a/tests/unit/test_intelligent_eval_read_model.py +++ b/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"]