examples: harden eval optimization regression loop

Fail closed on ambiguous replay data, unsafe artifact labels, non-finite metrics, key-case pass regressions, and common tool error payloads.

Isolate optimizer prompt writes from source files, reject unsafe in-process concurrency, and add adversarial regression coverage for gate, attribution, path, apply, and concurrency behavior.

Fixes #91

RELEASE NOTES: Harden the evaluation and optimization example against ambiguous data, unsafe artifacts, and regression-gate edge cases.

Author: 16yunH

examples/optimization/eval_optimize_loop/DESIGN.md

@@ -1,3 +1,3 @@
 # 设计说明（约 400 字）
 
-闭环以 `EvalOptimizePipeline.run()` 作为唯一外部接口，把数据校验、基线评估、提示词优化、候选复评、差异计算、门禁和报告隐藏在同一深模块内。优化阶段直接复用 `AgentOptimizer`、`TargetPrompt` 与真实 GEPA；无密钥模式仅通过 `ModelRegistry` 注入离线 agent、judge 和 reflector，因而仍会经过 `LlmAgent`、内置 rubric 解析及优化器回调，不依赖 monkeypatch。外层将每个候选运行物化为 trace，再用 `AgentEvaluator` 独立评测 train 与 validation，避免采信优化器自己的聚合分数。候选集合按完整提示词哈希去重，即使优化器内部提前拒绝某次提案，外层仍会捕获并独立复评，防止遗漏验证回退。逐 case 差异区分新增通过、新增失败、升分、降分和不变；归因按工具执行、参数、格式、知识召回、rubric、最终回答的根因优先级输出证据。SDK 聚合器偶尔不保留逐 rubric 详情，格式归因依次采用详情、回放证据和请求中明确的 JSON、单行或 Markdown 约束，否定式要求不触发检查；judge 无有效 verdict 时单列 evaluation_error，避免把系统错误或内容失败误报为格式失败。Gate 对验证增益、配对 bootstrap 下界、新增 hard fail、关键 case、过拟合及资源预算做可配置 AND，并先过滤不安全候选，再计算质量、token、P95 时延 Pareto。运行前检查跨集合重复、近重复和答案泄漏；运行后保存完整 prompt、seed、哈希、耗时、成本、trace 与 JSON/Markdown 报告。默认始终恢复源 prompt，只有显式允许且 Gate 全通过才原子回写。
+闭环以 `EvalOptimizePipeline.run()` 作为唯一外部接口，把数据校验、基线评估、提示词优化、候选复评、差异计算、门禁和报告隐藏在同一深模块内。优化阶段直接复用 `AgentOptimizer`、`TargetPrompt` 与真实 GEPA；无密钥模式仅通过 `ModelRegistry` 注入离线 agent、judge 和 reflector，因而仍会经过 `LlmAgent`、内置 rubric 解析及优化器回调，不依赖 monkeypatch。外层将每个候选运行物化为 trace，再用 `AgentEvaluator` 独立评测 train 与 validation，避免采信优化器自己的聚合分数。候选集合按完整提示词哈希去重，即使优化器内部提前拒绝某次提案，外层仍会捕获并独立复评，防止遗漏验证回退。逐 case 差异区分新增通过、新增失败、升分、降分和不变；归因按工具执行、参数、格式、知识召回、rubric、最终回答的根因优先级输出证据。SDK 聚合器偶尔不保留逐 rubric 详情，格式归因依次采用详情、回放证据和请求中明确的 JSON、单行或 Markdown 约束，否定式要求不触发检查；judge 无有效 verdict 时单列 evaluation_error，避免把系统错误或内容失败误报为格式失败。Gate 对验证增益、配对 bootstrap 下界、新增 hard fail、关键 case、过拟合及资源预算做可配置 AND，并先过滤不安全候选，再计算质量、token、P95 时延 Pareto。运行前检查跨集合重复、近重复和答案泄漏；运行后保存完整 prompt、seed、哈希、耗时、成本、trace 与 JSON/Markdown 报告。默认仅在隔离 workspace 内改写 prompt，不触碰源文件，只有显式允许且 Gate 全通过才原子回写。

examples/optimization/eval_optimize_loop/README.md

@@ -18,7 +18,7 @@ python examples/optimization/eval_optimize_loop/run_pipeline.py \
   --output-dir /tmp/eval-optimize-loop
 ```
 
-命令正常完成时输出 `decision=accepted` 与报告路径。默认不会覆盖源 prompt；只有显式传入 `--apply-if-accepted` 且所有必选 Gate 均通过，才会原子回写最终候选。
+命令正常完成时输出 `decision=accepted` 与报告路径。优化器只改写输出目录中的隔离 prompt workspace，默认运行不会触碰源 prompt；只有显式传入 `--apply-if-accepted` 且所有必选 Gate 均通过，才会原子回写最终候选。
 
 ## 为什么不是“伪造优化”
 
@@ -71,7 +71,7 @@ reflection
 2. validation pass-rate 增益达到 `min_validation_gain`；
 3. paired bootstrap 区间下界达到配置值；
 4. 不新增 hard failure；
-5. key case 不降分；
+5. key case 不允许从通过变失败，也不允许降分；
 6. train 提升而 validation 下降时判定 overfit 并拒绝；趋势以 pass-rate 为主，持平时再比较 average score；
 7. metric calls、token、耗时和成本均在预算内。
 
@@ -82,6 +82,7 @@ reflection
 - baseline 与 candidate 在同一 case 上配对，使用固定 seed 的 2,000 次 bootstrap，报告 validation pass-rate delta 的 95% 区间。
 - 先过安全 Gate，再在合格候选中按 validation 质量、token 与 P95 latency 标记 Pareto 前沿。
 - 运行前检查重复 id、train/validation 精确重复、去空白后的重复、相似度 ≥0.92 的近重复，以及 baseline/candidate prompt 中直接出现 validation reference answer；命中即 fail-closed。
+- artifact label 只允许有限长度的字母、数字、下划线和连字符，候选 id 必须唯一；离线回放遇到空问题或同一问题映射多个 case 时会拒绝运行，避免路径逃逸和静默错配。
 - baseline、每个候选和每个 split 的 trace 均保存到 artifact 目录，可由 `AgentEvaluator` 重新播放。
 
 ## 输出
@@ -112,4 +113,4 @@ pytest -q tests/evaluation/test_eval_optimize_loop_example.py
 
 ## 接入真实业务
 
-保留 `EvalOptimizePipeline.run()` 这个外部 interface，将 `offline.py` 的三个 adapter 替换为业务实现：`call_agent` 驱动真实 Agent，reflection/judge 的 `provider_name` 改为实际 provider，trace 物化器读取业务运行日志。`optimizer.json`、`TargetPrompt`、逐候选复评、diff、Gate 和报告模型无需改变。生产环境建议每次使用唯一 output directory；离线 registry 使用进程级固定 replay 配置，不支持在同一 Python 进程中并发启动两个 pipeline。
+保留 `EvalOptimizePipeline.run()` 这个外部 interface，将 `offline.py` 的三个 adapter 替换为业务实现：`call_agent` 驱动真实 Agent，reflection/judge 的 `provider_name` 改为实际 provider，trace 物化器读取业务运行日志。`optimizer.json`、`TargetPrompt`、逐候选复评、diff、Gate 和报告模型无需改变。生产环境应为每次运行使用唯一 output directory；不同进程会在各自 workspace 优化而不触碰源 prompt，同一 Python 进程中的并发离线运行会明确拒绝，避免共享 registry 串扰。

examples/optimization/eval_optimize_loop/loop/analysis.py

@@ -83,6 +83,13 @@ def validate_data_quality(
 
         all_ids = [case.eval_id for case in train_set.eval_cases + validation_set.eval_cases]
         duplicate_ids = sorted(case_id for case_id, count in Counter(all_ids).items() if count > 1)
+        query_groups: dict[str, list[str]] = {}
+        for case in train_set.eval_cases + validation_set.eval_cases:
+            replay_query = self._case_replay_query(case)
+            if not replay_query:
+                raise ValueError(f"case {case.eval_id!r} has an empty user query")
+            query_groups.setdefault(replay_query, []).append(case.eval_id)
+        duplicate_queries = sorted(" == ".join(case_ids) for case_ids in query_groups.values() if len(case_ids) > 1)
         train_fingerprints = {self._case_fingerprint(case): case.eval_id for case in train_set.eval_cases}
         validation_fingerprints = {self._case_fingerprint(case): case.eval_id for case in validation_set.eval_cases}
         overlap = sorted(set(train_fingerprints) & set(validation_fingerprints))
@@ -106,9 +113,10 @@ def validate_data_quality(
             normalized = _normalize(expected)
             if len(normalized) >= 12 and normalized in prompt_normalized:
                 leakage.append(case.eval_id)
-        if duplicate_ids or cross_split or near_cross_split or leakage:
+        if duplicate_ids or duplicate_queries or cross_split or near_cross_split or leakage:
             raise ValueError("data quality check failed: "
-                             f"duplicate_ids={duplicate_ids}, cross_split={cross_split}, "
+                             f"duplicate_ids={duplicate_ids}, duplicate_queries={duplicate_queries}, "
+                             f"cross_split={cross_split}, "
                              f"near_cross_split={near_cross_split}, prompt_leakage={leakage}")
         return DataQualityAudit(
             passed=True,
@@ -131,6 +139,11 @@ def _case_normalized_content(case: EvalCase) -> str:
                             for invocation in conversation)
         return _normalize(payload)
 
+    @staticmethod
+    def _case_replay_query(case: EvalCase) -> str:
+        conversation = case.conversation or []
+        return _text(conversation[0].user_content) if conversation else ""
+
     @staticmethod
     def _expected_response(case: EvalCase) -> str:
         conversation = case.conversation or []
@@ -324,7 +337,8 @@ def gate(
         baseline_validation_by_id = {case.case_id: case for case in baseline.validation.cases}
         key_regressions = sorted(
             case.case_id for case in candidate_validation.cases
-            if case.key_case and baseline_validation_by_id[case.case_id].score - case.score > epsilon)
+            if case.key_case and ((baseline_validation_by_id[case.case_id].passed and not case.passed)
+                                  or baseline_validation_by_id[case.case_id].score - case.score > epsilon))
 
         def _trend(split_delta: SplitDelta) -> int:
             if split_delta.pass_rate_delta > epsilon:

examples/optimization/eval_optimize_loop/loop/models.py

@@ -10,6 +10,7 @@
 from __future__ import annotations
 
 import json
+import re
 from pathlib import Path
 from typing import Any
 from typing import Literal
@@ -18,6 +19,39 @@
 from pydantic import BaseModel
 from pydantic import ConfigDict
 from pydantic import Field
+from pydantic import field_validator
+from pydantic import model_validator
+
+_SAFE_ARTIFACT_LABEL = r"^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$"
+_WINDOWS_DEVICE_NAMES = {
+    "aux",
+    "con",
+    "nul",
+    "prn",
+    *(f"com{index}" for index in range(1, 10)),
+    *(f"lpt{index}" for index in range(1, 10)),
+}
+
+
+def _validate_artifact_label(
+    value: str,
+    *,
+    subject: str,
+    reserved: set[str] | None = None,
+) -> str:
+    if re.fullmatch(_SAFE_ARTIFACT_LABEL, value) is None:
+        raise ValueError(f"{subject} must be a safe artifact label")
+    folded = value.casefold()
+    if folded in _WINDOWS_DEVICE_NAMES or folded in (reserved or set()):
+        raise ValueError(f"{subject} is a reserved artifact label: {value!r}")
+    return value
+
+
+def _casefold_duplicates(values: list[str]) -> list[str]:
+    groups: dict[str, list[str]] = {}
+    for value in values:
+        groups.setdefault(value.casefold(), []).append(value)
+    return sorted(" == ".join(group) for group in groups.values() if len(group) > 1)
 
 
 class _StrictModel(BaseModel):
@@ -27,9 +61,18 @@ class _StrictModel(BaseModel):
 class CandidateSource(_StrictModel):
     """A deterministic prompt proposal used by the offline reflection model."""
 
-    candidate_id: str
+    candidate_id: str = Field(pattern=_SAFE_ARTIFACT_LABEL)
     path: Path
 
+    @field_validator("candidate_id")
+    @classmethod
+    def _candidate_id_is_artifact_safe(cls, value: str) -> str:
+        return _validate_artifact_label(
+            value,
+            subject="candidate_id",
+            reserved={"baseline"},
+        )
+
 
 class PipelineSpec(_StrictModel):
     """All paths and run controls needed by :class:`EvalOptimizePipeline`."""
@@ -40,14 +83,29 @@ class PipelineSpec(_StrictModel):
     gate_config: Path
     train_dataset: Path
     validation_dataset: Path
-    target_prompts: dict[str, Path]
-    candidate_sources: list[CandidateSource]
+    target_prompts: dict[str, Path] = Field(min_length=1)
+    candidate_sources: list[CandidateSource] = Field(min_length=1)
     output_dir: Path
     seed: int = 91
     bootstrap_samples: int = Field(default=2000, ge=100)
     confidence_level: float = Field(default=0.95, gt=0.0, lt=1.0)
     apply_if_accepted: bool = False
 
+    @model_validator(mode="after")
+    def _validate_artifact_labels(self) -> "PipelineSpec":
+        candidate_ids = [source.candidate_id for source in self.candidate_sources]
+        duplicate_ids = _casefold_duplicates(candidate_ids)
+        if duplicate_ids:
+            raise ValueError(f"candidate_ids must be case-insensitively unique: {duplicate_ids}")
+        prompt_names = list(self.target_prompts)
+        duplicate_prompt_names = _casefold_duplicates(prompt_names)
+        if duplicate_prompt_names:
+            raise ValueError("target prompt names must be case-insensitively unique: "
+                             f"{duplicate_prompt_names}")
+        for name in prompt_names:
+            _validate_artifact_label(name, subject="target prompt name")
+        return self
+
     @classmethod
     def from_file(
         cls,
@@ -94,8 +152,8 @@ def _resolve(value: str | Path) -> Path:
 
 class MetricOutcome(_StrictModel):
     metric_name: str
-    score: Optional[float] = None
-    threshold: float
+    score: Optional[float] = Field(default=None, allow_inf_nan=False)
+    threshold: float = Field(allow_inf_nan=False)
     passed: bool
     reason: str = ""
 
@@ -115,7 +173,7 @@ class TrajectoryStep(_StrictModel):
 class CaseEvaluation(_StrictModel):
     case_id: str
     passed: bool
-    score: float
+    score: float = Field(allow_inf_nan=False)
     key_case: bool = False
     hard_fail: bool = False
     metrics: list[MetricOutcome] = Field(default_factory=list)
@@ -128,8 +186,8 @@ class CaseEvaluation(_StrictModel):
 
 class SplitEvaluation(_StrictModel):
     split: Literal["train", "validation"]
-    pass_rate: float
-    average_score: float
+    pass_rate: float = Field(ge=0.0, le=1.0, allow_inf_nan=False)
+    average_score: float = Field(allow_inf_nan=False)
     cases: list[CaseEvaluation]
 
 
@@ -152,24 +210,24 @@ class CaseDelta(_StrictModel):
     status: DeltaStatus
     baseline_passed: bool
     candidate_passed: bool
-    baseline_score: float
-    candidate_score: float
-    score_delta: float
+    baseline_score: float = Field(allow_inf_nan=False)
+    candidate_score: float = Field(allow_inf_nan=False)
+    score_delta: float = Field(allow_inf_nan=False)
 
 
 class PairedConfidenceInterval(_StrictModel):
-    point_estimate: float
-    lower: float
-    upper: float
-    confidence_level: float
-    bootstrap_samples: int
+    point_estimate: float = Field(allow_inf_nan=False)
+    lower: float = Field(allow_inf_nan=False)
+    upper: float = Field(allow_inf_nan=False)
+    confidence_level: float = Field(gt=0.0, lt=1.0, allow_inf_nan=False)
+    bootstrap_samples: int = Field(ge=1)
     seed: int
 
 
 class SplitDelta(_StrictModel):
     split: Literal["train", "validation"]
-    pass_rate_delta: float
-    average_score_delta: float
+    pass_rate_delta: float = Field(allow_inf_nan=False)
+    average_score_delta: float = Field(allow_inf_nan=False)
     paired_pass_rate_ci: PairedConfidenceInterval
     newly_passed: list[str] = Field(default_factory=list)
     newly_failed: list[str] = Field(default_factory=list)
@@ -200,15 +258,15 @@ class GateDecision(_StrictModel):
 
 
 class ResourceUsage(_StrictModel):
-    metric_calls: int = 0
-    reflection_calls: int = 0
-    judge_calls: Optional[int] = None
-    prompt_tokens: int = 0
-    completion_tokens: int = 0
-    total_tokens: int = 0
-    cost_usd: Optional[float] = None
-    duration_seconds: float = 0.0
-    p95_latency_ms: Optional[float] = None
+    metric_calls: int = Field(default=0, ge=0)
+    reflection_calls: int = Field(default=0, ge=0)
+    judge_calls: Optional[int] = Field(default=None, ge=0)
+    prompt_tokens: int = Field(default=0, ge=0)
+    completion_tokens: int = Field(default=0, ge=0)
+    total_tokens: int = Field(default=0, ge=0)
+    cost_usd: Optional[float] = Field(default=None, ge=0.0, allow_inf_nan=False)
+    duration_seconds: float = Field(default=0.0, ge=0.0, allow_inf_nan=False)
+    p95_latency_ms: Optional[float] = Field(default=None, ge=0.0, allow_inf_nan=False)
     cost_measurement: str = "unavailable"
 
 
@@ -236,17 +294,17 @@ class OptimizerAudit(_StrictModel):
     status: str
     stop_reason: Optional[str] = None
     used_agent_optimizer: bool
-    baseline_pass_rate: float
-    best_pass_rate: float
-    rounds: int
+    baseline_pass_rate: float = Field(ge=0.0, le=1.0, allow_inf_nan=False)
+    best_pass_rate: float = Field(ge=0.0, le=1.0, allow_inf_nan=False)
+    rounds: int = Field(ge=0)
     resources: ResourceUsage
     artifact_dir: str
 
 
 class DataQualityAudit(_StrictModel):
     passed: bool
-    train_cases: int
-    validation_cases: int
+    train_cases: int = Field(ge=0)
+    validation_cases: int = Field(ge=0)
     duplicate_ids: list[str] = Field(default_factory=list)
     cross_split_duplicates: list[str] = Field(default_factory=list)
     near_cross_split_duplicates: list[str] = Field(default_factory=list)
@@ -257,7 +315,7 @@ class RunAudit(_StrictModel):
     run_id: str
     started_at: str
     finished_at: str
-    duration_seconds: float
+    duration_seconds: float = Field(ge=0.0, allow_inf_nan=False)
     seed: int
     config_sha256: str
     train_sha256: str
@@ -267,9 +325,9 @@ class RunAudit(_StrictModel):
 
 
 class FailureAttributionSummary(_StrictModel):
-    explained_failed_cases: int
-    total_failed_cases: int
-    coverage_rate: float
+    explained_failed_cases: int = Field(ge=0)
+    total_failed_cases: int = Field(ge=0)
+    coverage_rate: float = Field(ge=0.0, le=1.0, allow_inf_nan=False)
     category_counts: dict[str, int] = Field(default_factory=dict)
     by_case: dict[str, list[FailureReason]] = Field(default_factory=dict)
 

examples/optimization/eval_optimize_loop/loop/offline.py

@@ -93,6 +93,12 @@ def configure(
                 if not conversation:
                     continue
                 query = _content_text(conversation[0].user_content)
+                if not query:
+                    raise ValueError(f"case {case.eval_id!r} has an empty offline replay query")
+                if query in catalog:
+                    previous = catalog[query]
+                    raise ValueError("offline replay queries must be unique; "
+                                     f"cases {previous.eval_id!r} and {case.eval_id!r} share the same query")
                 catalog[query] = case
         with cls._configuration_lock:
             cls._catalog = catalog