fix(evaluation): 补齐异常脱敏与 Trace 错误上下文

Author: guocfu

examples/optimization/eval_optimize_loop/pipeline.py

@@ -1043,24 +1043,37 @@ async def _execute_trace_stage(
         phase="candidate", split="validation",
     )
     progress.enter("analysis")
-    analysis = build_evaluation_analysis(
-        baseline_train=baseline_train,
-        baseline_validation=baseline_validation,
-        candidate_train=candidate_train,
-        candidate_validation=candidate_validation,
-        hard_case_ids=set(prepared.config.case_labels.hard_case_ids),
-        critical_case_ids=set(prepared.config.case_labels.critical_case_ids),
-        severe_case_score_drop=prepared.config.gate.severe_case_score_drop,
-    )
+    try:
+        analysis = build_evaluation_analysis(
+            baseline_train=baseline_train,
+            baseline_validation=baseline_validation,
+            candidate_train=candidate_train,
+            candidate_validation=candidate_validation,
+            hard_case_ids=set(prepared.config.case_labels.hard_case_ids),
+            critical_case_ids=set(prepared.config.case_labels.critical_case_ids),
+            severe_case_score_drop=prepared.config.gate.severe_case_score_drop,
+        )
+    except EvaluationAnalysisError as exc:
+        raise PipelineStageExecutionError(
+            f"stage 3a analysis failed: {exc}"
+        ) from exc
     measurements = ResourceMeasurements(
         cost_usd=ObservableValue(status="unavailable", unit="USD", reason="Trace replay does not call a model."),
         total_tokens=ObservableValue(status="unavailable", unit="tokens", reason="Trace replay does not call a model."),
         duration_seconds=ObservableValue(status="available", value=perf_counter() - started_at, unit="seconds"),
     )
     progress.enter("gate")
-    gate_decision = evaluate_gate(
-        analysis, prepared.config.gate, prepared.config.budget, measurements
-    )
+    try:
+        gate_decision = evaluate_gate(
+            analysis,
+            prepared.config.gate,
+            prepared.config.budget,
+            measurements,
+        )
+    except GateEvaluationError as exc:
+        raise PipelineStageExecutionError(
+            f"stage 3b gate failed: {exc}"
+        ) from exc
     progress.enter("writeback")
     writeback = WritebackResult(
         status="skipped", reason="trace_replay", attempted=False

examples/optimization/eval_optimize_loop/report_builder.py

@@ -50,7 +50,8 @@ def _not_applicable_optimizer_value(
     )
 
 
-def _redact_error_message(error: Exception) -> str:
+def redact_error_message(error: Exception) -> str:
+    """移除异常文本中的环境凭据、认证字段和连接地址。"""
     message = str(error)
     environment_values = {
         os.environ.get(name, "")
@@ -152,7 +153,7 @@ def build_failure_report(
     return FailureReport(
         run_id=prepared.workspace.run_id, execution_mode=prepared.config.execution.mode,
         failed_phase=progress.current_phase, exception_type=type(error).__name__,
-        error_message=_redact_error_message(error), generated_at=generated_at,
+        error_message=redact_error_message(error), generated_at=generated_at,
         input_snapshot=prepared.input_snapshot,
         source_prompt_hashes=dict(sorted(source_prompt_hashes.items())),
         completed_phases=progress.completed_phases, existing_artifacts=sorted(existing_artifacts),

examples/optimization/eval_optimize_loop/run_real_integration.py

@@ -25,6 +25,7 @@
     from examples.optimization.eval_optimize_loop.real_agent import BusinessModelConfig
     from examples.optimization.eval_optimize_loop.real_agent import RealBusinessAgent
     from examples.optimization.eval_optimize_loop.real_agent import load_business_model_config
+    from examples.optimization.eval_optimize_loop.report_builder import redact_error_message
     from examples.optimization.eval_optimize_loop.schemas import OptimizerRuntimeParameters
 else:
     from .config import load_pipeline_config
@@ -33,6 +34,7 @@
     from .real_agent import BusinessModelConfig
     from .real_agent import RealBusinessAgent
     from .real_agent import load_business_model_config
+    from .report_builder import redact_error_message
     from .schemas import OptimizerRuntimeParameters
 
 
@@ -116,7 +118,10 @@ def main() -> int:
     try:
         prepared, result = asyncio.run(_run(args, business_config, parameters))
     except Exception as exc:
-        print(f"Real integration failed: {exc}", file=sys.stderr)
+        print(
+            f"Real integration failed: {redact_error_message(exc)}",
+            file=sys.stderr,
+        )
         return 1
 
     print(f"Completed real-model pipeline: {prepared.workspace.run_dir}")

tests/evaluation/test_eval_optimize_loop_real_integration.py

@@ -335,6 +335,40 @@ async def fail(*args, **kwargs):
     assert "offline failure" in capsys.readouterr().err
 
 
+def test_real_cli_redacts_environment_secrets_from_pipeline_error(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+):
+    api_key = "integration-secret-key"
+    base_url = "https://private-gateway.example.test/v1"
+
+    async def fail(*args, **kwargs):
+        raise RuntimeError(
+            f"request failed api_key={api_key} base_url={base_url}"
+        )
+
+    monkeypatch.setenv("TRPC_AGENT_API_KEY", api_key)
+    monkeypatch.setenv("TRPC_AGENT_BASE_URL", base_url)
+    monkeypatch.setenv("TRPC_AGENT_MODEL_NAME", "business-model")
+    monkeypatch.setattr(run_real_integration, "_run", fail)
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            "run_real_integration.py",
+            "--run-real",
+            "--optimizer-model-name",
+            "optimizer-model",
+        ],
+    )
+
+    assert run_real_integration.main() == 1
+    error = capsys.readouterr().err
+    assert api_key not in error
+    assert base_url not in error
+    assert "[REDACTED]" in error
+
+
 def test_real_cli_requires_explicit_confirmation_before_creating_workspace(tmp_path: Path):
     root = _copy_example(tmp_path)
     completed = subprocess.run(

tests/evaluation/test_eval_optimize_loop_stage6_trace.py

@@ -14,6 +14,7 @@
 
 from examples.optimization.eval_optimize_loop import pipeline as pipeline_module
 from examples.optimization.eval_optimize_loop.pipeline import prepare_run
+from examples.optimization.eval_optimize_loop.pipeline import PipelineStageExecutionError
 from examples.optimization.eval_optimize_loop.pipeline import run_trace_stage
 from examples.optimization.eval_optimize_loop.schemas import ArtifactIndex
 from examples.optimization.eval_optimize_loop.schemas import OptimizationReport
@@ -97,3 +98,41 @@ def __init__(self, *_args, **_kwargs):
     artifact_ids = {artifact.artifact_id for artifact in index.artifacts}
     assert "input.trace.candidate_train" in artifact_ids
     assert "input.trace.candidate_validation" in artifact_ids
+
+
+@pytest.mark.parametrize(
+    ("target", "error_type", "expected"),
+    [
+        (
+            "build_evaluation_analysis",
+            pipeline_module.EvaluationAnalysisError,
+            "stage 3a analysis failed",
+        ),
+        (
+            "evaluate_gate",
+            pipeline_module.GateEvaluationError,
+            "stage 3b gate failed",
+        ),
+    ],
+)
+@pytest.mark.asyncio
+async def test_trace_stage_wraps_analysis_and_gate_errors_with_phase_context(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    target: str,
+    error_type: type[Exception],
+    expected: str,
+):
+    root = _copy_example(tmp_path, f"trace_{target}")
+    prepared = prepare_run(
+        root / "pipeline.trace.json",
+        run_id=f"trace_{target}",
+    )
+
+    def fail(*_args, **_kwargs):
+        raise error_type("injected failure")
+
+    monkeypatch.setattr(pipeline_module, target, fail)
+
+    with pytest.raises(PipelineStageExecutionError, match=expected):
+        await run_trace_stage(prepared, scenario="improve")