"""Config snapshot management for intelligent evaluations (配置快照管理). Automatically saves config snapshots when: - Eval is created (snapshot_type: created) - Plan is submitted (snapshot_type: plan_submitted) - Config is updated (snapshot_type: config_updated) """ from typing import Any, Optional from sqlmodel import Session, select from agenteval.storage.db import IntelligentEvalConfigSnapshotDB, IntelligentEvalDB def save_snapshot( eval_db: IntelligentEvalDB, snapshot_type: str, created_by: str, session: Session, ) -> IntelligentEvalConfigSnapshotDB: """Save a config snapshot. Args: eval_db: Evaluation database record snapshot_type: Type of snapshot (created / plan_submitted / config_updated) created_by: Who created the snapshot (user / openclaw) session: Database session Returns: Created snapshot """ snapshot = IntelligentEvalConfigSnapshotDB( eval_id=eval_db.id, snapshot_type=snapshot_type, goal=eval_db.goal, seeds=eval_db.seeds, intent=eval_db.intent, role_description=eval_db.role_description, time_window_hours=eval_db.time_window_hours, plan=eval_db.plan, created_by=created_by, ) session.add(snapshot) session.commit() session.refresh(snapshot) return snapshot def list_snapshots(eval_id: str, session: Session) -> list[IntelligentEvalConfigSnapshotDB]: """List all snapshots for an evaluation. Returns: List of snapshots, ordered by created_at descending (newest first) """ snapshots = session.exec( select(IntelligentEvalConfigSnapshotDB) .where(IntelligentEvalConfigSnapshotDB.eval_id == eval_id) .order_by(IntelligentEvalConfigSnapshotDB.created_at.desc()) ).all() return list(snapshots) def get_snapshot(snapshot_id: str, session: Session) -> Optional[IntelligentEvalConfigSnapshotDB]: """Get a single snapshot by ID. Returns: Snapshot, or None if not found """ return session.get(IntelligentEvalConfigSnapshotDB, snapshot_id) def compare_snapshots( snapshot1: IntelligentEvalConfigSnapshotDB, snapshot2: IntelligentEvalConfigSnapshotDB, ) -> dict[str, Any]: """Compare two snapshots and return differences. Returns: Dict with differences, format: { "field_name": { "old": value1, "new": value2, } } """ diffs = {} # Compare simple fields fields = ["goal", "intent", "role_description", "time_window_hours"] for field in fields: val1 = getattr(snapshot1, field) val2 = getattr(snapshot2, field) if val1 != val2: diffs[field] = {"old": val1, "new": val2} # Compare JSON fields seeds1 = snapshot1.get_seeds() seeds2 = snapshot2.get_seeds() if seeds1 != seeds2: diffs["seeds"] = {"old": seeds1, "new": seeds2} plan1 = snapshot1.get_plan() plan2 = snapshot2.get_plan() if plan1 != plan2: diffs["plan"] = {"old": plan1, "new": plan2} return diffs