fix(eval_optimize_loop): address adversarial review findings

Critical fixes:
- _build_split_result: skip empty case_results (all([])=True phantom pass)
- overfitting detection: require val_pass_rate_delta < 0 (not <= 0)
- total_cases: count only evaluable cases, not dict key count
- _EvaluationCasesFailed fallback: warn on ImportError before using AssertionError

Important fixes:
- gate.py: protected_case_ids check both train and val score deltas
- reporting.py: metric breakdown title shows correct split label
- run_pipeline.py: use public output_dir property instead of private _config
- pipeline.py: expose output_dir property, _collect_cost method

Other:
- failure_attribution.py: document format patterns as heuristic defaults

Author: DNKYr

examples/optimization/eval_optimize_loop/failure_attribution.py

@@ -16,6 +16,9 @@
     "llm_rubric_knowledge_recall": "knowledge_recall_insufficient",
 }
 
+# Default format-violation patterns. These are heuristics tuned for the
+# sample evalsets. Real pipelines should make format patterns configurable
+# (e.g. via pipeline.json) to match their expected output format.
 _FORMAT_PATTERNS: list[tuple[str, re.Pattern]] = [
     ("format_violation", re.compile(r"答案：")),
 ]
@@ -45,8 +48,8 @@ def _extract_expected_text(case_results: list[EvalCaseResult]) -> Optional[str]:
 
 def attribute_failures(eval_results: dict[str, list[EvalCaseResult]]) -> FailureAttribution:
     categories: dict[str, FailureCategory] = {}
-    total_cases = len(eval_results)
     failed_cases = 0
+    evaluable_cases = sum(1 for v in eval_results.values() if v)
 
     for case_id, case_results in eval_results.items():
         last_result = case_results[-1] if case_results else None
@@ -84,7 +87,7 @@ def attribute_failures(eval_results: dict[str, list[EvalCaseResult]]) -> Failure
                             categories[sub_cat].case_ids.append(case_id)
 
     return FailureAttribution(
-        total_cases=total_cases,
+        total_cases=evaluable_cases,
         failed_cases=failed_cases,
         categories=categories,
     )

examples/optimization/eval_optimize_loop/gate.py

@@ -14,7 +14,7 @@ def apply_gate(
     overfitting_warning = False
 
     # Rule 1: Overfitting check (always on, warning only)
-    if delta.train_pass_rate_delta > 0 and delta.val_pass_rate_delta <= 0:
+    if delta.train_pass_rate_delta > 0 and delta.val_pass_rate_delta < 0:
         overfitting_warning = True
         reasons.append(
             f"overfitting_warning: train pass_rate improved by {delta.train_pass_rate_delta:.4f} "
@@ -38,11 +38,13 @@ def apply_gate(
     else:
         reasons.append(f"new_fails check passed: allow_new_fails={gate_config.allow_new_fails}, newly_failing={len(delta.val.newly_failing)}")
 
-    # Rule 4: protected_cases
+    # Rule 4: protected_cases (checks both train and val)
     degraded_protected: list[str] = []
     for case_id in gate_config.protected_case_ids:
-        case_scores = delta.val.score_deltas.get(case_id, {})
-        if case_scores and any(v < 0 for v in case_scores.values()):
+        train_scores = delta.train.score_deltas.get(case_id, {})
+        val_scores = delta.val.score_deltas.get(case_id, {})
+        all_scores = {**train_scores, **val_scores}
+        if all_scores and any(v < 0 for v in all_scores.values()):
             degraded_protected.append(case_id)
     if degraded_protected:
         reject_reasons.append(f"protected cases degraded: {degraded_protected}")

examples/optimization/eval_optimize_loop/pipeline.py

@@ -20,6 +20,13 @@
 try:
     from trpc_agent_sdk.evaluation._agent_evaluator import _EvaluationCasesFailed
 except ImportError:
+    import warnings
+    warnings.warn(
+        "Could not import _EvaluationCasesFailed; falling back to AssertionError. "
+        "This may cause the pipeline to silently swallow unrelated assertions. "
+        "Update trpc_agent_sdk to the latest version.",
+        RuntimeWarning,
+    )
     _EvaluationCasesFailed = AssertionError
 
 from .delta import compute_delta
@@ -163,12 +170,9 @@ async def _run_live() -> tuple:
         delta = compute_delta(baseline_split, candidate_split)
 
         duration = time.monotonic() - t0
-        # Note: cost_usd is always 0.0 here because the pipeline does not
-        # yet collect per-round LLM costs from the optimizer.  The gate's
-        # max_cost_usd rule will therefore not reject on cost grounds.
-        # Users should monitor costs via their LLM provider dashboard.
+        cost_usd = self._collect_cost()
         gate = apply_gate(
-            delta, self._config.gate, cost_usd=0.0, duration_seconds=duration
+            delta, self._config.gate, cost_usd=cost_usd, duration_seconds=duration
         )
 
         finished_at = datetime.now(timezone.utc).isoformat()
@@ -313,6 +317,8 @@ def _build_split_result(self, eval_result: EvaluateResult) -> SplitResult:
         passed_count = 0
 
         for case_id, case_results in cases_by_id.items():
+            if not case_results:
+                continue
             all_passed = all(
                 cr.final_eval_status == EvalStatus.PASSED and not cr.error_message
                 for cr in case_results
@@ -354,3 +360,13 @@ def _build_split_result(self, eval_result: EvaluateResult) -> SplitResult:
             metric_breakdown=metric_breakdown,
             per_case=per_case,
         )
+
+    @property
+    def output_dir(self) -> str:
+        return self._config.output_dir
+
+    def _collect_cost(self) -> float:
+        # Cost tracking from AgentOptimizer is not yet implemented.
+        # The gate's max_cost_usd rule is therefore dormant — users
+        # should monitor costs via their LLM provider dashboard.
+        return 0.0

examples/optimization/eval_optimize_loop/reporting.py

@@ -60,8 +60,9 @@ def _build_markdown(result: PipelineResult) -> list[str]:
     if result.baseline:
         base_ref = result.baseline.get("val") or result.baseline.get("train")
         cand_ref = result.candidate.get("val") or result.candidate.get("train")
+        split_label = "val" if result.baseline.get("val") else "train"
         if base_ref and base_ref.metric_breakdown:
-            lines.append("### Metric Breakdown (val)")
+            lines.append(f"### Metric Breakdown ({split_label})")
             lines.append("")
             lines.append("| Metric | Baseline | Candidate | Delta |")
             lines.append("|---|---|---|---|")

examples/optimization/eval_optimize_loop/run_pipeline.py

@@ -38,7 +38,7 @@ async def main() -> None:
     pipeline = EvalOptimizePipeline.from_config(args.pipeline_config)
     result = await pipeline.run()
 
-    output_dir = Path(pipeline._config.output_dir).resolve()
+    output_dir = Path(pipeline.output_dir).resolve()
     print(f"\nPipeline complete: {result.gate_decision}")
     print(f"  Duration: {result.duration_seconds:.2f}s")
     print(f"  Reports: {output_dir}/")

examples/optimization/eval_optimize_loop/tests/test_failure_attribution.py

@@ -143,7 +143,7 @@ def test_attribute_failures_empty_case_results():
         ],
     }
     attr = attribute_failures(results)
-    assert attr.total_cases == 2
+    assert attr.total_cases == 1
     assert attr.failed_cases == 1
     assert "final_response_mismatch" in attr.categories
     assert "case_b" in attr.categories["final_response_mismatch"].case_ids

examples/optimization/eval_optimize_loop/tests/test_gate.py

@@ -89,14 +89,22 @@ def test_gate_reject_duration_over_budget():
 
 
 def test_gate_overfitting_warning():
-    """Train improves but val does not → overfitting_warning=True."""
-    delta = _make_delta(train_pass_rate_delta=0.5, val_pass_rate_delta=0.0)
+    """Train improves but val degrades → overfitting_warning=True."""
+    delta = _make_delta(train_pass_rate_delta=0.5, val_pass_rate_delta=-0.1)
     gate = GateConfig(min_improvement=0.0, allow_new_fails=False)
     decision = apply_gate(delta, gate, cost_usd=0.0, duration_seconds=10.0)
     assert decision.overfitting_warning is True
     assert any("overfitting" in r.lower() for r in decision.reasons)
 
 
+def test_gate_no_overfitting_when_val_flat():
+    """Train improves but val is flat (0.0) -> no overfitting warning."""
+    delta = _make_delta(train_pass_rate_delta=0.5, val_pass_rate_delta=0.0)
+    gate = GateConfig(min_improvement=0.0, allow_new_fails=False)
+    decision = apply_gate(delta, gate, cost_usd=0.0, duration_seconds=10.0)
+    assert decision.overfitting_warning is False
+
+
 def test_gate_reject_overfitting_with_degradation():
     """Train improves but val degrades → overfitting_warning + REJECT (below min_improvement)."""
     delta = _make_delta(train_pass_rate_delta=0.5, val_pass_rate_delta=-0.2)