diff --git a/examples/optimization/eval_optimize_loop/DESIGN.md b/examples/optimization/eval_optimize_loop/DESIGN.md
new file mode 100644
index 000000000..e05471436
--- /dev/null
+++ b/examples/optimization/eval_optimize_loop/DESIGN.md
@@ -0,0 +1,49 @@
+# Evaluation + Optimization 闭环设计
+
+## 方案说明
+
+本示例复用 `AgentEvaluator`、`AgentOptimizer` 和 `TargetPrompt`，不修改生产
+源码。Pipeline 先分别评测 train/validation，保存每条 case 的 metric、状态、
+失败原因和关键轨迹；再按执行异常、回复不匹配、工具名称、工具参数、rubric、
+知识召回和格式问题进行确定性归因。优化阶段只修改 working copy，候选完成
+train/validation 回放后，按 case 输出新增通过、新增失败、分数提升、分数下降
+和 unchanged。Gate 检查验证集提升阈值、无新增 hard fail、critical case 不退化、
+验证集退化和成本/耗时预算；训练提升而验证退化直接判定过拟合。fake-model、
+fake-judge 和 trace mode 使用同一比较与 gate 链路，保证无 API Key 也能复现。
+报告 JSON/Markdown 保存输入 hash、候选、逐 case delta、归因、成本、耗时和理由；
+默认不回写源 prompt；仅 `real` 模式显式 `--write-back` 且 gate 接受时才写回，
+fake/trace 模式拒绝回写，避免合成候选污染检入文件。
+
+## 阶段与 Review
+
+- A：模型、输入校验和 evaluator 适配；Review A 检查结果保留、泄漏和边界。
+- B：优化、归因、逐 case diff、gate 和 working-copy；Review B 检查过拟合、
+  hard fail、成本和 prompt 恢复。
+- C：CLI、fake/trace、报告和样例；Review C 对照 Issue #91 核对交付物。
+- D：目标测试、覆盖率、flake8、函数复杂度和最终 diff；Review D 做交付审查。
+
+## 主要文件
+
+实现：`loop/models.py`、`loop/evaluation.py`、`loop/analysis.py`、
+`loop/pipeline.py`、`loop/reporting.py`、`agent/agent.py`、`run_pipeline.py`。
+
+资源：`data/train.evalset.json`、`data/val.evalset.json`、
+`data/fake_trace.json`、`optimizer.json`、`gate.json`、
+`optimization_report.json`、`README.md`。
+
+`optimization_report.json` 是 Issue #91 要求的示例输出，不是稳定契约。
+其中时间戳、Git SHA、Python 版本和耗时仅展示审计字段，实际运行由 pipeline
+在输出目录重新生成，测试不依赖这些环境相关值。
+
+测试：`tests/evaluation/test_eval_optimize_loop_*.py`。
+
+## 验收
+
+```bash
+uv run pytest tests/evaluation/test_eval_optimize_loop_*.py \
+  --cov=examples.optimization.eval_optimize_loop --cov-fail-under=90
+uv run flake8 --max-complexity=15 --max-line-length=120 \
+  examples/optimization/eval_optimize_loop
+uv run python examples/optimization/eval_optimize_loop/run_pipeline.py \
+  --fake-model --fake-judge
+```
diff --git a/examples/optimization/eval_optimize_loop/README.md b/examples/optimization/eval_optimize_loop/README.md
new file mode 100644
index 000000000..90906d6c1
--- /dev/null
+++ b/examples/optimization/eval_optimize_loop/README.md
@@ -0,0 +1,16 @@
+# Evaluation + Optimization Loop
+
+本示例把 `AgentEvaluator`、`AgentOptimizer` 和 `TargetPrompt` 组合成可审计的
+baseline → optimize → candidate → gate 闭环。默认 `fake-model` 不需要 API Key，
+运行时间通常小于 30 秒；`real` 模式读取 `TRPC_AGENT_API_KEY`、
+`TRPC_AGENT_BASE_URL` 和 `TRPC_AGENT_MODEL_NAME`（也可用 `--model-name` 覆盖）。
+`trace` 模式读取预录制的
+baseline/candidate evalset，适合离线回归。
+
+```bash
+uv run python examples/optimization/eval_optimize_loop/run_pipeline.py
+```
+
+输出目录包含 `optimization_report.json`、`optimization_report.md` 和临时工作副本。
+只有 `real` 模式显式传入 `--write-back` 且 gate 接受时才会更新 prompt 源文件；
+fake/trace 模式会拒绝回写，避免合成候选污染源文件。
diff --git a/examples/optimization/eval_optimize_loop/agent/agent.py b/examples/optimization/eval_optimize_loop/agent/agent.py
new file mode 100644
index 000000000..e66166845
--- /dev/null
+++ b/examples/optimization/eval_optimize_loop/agent/agent.py
@@ -0,0 +1,88 @@
+"""Real and deterministic agents used by the optimization loop example."""
+
+from __future__ import annotations
+
+import os
+import uuid
+from pathlib import Path
+
+from trpc_agent_sdk.agents import LlmAgent
+from trpc_agent_sdk.models import OpenAIModel
+from trpc_agent_sdk.runners import Runner
+from trpc_agent_sdk.sessions import InMemorySessionService
+from trpc_agent_sdk.types import Content
+from trpc_agent_sdk.types import Part
+
+APP_NAME = "eval_optimize_loop_agent"
+USER_ID = "eval-optimize-loop"
+CANDIDATE_MARKER = "OPTIMIZED_CANDIDATE"
+UNKNOWN_RESPONSE = '{"queue":"unknown"}'
+
+
+async def fake_call_agent(prompt_path: Path, query: str) -> str:
+    """Return a prompt-sensitive deterministic response without an API key."""
+    prompt = prompt_path.read_text(encoding="utf-8")
+    if CANDIDATE_MARKER in prompt and ("1006" in query or "download" in query):
+        return '{"queue":"billing"}'
+    if "refund" in query and CANDIDATE_MARKER not in prompt:
+        return UNKNOWN_RESPONSE
+    if "download" in query:
+        return UNKNOWN_RESPONSE
+    return _expected_queue(query)
+
+
+async def real_call_agent(prompt_path: Path, query: str) -> str:
+    """Run one real OpenAI-compatible Agent invocation."""
+    agent = _create_agent(prompt_path)
+    sessions = InMemorySessionService()
+    runner = Runner(app_name=APP_NAME, agent=agent, session_service=sessions)
+    session_id = uuid.uuid4().hex
+    await sessions.create_session(
+        app_name=APP_NAME,
+        user_id=USER_ID,
+        session_id=session_id,
+        state={},
+    )
+    message = Content(role="user", parts=[Part.from_text(text=query)])
+    return await _consume_final_text(runner, session_id, message)
+
+
+def _create_agent(prompt_path: Path) -> LlmAgent:
+    api_key = _required_env("TRPC_AGENT_API_KEY")
+    base_url = _required_env("TRPC_AGENT_BASE_URL")
+    model_name = _required_env("TRPC_AGENT_MODEL_NAME")
+    model = OpenAIModel(model_name=model_name, api_key=api_key, base_url=base_url)
+    return LlmAgent(
+        name=APP_NAME,
+        description="Support queue classifier.",
+        model=model,
+        instruction=prompt_path.read_text(encoding="utf-8"),
+    )
+
+
+def _required_env(name: str) -> str:
+    value = os.getenv(name)
+    if not value:
+        raise RuntimeError(f"required environment variable is missing: {name}")
+    return value
+
+
+async def _consume_final_text(runner: Runner, session_id: str, message: Content) -> str:
+    output = []
+    async for event in runner.run_async(
+            user_id=USER_ID,
+            session_id=session_id,
+            new_message=message,
+    ):
+        if not event.is_final_response() or not event.content:
+            continue
+        output.extend(part.text or "" for part in event.content.parts or [] if not part.thought)
+    return "".join(output).strip()
+
+
+def _expected_queue(query: str) -> str:
+    if "invoice" in query or "payment" in query or "refund" in query:
+        return '{"queue":"billing"}'
+    if "password" in query:
+        return '{"queue":"account"}'
+    return '{"queue":"technical"}'
diff --git a/examples/optimization/eval_optimize_loop/agent/prompts/system.md b/examples/optimization/eval_optimize_loop/agent/prompts/system.md
new file mode 100644
index 000000000..ef6767a70
--- /dev/null
+++ b/examples/optimization/eval_optimize_loop/agent/prompts/system.md
@@ -0,0 +1,3 @@
+Classify support requests.
+
+Return a short answer containing the selected queue.
diff --git a/examples/optimization/eval_optimize_loop/data/fake_trace.json b/examples/optimization/eval_optimize_loop/data/fake_trace.json
new file mode 100644
index 000000000..9240a0fb1
--- /dev/null
+++ b/examples/optimization/eval_optimize_loop/data/fake_trace.json
@@ -0,0 +1,54 @@
+{
+  "baseline": {
+    "train": [
+      {
+        "eval_id": "trace_train_1",
+        "eval_mode": "trace",
+        "conversation": [
+          {
+            "user_content": {"role": "user", "parts": [{"text": "trace invoice"}]},
+            "final_response": {"role": "model", "parts": [{"text": "{\"queue\":\"billing\"}"}]}
+          }
+        ]
+      }
+    ],
+    "validation": [
+      {
+        "eval_id": "trace_validation_1",
+        "eval_mode": "trace",
+        "conversation": [
+          {
+            "user_content": {"role": "user", "parts": [{"text": "trace password"}]},
+            "final_response": {"role": "model", "parts": [{"text": "{\"queue\":\"account\"}"}]}
+          }
+        ]
+      }
+    ]
+  },
+  "candidate": {
+    "train": [
+      {
+        "eval_id": "trace_train_1",
+        "eval_mode": "trace",
+        "conversation": [
+          {
+            "user_content": {"role": "user", "parts": [{"text": "trace invoice"}]},
+            "final_response": {"role": "model", "parts": [{"text": "{\"queue\":\"billing\"}"}]}
+          }
+        ]
+      }
+    ],
+    "validation": [
+      {
+        "eval_id": "trace_validation_1",
+        "eval_mode": "trace",
+        "conversation": [
+          {
+            "user_content": {"role": "user", "parts": [{"text": "trace password"}]},
+            "final_response": {"role": "model", "parts": [{"text": "{\"queue\":\"account\"}"}]}
+          }
+        ]
+      }
+    ]
+  }
+}
diff --git a/examples/optimization/eval_optimize_loop/data/train.evalset.json b/examples/optimization/eval_optimize_loop/data/train.evalset.json
new file mode 100644
index 000000000..44b10472c
--- /dev/null
+++ b/examples/optimization/eval_optimize_loop/data/train.evalset.json
@@ -0,0 +1,54 @@
+{
+  "eval_set_id": "eval_optimize_train",
+  "name": "Evaluation optimization loop training cases",
+  "eval_cases": [
+    {
+      "eval_id": "train_billing_format",
+      "conversation": [
+        {
+          "invocation_id": "train-1",
+          "user_content": {
+            "role": "user",
+            "parts": [{"text": "I was charged twice for invoice 42."}]
+          },
+          "final_response": {
+            "role": "model",
+            "parts": [{"text": "{\"queue\":\"billing\"}"}]
+          }
+        }
+      ]
+    },
+    {
+      "eval_id": "train_account_recall",
+      "conversation": [
+        {
+          "invocation_id": "train-2",
+          "user_content": {
+            "role": "user",
+            "parts": [{"text": "I cannot reset my account password."}]
+          },
+          "final_response": {
+            "role": "model",
+            "parts": [{"text": "{\"queue\":\"account\"}"}]
+          }
+        }
+      ]
+    },
+    {
+      "eval_id": "train_technical_router",
+      "conversation": [
+        {
+          "invocation_id": "train-3",
+          "user_content": {
+            "role": "user",
+            "parts": [{"text": "The SDK times out while opening a stream."}]
+          },
+          "final_response": {
+            "role": "model",
+            "parts": [{"text": "{\"queue\":\"technical\"}"}]
+          }
+        }
+      ]
+    }
+  ]
+}
diff --git a/examples/optimization/eval_optimize_loop/data/val.evalset.json b/examples/optimization/eval_optimize_loop/data/val.evalset.json
new file mode 100644
index 000000000..8f901f63c
--- /dev/null
+++ b/examples/optimization/eval_optimize_loop/data/val.evalset.json
@@ -0,0 +1,54 @@
+{
+  "eval_set_id": "eval_optimize_validation",
+  "name": "Evaluation optimization loop validation cases",
+  "eval_cases": [
+    {
+      "eval_id": "val_candidate_improves",
+      "conversation": [
+        {
+          "invocation_id": "validation-1",
+          "user_content": {
+            "role": "user",
+            "parts": [{"text": "Please refund the duplicate subscription payment."}]
+          },
+          "final_response": {
+            "role": "model",
+            "parts": [{"text": "{\"queue\":\"billing\"}"}]
+          }
+        }
+      ]
+    },
+    {
+      "eval_id": "val_candidate_no_effect",
+      "conversation": [
+        {
+          "invocation_id": "validation-2",
+          "user_content": {
+            "role": "user",
+            "parts": [{"text": "Where can I download last year's invoices?"}]
+          },
+          "final_response": {
+            "role": "model",
+            "parts": [{"text": "{\"queue\":\"billing\"}"}]
+          }
+        }
+      ]
+    },
+    {
+      "eval_id": "val_candidate_regresses",
+      "conversation": [
+        {
+          "invocation_id": "validation-3",
+          "user_content": {
+            "role": "user",
+            "parts": [{"text": "My API connection closes with error 1006."}]
+          },
+          "final_response": {
+            "role": "model",
+            "parts": [{"text": "{\"queue\":\"technical\"}"}]
+          }
+        }
+      ]
+    }
+  ]
+}
diff --git a/examples/optimization/eval_optimize_loop/gate.json b/examples/optimization/eval_optimize_loop/gate.json
new file mode 100644
index 000000000..b494eec4c
--- /dev/null
+++ b/examples/optimization/eval_optimize_loop/gate.json
@@ -0,0 +1,12 @@
+{
+  "primary_metric": "final_response_avg_score",
+  "min_score_delta": 0.1,
+  "critical_case_ids": [],
+  "max_critical_regression": 0.0,
+  "hard_case_ids": [],
+  "hard_metric_names": [],
+  "max_total_cost": null,
+  "max_duration_seconds": 180.0,
+  "train_epsilon": 0.000001,
+  "validation_epsilon": 0.000001
+}
diff --git a/examples/optimization/eval_optimize_loop/loop/analysis.py b/examples/optimization/eval_optimize_loop/loop/analysis.py
new file mode 100644
index 000000000..42b9f47d5
--- /dev/null
+++ b/examples/optimization/eval_optimize_loop/loop/analysis.py
@@ -0,0 +1,278 @@
+"""Failure attribution, case deltas, and acceptance gate logic."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Iterable
+
+from .models import Attribution
+from .models import CaseDelta
+from .models import CaseSnapshot
+from .models import ChangeKind
+from .models import CostSummary
+from .models import EvaluationSnapshot
+from .models import FailureCategory
+from .models import GateCheck
+from .models import GateConfig
+from .models import GateDecision
+from .models import SCORE_EPSILON
+from .models import SplitDelta
+
+FORMAT_MARKERS = ("format", "json", "schema", "parse")
+RUBRIC_MARKERS = ("rubric", "judge", "quality")
+KNOWLEDGE_MARKERS = ("knowledge", "recall", "retriev")
+
+
+def attribute_cases(snapshot: EvaluationSnapshot) -> tuple[list[Attribution], dict[FailureCategory, int]]:
+    """Attribute all failed cases with one deterministic primary rule."""
+    attributions = [attribute_case(case, snapshot.primary_metric) for case in snapshot.cases if not case.passed]
+    counts: dict[FailureCategory, int] = {}
+    for attribution in attributions:
+        counts[attribution.category] = counts.get(attribution.category, 0) + 1
+    return attributions, counts
+
+
+def attribute_case(case: CaseSnapshot, primary_metric: str) -> Attribution:
+    """Return an explainable category and evidence for one failed case."""
+    if case.hard_failure or case.error_message:
+        return Attribution(
+            case_id=case.case_id,
+            category=FailureCategory.EXECUTION,
+            rule_id="execution.error",
+            evidence=case.error_message or "hard failure",
+        )
+    if _has_tool_mismatch(case):
+        return Attribution(
+            case_id=case.case_id,
+            category=FailureCategory.TOOL_CALL,
+            rule_id="trajectory.tool_name",
+            evidence="actual and expected tool names differ",
+        )
+    if _has_argument_mismatch(case):
+        return Attribution(
+            case_id=case.case_id,
+            category=FailureCategory.TOOL_ARGUMENT,
+            rule_id="trajectory.arguments",
+            evidence="tool names match but arguments differ",
+        )
+    reason_text = " ".join(case.reasons.values()).lower()
+    if _contains_any(reason_text, RUBRIC_MARKERS):
+        return Attribution(case.case_id, FailureCategory.RUBRIC, "metric.rubric", reason_text)
+    if _contains_any(reason_text, KNOWLEDGE_MARKERS):
+        return Attribution(case.case_id, FailureCategory.KNOWLEDGE, "metric.knowledge", reason_text)
+    if _contains_any(reason_text, FORMAT_MARKERS):
+        return Attribution(case.case_id, FailureCategory.FORMAT, "response.format", reason_text)
+    if case.metric_statuses.get(primary_metric) == "FAILED":
+        return Attribution(
+            case_id=case.case_id,
+            category=FailureCategory.RESPONSE,
+            rule_id="response.mismatch",
+            evidence="final response metric did not meet threshold",
+        )
+    return Attribution(
+        case_id=case.case_id,
+        category=FailureCategory.OTHER,
+        rule_id="fallback.other",
+        evidence="no specific rule matched",
+    )
+
+
+def compare_snapshots(
+    baseline: EvaluationSnapshot,
+    candidate: EvaluationSnapshot,
+) -> SplitDelta:
+    """Compare baseline and candidate case outcomes for one split."""
+    baseline_cases = {case.case_id: case for case in baseline.cases}
+    candidate_cases = {case.case_id: case for case in candidate.cases}
+    case_ids = sorted(set(baseline_cases) | set(candidate_cases))
+    deltas = [
+        _case_delta(case_id, baseline_cases.get(case_id), candidate_cases.get(case_id), baseline.primary_metric)
+        for case_id in case_ids
+    ]
+    return SplitDelta(
+        split=baseline.split,
+        baseline_score=baseline.primary_score,
+        candidate_score=candidate.primary_score,
+        score_delta=_score_delta(baseline.primary_score, candidate.primary_score),
+        cases=deltas,
+    )
+
+
+@dataclass(frozen=True)
+class GateInput:
+    """Inputs for one gate evaluation."""
+
+    config: GateConfig
+    train_delta: SplitDelta
+    validation_delta: SplitDelta
+    baseline_validation: EvaluationSnapshot
+    candidate_validation: EvaluationSnapshot
+    cost: CostSummary
+    duration_seconds: float
+
+
+def evaluate_gate(inputs: GateInput) -> GateDecision:
+    """Apply all configured acceptance predicates with fail-closed semantics."""
+    checks = [
+        _score_check(inputs.validation_delta, inputs.config),
+        _regression_check(inputs.validation_delta),
+        _hard_failure_check(inputs),
+        _critical_check(inputs),
+        _cost_check(inputs.cost, inputs.config),
+        _duration_check(inputs.duration_seconds, inputs.config),
+    ]
+    overfitting = _is_overfitting(inputs.train_delta, inputs.validation_delta, inputs.config)
+    checks.append(
+        GateCheck(
+            name="overfitting",
+            passed=not overfitting,
+            reason="train improved while validation regressed" if overfitting else "no overfitting signal",
+        ))
+    reasons = [check.reason for check in checks if not check.passed]
+    return GateDecision(accepted=not reasons, overfitting=overfitting, checks=checks, reasons=reasons)
+
+
+def _case_delta(
+    case_id: str,
+    baseline: CaseSnapshot | None,
+    candidate: CaseSnapshot | None,
+    primary_metric: str,
+) -> CaseDelta:
+    baseline_score = baseline.metric_scores.get(primary_metric) if baseline else None
+    candidate_score = candidate.metric_scores.get(primary_metric) if candidate else None
+    baseline_passed = baseline.passed if baseline else False
+    candidate_passed = candidate.passed if candidate else False
+    change = _change_kind(baseline_passed, candidate_passed, baseline_score, candidate_score)
+    return CaseDelta(
+        case_id=case_id,
+        baseline_passed=baseline_passed,
+        candidate_passed=candidate_passed,
+        baseline_score=baseline_score,
+        candidate_score=candidate_score,
+        score_delta=_score_delta(baseline_score, candidate_score),
+        change=change,
+        hard_failure_added=bool(candidate is None
+                                or (candidate.hard_failure and not (baseline and baseline.hard_failure))),
+    )
+
+
+def _change_kind(
+    baseline_passed: bool,
+    candidate_passed: bool,
+    baseline_score: float | None,
+    candidate_score: float | None,
+) -> ChangeKind:
+    if not baseline_passed and candidate_passed:
+        return ChangeKind.NEW_PASS
+    if baseline_passed and not candidate_passed:
+        return ChangeKind.NEW_FAIL
+    delta = _score_delta(baseline_score, candidate_score)
+    if delta is None or abs(delta) <= SCORE_EPSILON:
+        return ChangeKind.UNCHANGED
+    return ChangeKind.IMPROVED if delta > 0 else ChangeKind.REGRESSED
+
+
+def _score_delta(baseline: float | None, candidate: float | None) -> float | None:
+    if baseline is None or candidate is None:
+        return None
+    return candidate - baseline
+
+
+def _score_check(delta: SplitDelta, config: GateConfig) -> GateCheck:
+    value = delta.score_delta
+    passed = value is not None and value + SCORE_EPSILON >= config.min_score_delta
+    reason = f"validation delta={value!r}, required>={config.min_score_delta}"
+    return GateCheck(name="validation_score", passed=passed, reason=reason)
+
+
+def _regression_check(delta: SplitDelta) -> GateCheck:
+    value = delta.score_delta
+    passed = value is not None and value >= -SCORE_EPSILON
+    reason = "validation did not regress" if passed else f"validation regression={value!r}"
+    return GateCheck(name="validation_regression", passed=passed, reason=reason)
+
+
+def _hard_failure_check(inputs: GateInput) -> GateCheck:
+    added = [case.case_id for case in inputs.validation_delta.cases if case.hard_failure_added]
+    candidate_cases = {case.case_id: case for case in inputs.candidate_validation.cases}
+    configured = [
+        case_id for case_id in inputs.config.hard_case_ids
+        if case_id not in candidate_cases or not candidate_cases[case_id].passed
+    ]
+    metric_failures = [
+        case.case_id for case in inputs.candidate_validation.cases if any(
+            case.metric_statuses.get(metric) != "PASSED" for metric in inputs.config.hard_metric_names)
+    ]
+    failures = sorted(set(added + configured + metric_failures))
+    return GateCheck(
+        name="hard_failures",
+        passed=not failures,
+        reason="no hard failures" if not failures else f"hard failures: {failures}",
+    )
+
+
+def _critical_check(inputs: GateInput) -> GateCheck:
+    baseline = {case.case_id: case for case in inputs.baseline_validation.cases}
+    candidate = {case.case_id: case for case in inputs.candidate_validation.cases}
+    failures = []
+    for case_id in inputs.config.critical_case_ids:
+        before = baseline.get(case_id)
+        after = candidate.get(case_id)
+        if before is None or after is None:
+            failures.append(case_id)
+            continue
+        score_delta = _score_delta(
+            before.metric_scores.get(inputs.config.primary_metric),
+            after.metric_scores.get(inputs.config.primary_metric),
+        )
+        if not after.passed or (score_delta is not None and score_delta < -inputs.config.max_critical_regression):
+            failures.append(case_id)
+    return GateCheck(
+        name="critical_cases",
+        passed=not failures,
+        reason="critical cases preserved" if not failures else f"critical regressions: {failures}",
+    )
+
+
+def _cost_check(cost: CostSummary, config: GateConfig) -> GateCheck:
+    if config.max_total_cost is None:
+        return GateCheck(name="cost", passed=True, reason="cost budget disabled")
+    passed = cost.cost_complete and cost.total_cost <= config.max_total_cost + SCORE_EPSILON
+    reason = f"cost={cost.total_cost}, complete={cost.cost_complete}, limit={config.max_total_cost}"
+    return GateCheck(name="cost", passed=passed, reason=reason)
+
+
+def _duration_check(duration: float, config: GateConfig) -> GateCheck:
+    if config.max_duration_seconds is None:
+        return GateCheck(name="duration", passed=True, reason="duration budget disabled")
+    passed = duration <= config.max_duration_seconds
+    return GateCheck(
+        name="duration",
+        passed=passed,
+        reason=f"duration={duration:.3f}s, limit={config.max_duration_seconds}s",
+    )
+
+
+def _is_overfitting(train: SplitDelta, validation: SplitDelta, config: GateConfig) -> bool:
+    train_delta = train.score_delta
+    validation_delta = validation.score_delta
+    return (train_delta is not None and validation_delta is not None and train_delta > config.train_epsilon
+            and validation_delta < -config.validation_epsilon)
+
+
+def _has_tool_mismatch(case: CaseSnapshot) -> bool:
+    return [_tool_names(item) for item in case.actual] != [_tool_names(item) for item in case.expected]
+
+
+def _has_argument_mismatch(case: CaseSnapshot) -> bool:
+    actual = [item.tool_calls for item in case.actual]
+    expected = [item.tool_calls for item in case.expected]
+    return bool(actual or expected) and actual != expected
+
+
+def _tool_names(invocation) -> list[str]:
+    return [call.get("name", "") for call in invocation.tool_calls]
+
+
+def _contains_any(value: str, markers: Iterable[str]) -> bool:
+    return any(marker in value for marker in markers)
diff --git a/examples/optimization/eval_optimize_loop/loop/evaluation.py b/examples/optimization/eval_optimize_loop/loop/evaluation.py
new file mode 100644
index 000000000..1d6880df1
--- /dev/null
+++ b/examples/optimization/eval_optimize_loop/loop/evaluation.py
@@ -0,0 +1,354 @@
+"""AgentEvaluator adapter and leakage-safe input validation."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import shutil
+import tempfile
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from statistics import mean
+from typing import Awaitable
+from typing import Callable
+
+from trpc_agent_sdk.evaluation import AgentEvaluator
+from trpc_agent_sdk.evaluation._agent_evaluator import _EvaluationCasesFailed
+from trpc_agent_sdk.evaluation import EvalSet
+from trpc_agent_sdk.evaluation import EvalStatus
+from trpc_agent_sdk.evaluation import OptimizeConfigFile
+from trpc_agent_sdk.evaluation import get_all_tool_calls
+from trpc_agent_sdk.evaluation._eval_result import EvalCaseResult
+from trpc_agent_sdk.evaluation._eval_result import EvaluateResult
+from trpc_agent_sdk.evaluation._optimize_config import load_optimize_config
+
+from .models import CaseSnapshot
+from .models import EvaluationSnapshot
+from .models import GateConfig
+from .models import InputBundle
+from .models import InputPaths
+from .models import InvocationSnapshot
+from .models import SplitName
+
+CallAgent = Callable[[str], Awaitable[str]]
+HASH_NAME = "sha256"
+
+
+@dataclass(frozen=True)
+class EvaluationRequest:
+    """Inputs needed for one evaluator invocation."""
+
+    dataset_path: Path
+    optimizer_path: Path
+    split: SplitName
+    call_agent: CallAgent | None
+    fake_judge: bool = False
+
+
+def load_gate_config(path: Path) -> GateConfig:
+    """Load and validate gate.json."""
+    return GateConfig.model_validate_json(path.read_text(encoding="utf-8"))
+
+
+def load_eval_set(path: Path) -> EvalSet:
+    """Load one evalset using the SDK schema."""
+    return EvalSet.model_validate_json(path.read_text(encoding="utf-8"))
+
+
+def validate_inputs(paths: InputPaths) -> tuple[InputBundle, OptimizeConfigFile, GateConfig]:
+    """Validate paths, schemas, metric references, and split leakage."""
+    input_paths = {
+        "prompt": paths.prompt_path,
+        "train": paths.train_path,
+        "validation": paths.validation_path,
+        "optimizer": paths.optimizer_path,
+        "gate": paths.gate_path,
+    }
+    for path in input_paths.values():
+        if not path.is_file():
+            raise FileNotFoundError(path)
+    if paths.train_path.resolve() == paths.validation_path.resolve():
+        raise ValueError("train and validation paths must differ")
+
+    train_set = load_eval_set(paths.train_path)
+    validation_set = load_eval_set(paths.validation_path)
+    optimizer = load_optimize_config(str(paths.optimizer_path))
+    gate = load_gate_config(paths.gate_path)
+    _validate_split_leakage(train_set, validation_set)
+    _validate_gate_references(gate, optimizer, validation_set)
+    hashes = {name: _file_hash(path) for name, path in input_paths.items()}
+    bundle = InputBundle(
+        prompt_path=paths.prompt_path.resolve(),
+        train_path=paths.train_path.resolve(),
+        validation_path=paths.validation_path.resolve(),
+        optimizer_path=paths.optimizer_path.resolve(),
+        gate_path=paths.gate_path.resolve(),
+        hashes=hashes,
+    )
+    return bundle, optimizer, gate
+
+
+async def evaluate_split(request: EvaluationRequest) -> EvaluationSnapshot:
+    """Run AgentEvaluator and retain results even when cases fail."""
+    config = load_optimize_config(str(request.optimizer_path))
+    started = time.monotonic()
+    with tempfile.TemporaryDirectory(prefix="eval-optimize-") as temp_dir:
+        dataset_path = _dataset_for_sdk(request.dataset_path, Path(temp_dir))
+        metrics_path = Path(temp_dir) / "eval_config.json"
+        metrics_path.write_text(
+            config.evaluate.model_dump_json(by_alias=True),
+            encoding="utf-8",
+        )
+        executor = AgentEvaluator.get_executer(
+            dataset_path,
+            call_agent=request.call_agent,
+            num_runs=config.evaluate.num_runs,
+            print_detailed_results=False,
+            print_summary_report=False,
+            eval_metrics_file_path_or_dir=str(metrics_path),
+            case_parallelism=config.optimize.eval_case_parallelism,
+        )
+        try:
+            await executor.evaluate()
+        except _EvaluationCasesFailed:
+            # SDK reserves this subclass for partial case failures; unrelated
+            # AssertionError instances must propagate to the pipeline failure.
+            if executor.get_result() is None:
+                raise
+        result = executor.get_result()
+    if result is None:
+        raise RuntimeError("AgentEvaluator completed without a result")
+    return _snapshot_result(
+        result,
+        request.split,
+        config,
+        time.monotonic() - started,
+        request.fake_judge,
+    )
+
+
+def _snapshot_result(
+    result: EvaluateResult,
+    split: SplitName,
+    config: OptimizeConfigFile,
+    duration: float,
+    fake_judge: bool,
+) -> EvaluationSnapshot:
+    metrics_config = config.evaluate.get_eval_metrics()
+    primary_metric = metrics_config[0]
+    primary = primary_metric.metric_name
+    grouped: dict[str, list[EvalCaseResult]] = {}
+    for set_result in result.results_by_eval_set_id.values():
+        for case_id, runs in set_result.eval_results_by_eval_id.items():
+            grouped.setdefault(case_id, []).extend(runs)
+    cases = [
+        _snapshot_case(
+            case_id,
+            split,
+            runs,
+            primary,
+            fake_judge,
+            primary_metric.threshold,
+        ) for case_id, runs in sorted(grouped.items())
+    ]
+    metric_names = [metric.metric_name for metric in metrics_config]
+    metrics = {name: _mean_optional(case.metric_scores.get(name) for case in cases) for name in metric_names}
+    primary_score = None if any(case.hard_failure for case in cases) else metrics.get(primary)
+    passed = sum(case.passed for case in cases)
+    return EvaluationSnapshot(
+        split=split,
+        primary_metric=primary,
+        primary_score=primary_score,
+        pass_rate=passed / len(cases) if cases else 0.0,
+        metric_scores=metrics,
+        cases=cases,
+        duration_seconds=max(duration, 0.0),
+    )
+
+
+def _snapshot_case(
+    case_id: str,
+    split: SplitName,
+    runs: list[EvalCaseResult],
+    primary: str,
+    fake_judge: bool = False,
+    primary_threshold: float = 1.0,
+) -> CaseSnapshot:
+    metric_names = sorted({metric.metric_name for run in runs for metric in run.overall_eval_metric_results})
+    scores = {
+        name:
+        _mean_optional(metric.score for run in runs for metric in run.overall_eval_metric_results
+                       if metric.metric_name == name)
+        for name in metric_names
+    }
+    if fake_judge:
+        scores[primary] = _fake_judge_score(runs)
+    statuses = {name: _aggregate_metric_status(runs, name) for name in metric_names}
+    reasons = _metric_reasons(runs)
+    errors = [run.error_message for run in runs if run.error_message]
+    passed = bool(runs) and all(run.final_eval_status == EvalStatus.PASSED for run in runs)
+    metric_not_evaluated = any(metric.eval_status == EvalStatus.NOT_EVALUATED for run in runs
+                               for metric in run.overall_eval_metric_results)
+    hard_failure = not runs or scores.get(primary) is None or metric_not_evaluated or any(
+        run.final_eval_status == EvalStatus.NOT_EVALUATED or run.error_message for run in runs)
+    if fake_judge and scores[primary] is not None:
+        statuses[primary] = (EvalStatus.PASSED.name if scores[primary] >= primary_threshold else EvalStatus.FAILED.name)
+        passed = scores[primary] >= primary_threshold
+    passed = passed and not hard_failure
+    actual, expected = _invocation_snapshots(runs)
+    return CaseSnapshot(
+        case_id=case_id,
+        split=split,
+        passed=passed,
+        hard_failure=hard_failure,
+        metric_scores=scores,
+        metric_statuses=statuses,
+        reasons=reasons,
+        error_message="; ".join(errors),
+        actual=actual,
+        expected=expected,
+    )
+
+
+def _aggregate_metric_status(runs: list[EvalCaseResult], metric_name: str) -> str:
+    statuses = [
+        metric.eval_status for run in runs for metric in run.overall_eval_metric_results
+        if metric.metric_name == metric_name
+    ]
+    if not statuses or EvalStatus.NOT_EVALUATED in statuses:
+        return EvalStatus.NOT_EVALUATED.name
+    if all(status == EvalStatus.PASSED for status in statuses):
+        return EvalStatus.PASSED.name
+    return EvalStatus.FAILED.name
+
+
+def _metric_reasons(runs: list[EvalCaseResult]) -> dict[str, str]:
+    reasons: dict[str, list[str]] = {}
+    for run in runs:
+        for metric in run.overall_eval_metric_results:
+            reason = metric.details.reason if metric.details else None
+            if reason:
+                reasons.setdefault(metric.metric_name, []).append(reason)
+    return {name: "; ".join(dict.fromkeys(values)) for name, values in reasons.items()}
+
+
+def _invocation_snapshots(runs: list[EvalCaseResult]) -> tuple[list[InvocationSnapshot], list[InvocationSnapshot]]:
+    actual: list[InvocationSnapshot] = []
+    expected: list[InvocationSnapshot] = []
+    for run in runs:
+        for result in run.eval_metric_result_per_invocation:
+            actual.append(_snapshot_invocation(result.actual_invocation))
+            if result.expected_invocation:
+                expected.append(_snapshot_invocation(result.expected_invocation))
+    return _deduplicate_invocations(actual), _deduplicate_invocations(expected)
+
+
+def _snapshot_invocation(invocation) -> InvocationSnapshot:
+    text = ""
+    if invocation.final_response and invocation.final_response.parts:
+        text = "".join(part.text or "" for part in invocation.final_response.parts)
+    tool_calls = [{
+        "name": call.name,
+        "args": call.args or {}
+    } for call in get_all_tool_calls(invocation.intermediate_data)]
+    return InvocationSnapshot(final_text=text, tool_calls=tool_calls)
+
+
+def _deduplicate_invocations(values: list[InvocationSnapshot]) -> list[InvocationSnapshot]:
+    unique: dict[str, InvocationSnapshot] = {}
+    for value in values:
+        key = value.model_dump_json()
+        unique.setdefault(key, value)
+    return list(unique.values())
+
+
+def _mean_optional(values) -> float | None:
+    present = [float(value) for value in values if value is not None]
+    return mean(present) if present else None
+
+
+def _validate_split_leakage(train_set: EvalSet, validation_set: EvalSet) -> None:
+    train_ids = [case.eval_id for case in train_set.eval_cases]
+    validation_ids = [case.eval_id for case in validation_set.eval_cases]
+    if len(train_ids) != len(set(train_ids)) or len(validation_ids) != len(set(validation_ids)):
+        raise ValueError("case ids must be unique within each split")
+    overlap = sorted(set(train_ids) & set(validation_ids))
+    if overlap:
+        raise ValueError(f"case ids overlap across splits: {overlap}")
+    train_hashes = {_case_hash(case) for case in train_set.eval_cases}
+    validation_hashes = {_case_hash(case) for case in validation_set.eval_cases}
+    if train_hashes & validation_hashes:
+        raise ValueError("normalized case content overlaps across splits")
+
+
+def _validate_gate_references(
+    gate: GateConfig,
+    optimizer: OptimizeConfigFile,
+    validation_set: EvalSet,
+) -> None:
+    metrics = {metric.metric_name for metric in optimizer.evaluate.get_eval_metrics()}
+    if gate.primary_metric not in metrics:
+        raise ValueError(f"primary_metric {gate.primary_metric!r} is not configured")
+    unknown_metrics = sorted(set(gate.hard_metric_names) - metrics)
+    if unknown_metrics:
+        raise ValueError(f"unknown hard metrics: {unknown_metrics}")
+    validation_ids = {case.eval_id for case in validation_set.eval_cases}
+    referenced = set(gate.critical_case_ids) | set(gate.hard_case_ids)
+    unknown_cases = sorted(referenced - validation_ids)
+    if unknown_cases:
+        raise ValueError(f"gate case ids must belong to validation: {unknown_cases}")
+
+
+def _case_hash(case) -> str:
+    payload = case.model_dump(mode="json", by_alias=True)
+    payload.pop("eval_id", None)
+    payload.pop("evalId", None)
+    payload.pop("creation_timestamp", None)
+    payload.pop("creationTimestamp", None)
+    for field in ("conversation", "actualConversation"):
+        for invocation in payload.get(field) or []:
+            invocation.pop("invocation_id", None)
+            invocation.pop("invocationId", None)
+            invocation.pop("creation_timestamp", None)
+            invocation.pop("creationTimestamp", None)
+    encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+    return hashlib.new(HASH_NAME, encoded.encode("utf-8")).hexdigest()
+
+
+def _file_hash(path: Path) -> str:
+    return hashlib.new(HASH_NAME, path.read_bytes()).hexdigest()
+
+
+def _fake_judge_score(runs: list[EvalCaseResult]) -> float | None:
+    """Deterministic offline judge used only when --fake-judge is enabled."""
+    scores = []
+    for run in runs:
+        actual = _invocation_snapshots([run])[0]
+        expected = _invocation_snapshots([run])[1]
+        scores.append(float(bool(actual and expected and actual[0].final_text == expected[0].final_text)))
+    return _mean_optional(scores)
+
+
+def _portable_dataset_path(path: Path) -> str:
+    """Avoid the SDK's colon-based case selector on Windows paths."""
+    resolved = path.resolve()
+    try:
+        resolved.relative_to(Path.cwd().resolve())
+    except ValueError as exc:
+        raise ValueError("dataset is outside the current workspace") from exc
+    return os.path.relpath(resolved, Path.cwd())
+
+
+def _dataset_for_sdk(path: Path, temp_dir: Path) -> str:
+    try:
+        return _portable_dataset_path(path)
+    except (OSError, ValueError):
+        pass
+    local = temp_dir / path.name
+    shutil.copyfile(path, local)
+    try:
+        return os.path.relpath(local, Path.cwd())
+    except ValueError:
+        return str(local)
diff --git a/examples/optimization/eval_optimize_loop/loop/models.py b/examples/optimization/eval_optimize_loop/loop/models.py
new file mode 100644
index 000000000..634d76c94
--- /dev/null
+++ b/examples/optimization/eval_optimize_loop/loop/models.py
@@ -0,0 +1,253 @@
+"""Strict models for the evaluation and optimization loop example."""
+
+from __future__ import annotations
+
+from enum import Enum
+from pathlib import Path
+from typing import Any
+
+from pydantic import BaseModel
+from pydantic import ConfigDict
+from pydantic import Field
+from pydantic import model_validator
+
+SCORE_EPSILON = 1e-6
+
+
+class StrictModel(BaseModel):
+    """Base model that rejects misspelled or unsupported fields."""
+
+    model_config = ConfigDict(extra="forbid")
+
+
+class SplitName(str, Enum):
+    """Supported dataset splits."""
+
+    TRAIN = "train"
+    VALIDATION = "validation"
+
+
+class ChangeKind(str, Enum):
+    """Per-case candidate change relative to baseline."""
+
+    NEW_PASS = "new_pass"
+    NEW_FAIL = "new_fail"
+    IMPROVED = "improved"
+    REGRESSED = "regressed"
+    UNCHANGED = "unchanged"
+
+
+class FailureCategory(str, Enum):
+    """Stable failure-attribution categories required by the issue."""
+
+    EXECUTION = "execution_error"
+    TOOL_CALL = "tool_call_error"
+    TOOL_ARGUMENT = "tool_argument_error"
+    RUBRIC = "llm_rubric_failure"
+    KNOWLEDGE = "knowledge_recall_failure"
+    FORMAT = "format_failure"
+    RESPONSE = "final_response_mismatch"
+    OTHER = "other"
+
+
+class GateConfig(StrictModel):
+    """Acceptance policy loaded from gate.json."""
+
+    primary_metric: str
+    min_score_delta: float = 0.0
+    critical_case_ids: list[str] = Field(default_factory=list)
+    max_critical_regression: float = Field(default=0.0, ge=0.0)
+    hard_case_ids: list[str] = Field(default_factory=list)
+    hard_metric_names: list[str] = Field(default_factory=list)
+    max_total_cost: float | None = Field(default=None, ge=0.0)
+    max_duration_seconds: float | None = Field(default=None, gt=0.0)
+    train_epsilon: float = Field(default=SCORE_EPSILON, ge=0.0)
+    validation_epsilon: float = Field(default=SCORE_EPSILON, ge=0.0)
+
+    @model_validator(mode="after")
+    def _unique_ids(self) -> "GateConfig":
+        for field_name in ("critical_case_ids", "hard_case_ids", "hard_metric_names"):
+            values = getattr(self, field_name)
+            if len(values) != len(set(values)):
+                raise ValueError(f"{field_name} contains duplicates")
+        return self
+
+
+class InvocationSnapshot(StrictModel):
+    """Minimal invocation material needed for attribution and audit."""
+
+    final_text: str = ""
+    tool_calls: list[dict[str, Any]] = Field(default_factory=list)
+
+
+class CaseSnapshot(StrictModel):
+    """Aggregated result for one case across all configured runs."""
+
+    case_id: str
+    split: SplitName
+    passed: bool
+    hard_failure: bool
+    metric_scores: dict[str, float | None] = Field(default_factory=dict)
+    metric_statuses: dict[str, str] = Field(default_factory=dict)
+    reasons: dict[str, str] = Field(default_factory=dict)
+    error_message: str = ""
+    actual: list[InvocationSnapshot] = Field(default_factory=list)
+    expected: list[InvocationSnapshot] = Field(default_factory=list)
+
+
+class EvaluationSnapshot(StrictModel):
+    """Evaluation summary for a single prompt and dataset split."""
+
+    split: SplitName
+    primary_metric: str
+    primary_score: float | None
+    pass_rate: float
+    metric_scores: dict[str, float | None] = Field(default_factory=dict)
+    cases: list[CaseSnapshot] = Field(default_factory=list)
+    duration_seconds: float = Field(ge=0.0)
+
+
+class Attribution(StrictModel):
+    """Explainable primary attribution for one failed case."""
+
+    case_id: str
+    category: FailureCategory
+    rule_id: str
+    evidence: str
+
+
+class CaseDelta(StrictModel):
+    """Candidate change for a single case."""
+
+    case_id: str
+    baseline_passed: bool
+    candidate_passed: bool
+    baseline_score: float | None
+    candidate_score: float | None
+    score_delta: float | None
+    change: ChangeKind
+    hard_failure_added: bool = False
+
+
+class SplitDelta(StrictModel):
+    """Candidate changes for a dataset split."""
+
+    split: SplitName
+    baseline_score: float | None
+    candidate_score: float | None
+    score_delta: float | None
+    cases: list[CaseDelta] = Field(default_factory=list)
+
+
+class CostSummary(StrictModel):
+    """Known pipeline costs and their observability."""
+
+    optimizer_cost: float = Field(default=0.0, ge=0.0)
+    external_cost: float = Field(default=0.0, ge=0.0)
+    total_cost: float = Field(default=0.0, ge=0.0)
+    cost_complete: bool = False
+
+
+class GateCheck(StrictModel):
+    """One auditable gate predicate."""
+
+    name: str
+    passed: bool
+    reason: str
+
+
+class GateDecision(StrictModel):
+    """Final acceptance decision."""
+
+    accepted: bool
+    overfitting: bool
+    checks: list[GateCheck] = Field(default_factory=list)
+    reasons: list[str] = Field(default_factory=list)
+
+
+class AuditInfo(StrictModel):
+    """Reproduction and provenance fields."""
+
+    seed: int
+    input_hashes: dict[str, str]
+    model_name: str
+    num_runs: int
+    case_parallelism: int
+    python_version: str
+    sdk_version: str
+    git_sha: str
+    started_at: str
+    finished_at: str
+    stage_durations: dict[str, float]
+
+
+class OptimizationSummary(StrictModel):
+    """Subset of OptimizeResult consumed by this example."""
+
+    status: str
+    finish_reason: str
+    best_prompts: dict[str, str] = Field(default_factory=dict)
+    rounds: list[dict[str, Any]] = Field(default_factory=list)
+    total_cost: float = 0.0
+    error_message: str = ""
+
+
+class OptimizationReport(StrictModel):
+    """Top-level JSON and Markdown report schema."""
+
+    schema_version: str = "v1"
+    status: str
+    baseline: dict[SplitName, EvaluationSnapshot]
+    candidate: dict[SplitName, EvaluationSnapshot] = Field(default_factory=dict)
+    delta: dict[SplitName, SplitDelta] = Field(default_factory=dict)
+    attributions: list[Attribution] = Field(default_factory=list)
+    attribution_counts: dict[FailureCategory, int] = Field(default_factory=dict)
+    optimization: OptimizationSummary
+    gate: GateDecision
+    cost: CostSummary
+    audit: AuditInfo
+    source_updated: bool = False
+    failures: list[str] = Field(default_factory=list)
+
+
+class InputBundle(StrictModel):
+    """Validated input paths and content hashes."""
+
+    prompt_path: Path
+    train_path: Path
+    validation_path: Path
+    optimizer_path: Path
+    gate_path: Path
+    hashes: dict[str, str]
+
+
+class InputPaths(StrictModel):
+    """Unvalidated CLI input paths."""
+
+    prompt_path: Path
+    train_path: Path
+    validation_path: Path
+    optimizer_path: Path
+    gate_path: Path
+
+
+class PipelineOptions(StrictModel):
+    """CLI-independent options for one pipeline run."""
+
+    paths: InputPaths
+    output_dir: Path
+    mode: str = "fake-model"
+    fake_judge: bool = False
+    trace_file: Path | None = None
+    write_back: bool = False
+    model_name: str = "fake-model"
+    case_parallelism: int = 1
+    num_runs: int = 1
+
+
+class PipelineResult(StrictModel):
+    """Paths and report returned by the orchestration entry point."""
+
+    report: OptimizationReport
+    json_path: Path
+    markdown_path: Path
diff --git a/examples/optimization/eval_optimize_loop/loop/pipeline.py b/examples/optimization/eval_optimize_loop/loop/pipeline.py
new file mode 100644
index 000000000..124121b96
--- /dev/null
+++ b/examples/optimization/eval_optimize_loop/loop/pipeline.py
@@ -0,0 +1,467 @@
+"""Orchestration for baseline evaluation, optimization, replay, and gating."""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import json
+import platform
+import shutil
+import subprocess
+import time
+from dataclasses import dataclass
+from datetime import datetime
+from datetime import timedelta
+from pathlib import Path
+from typing import Any
+
+from trpc_agent_sdk.evaluation import AgentOptimizer
+from trpc_agent_sdk.evaluation import OptimizeConfigFile
+from trpc_agent_sdk.evaluation import TargetPrompt
+from trpc_agent_sdk.evaluation._target_prompt import _RollbackError
+
+from ..agent.agent import fake_call_agent
+from ..agent.agent import real_call_agent
+from .analysis import GateInput
+from .analysis import attribute_cases
+from .analysis import compare_snapshots
+from .analysis import evaluate_gate
+from .evaluation import EvaluationRequest
+from .evaluation import evaluate_split
+from .evaluation import validate_inputs
+from .models import AuditInfo
+from .models import CostSummary
+from .models import GateDecision
+from .models import EvaluationSnapshot
+from .models import InputBundle
+from .models import OptimizationReport
+from .models import OptimizationSummary
+from .models import PipelineOptions
+from .models import PipelineResult
+from .models import SplitName
+from .reporting import write_reports
+
+PROMPT_KEY = "system_prompt"
+PROMPT_RELATIVE_PATH = Path("prompts") / "system.md"
+REAL_MODE = "real"
+FAKE_MODEL_MODE = "fake-model"
+TRACE_MODE = "trace"
+
+
+@dataclass
+class _FailureContext:
+    bundle: InputBundle | None = None
+    optimizer_config: OptimizeConfigFile | None = None
+
+
+async def run_pipeline(options: PipelineOptions) -> PipelineResult:
+    """Run the loop and convert unexpected failures into an audit report."""
+    started = time.monotonic()
+    context = _FailureContext()
+    try:
+        return await _run_pipeline(options, started, context)
+    except Exception as exc:
+        return _failure_result(options, started, exc, context)
+
+
+async def _run_pipeline(
+    options: PipelineOptions,
+    started: float,
+    context: _FailureContext,
+) -> PipelineResult:
+    """Run the complete evaluation and optimization loop."""
+    bundle, optimizer_config, gate_config = validate_inputs(options.paths)
+    context.bundle = bundle
+    context.optimizer_config = optimizer_config
+    if options.write_back and options.mode != REAL_MODE:
+        raise ValueError("write-back requires real mode")
+    options = options.model_copy(
+        update={
+            "num_runs": optimizer_config.evaluate.num_runs,
+            "case_parallelism": optimizer_config.optimize.eval_case_parallelism,
+        })
+    execution = _execute_pipeline(options, started, bundle, gate_config)
+    if gate_config.max_duration_seconds is None:
+        return await execution
+    return await asyncio.wait_for(
+        execution,
+        timeout=gate_config.max_duration_seconds,
+    )
+
+
+async def _execute_pipeline(
+    options: PipelineOptions,
+    started: float,
+    bundle: InputBundle,
+    gate_config,
+) -> PipelineResult:
+    workspace = _prepare_workspace(bundle, options.output_dir)
+    baseline = await _evaluate_pair(workspace, bundle, options, "baseline")
+    optimization, candidate_prompts = await _optimize(
+        workspace,
+        bundle,
+        options,
+        gate_config.primary_metric,
+    )
+    candidate = await _evaluate_candidate(
+        workspace,
+        bundle,
+        options,
+        candidate_prompts,
+    )
+    deltas = {
+        split: compare_snapshots(baseline[split], candidate[split])
+        for split in (SplitName.TRAIN, SplitName.VALIDATION)
+    }
+    attributions, counts = _collect_attributions(baseline)
+    cost = CostSummary(
+        optimizer_cost=optimization.total_cost,
+        external_cost=0.0,
+        total_cost=optimization.total_cost,
+        cost_complete=options.mode != REAL_MODE,
+    )
+    duration = time.monotonic() - started
+    decision = evaluate_gate(
+        GateInput(
+            config=gate_config,
+            train_delta=deltas[SplitName.TRAIN],
+            validation_delta=deltas[SplitName.VALIDATION],
+            baseline_validation=baseline[SplitName.VALIDATION],
+            candidate_validation=candidate[SplitName.VALIDATION],
+            cost=cost,
+            duration_seconds=duration,
+        ))
+    report = _build_report(
+        bundle,
+        options,
+        baseline,
+        candidate,
+        deltas,
+        attributions,
+        counts,
+        optimization,
+        cost,
+        decision,
+        duration,
+        False,
+    )
+    json_path, markdown_path = await _write_back_and_report(
+        report,
+        bundle,
+        candidate_prompts,
+        options,
+    )
+    return PipelineResult(report=report, json_path=json_path, markdown_path=markdown_path)
+
+
+async def _write_back_and_report(
+    report: OptimizationReport,
+    bundle: InputBundle,
+    prompts: dict[str, str],
+    options: PipelineOptions,
+) -> tuple[Path, Path]:
+    if not report.gate.accepted or not options.write_back:
+        return write_reports(report, options.output_dir)
+    original = bundle.prompt_path.read_text(encoding="utf-8")
+    try:
+        await _maybe_write_back(bundle, prompts, report.gate.accepted, options.write_back)
+        report.source_updated = True
+        return write_reports(report, options.output_dir)
+    except Exception:
+        # TargetPrompt.write_all provides atomic rollback; this extra restore
+        # protects callers if failure occurs after the SDK write completes.
+        await _restore_prompt(bundle.prompt_path, original)
+        report.source_updated = False
+        raise
+
+
+async def _restore_prompt(path: Path, content: str) -> None:
+    target = TargetPrompt().add_path(PROMPT_KEY, str(path))
+    await target.write_all({PROMPT_KEY: content})
+
+
+def _failure_result(
+    options: PipelineOptions,
+    started: float,
+    error: Exception,
+    context: _FailureContext | None = None,
+) -> PipelineResult:
+    finished = datetime.now().astimezone()
+    started_wall = finished - timedelta(seconds=time.monotonic() - started)
+    reason = _failure_reason(error)
+    decision = GateDecision(accepted=False, overfitting=False, reasons=[reason])
+    input_hashes = context.bundle.hashes if context and context.bundle else {}
+    optimizer_config = context.optimizer_config if context else None
+    report = OptimizationReport(
+        status="REJECTED",
+        baseline={},
+        candidate={},
+        delta={},
+        optimization=OptimizationSummary(
+            status="FAILED",
+            finish_reason="pipeline_exception",
+            error_message=reason,
+        ),
+        gate=decision,
+        cost=CostSummary(cost_complete=False),
+        audit=AuditInfo(
+            seed=91,
+            input_hashes=input_hashes,
+            model_name=options.model_name,
+            num_runs=optimizer_config.evaluate.num_runs if optimizer_config else 1,
+            case_parallelism=(optimizer_config.optimize.eval_case_parallelism
+                              if optimizer_config else options.case_parallelism),
+            python_version=platform.python_version(),
+            sdk_version=_sdk_version(),
+            git_sha=_git_sha(),
+            started_at=started_wall.isoformat(),
+            finished_at=finished.isoformat(),
+            stage_durations={"pipeline": time.monotonic() - started},
+        ),
+        failures=[reason],
+    )
+    json_path, markdown_path = write_reports(report, options.output_dir)
+    return PipelineResult(report=report, json_path=json_path, markdown_path=markdown_path)
+
+
+def _failure_reason(error: Exception) -> str:
+    """Keep SDK rollback details in the audit failure reason."""
+    reason = f"{type(error).__name__}: pipeline stage failed"
+    if isinstance(error, _RollbackError):
+        return f"{reason}: {error}"
+    return reason
+
+
+async def _evaluate_pair(
+    prompt_path: Path,
+    bundle: InputBundle,
+    options: PipelineOptions,
+    phase: str,
+) -> dict[SplitName, EvaluationSnapshot]:
+    call_agent = None if options.mode == TRACE_MODE else _make_call_agent(prompt_path, options.mode)
+    train_path = bundle.train_path
+    validation_path = bundle.validation_path
+    if options.mode == TRACE_MODE:
+        if options.trace_file is None:
+            raise ValueError("trace_file is required in trace mode")
+        train_path = _trace_dataset_path(options.trace_file, phase, SplitName.TRAIN, options.output_dir)
+        validation_path = _trace_dataset_path(
+            options.trace_file,
+            phase,
+            SplitName.VALIDATION,
+            options.output_dir,
+        )
+    return {
+        SplitName.TRAIN:
+        await evaluate_split(
+            EvaluationRequest(
+                train_path,
+                bundle.optimizer_path,
+                SplitName.TRAIN,
+                call_agent,
+                options.fake_judge,
+            )),
+        SplitName.VALIDATION:
+        await evaluate_split(
+            EvaluationRequest(
+                validation_path,
+                bundle.optimizer_path,
+                SplitName.VALIDATION,
+                call_agent,
+                options.fake_judge,
+            )),
+    }
+
+
+def _trace_dataset_path(trace_file: Path, phase: str, split: SplitName, output_dir: Path) -> Path:
+    payload = json.loads(trace_file.read_text(encoding="utf-8"))
+    try:
+        cases = payload[phase][split.value]
+    except (KeyError, TypeError) as exc:
+        raise ValueError(f"trace file missing {phase}/{split.value}") from exc
+    path = output_dir / "work" / "trace" / phase / f"{split.value}.evalset.json"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    normalized = [{
+        **case, "actual_conversation": case.get("actual_conversation", case.get("conversation", []))
+    } for case in cases]
+    path.write_text(
+        json.dumps({
+            "eval_set_id": f"{phase}-{split.value}",
+            "eval_cases": normalized
+        }),
+        encoding="utf-8",
+    )
+    return path
+
+
+async def _optimize(
+    prompt_path: Path,
+    bundle: InputBundle,
+    options: PipelineOptions,
+    primary_metric: str,
+) -> tuple[OptimizationSummary, dict[str, str]]:
+    del primary_metric
+    baseline_prompt = prompt_path.read_text(encoding="utf-8")
+    if options.mode in (FAKE_MODEL_MODE, TRACE_MODE):
+        candidate = {PROMPT_KEY: baseline_prompt + "\n\nOPTIMIZED_CANDIDATE\n"}
+        return OptimizationSummary(
+            status="SUCCEEDED",
+            finish_reason="fake_candidate",
+            best_prompts=candidate,
+        ), candidate
+    if options.mode != REAL_MODE:
+        raise ValueError(f"unsupported mode: {options.mode}")
+    target = TargetPrompt().add_path(PROMPT_KEY, str(prompt_path))
+    call_agent = _make_call_agent(prompt_path, options.mode)
+    result = await AgentOptimizer.optimize(
+        config_path=str(bundle.optimizer_path),
+        call_agent=call_agent,
+        target_prompt=target,
+        train_dataset_path=str(bundle.train_path),
+        validation_dataset_path=str(bundle.validation_path),
+        output_dir=str(options.output_dir / "optimizer"),
+        update_source=False,
+        verbose=0,
+    )
+    summary = OptimizationSummary(
+        status=str(result.status),
+        finish_reason=str(result.finish_reason),
+        best_prompts=result.best_prompts,
+        rounds=[round_record.model_dump(mode="json") for round_record in result.rounds],
+        total_cost=float(result.total_llm_cost),
+        error_message=result.error_message,
+    )
+    candidate = result.best_prompts
+    if not candidate or PROMPT_KEY not in candidate:
+        candidate = {PROMPT_KEY: baseline_prompt}
+    return summary, candidate
+
+
+async def _evaluate_candidate(
+    prompt_path: Path,
+    bundle: InputBundle,
+    options: PipelineOptions,
+    prompts: dict[str, str],
+) -> dict[SplitName, EvaluationSnapshot]:
+    baseline = prompt_path.read_text(encoding="utf-8")
+    try:
+        prompt_path.write_text(prompts[PROMPT_KEY], encoding="utf-8")
+        return await _evaluate_pair(prompt_path, bundle, options, "candidate")
+    finally:
+        prompt_path.write_text(baseline, encoding="utf-8")
+
+
+def _prepare_workspace(bundle: InputBundle, output_dir: Path) -> Path:
+    prompt_path = output_dir / "work" / PROMPT_RELATIVE_PATH
+    prompt_path.parent.mkdir(parents=True, exist_ok=True)
+    shutil.copyfile(bundle.prompt_path, prompt_path)
+    return prompt_path
+
+
+def _make_call_agent(prompt_path: Path, mode: str):
+    if mode == FAKE_MODEL_MODE:
+
+        async def call(query: str) -> str:
+            return await fake_call_agent(prompt_path, query)
+
+        return call
+    if mode == REAL_MODE:
+
+        async def call(query: str) -> str:
+            return await real_call_agent(prompt_path, query)
+
+        return call
+    raise ValueError(f"unsupported call-agent mode: {mode}")
+
+
+def _collect_attributions(snapshots: dict[SplitName, EvaluationSnapshot], ) -> tuple[list, dict]:
+    all_attributions = []
+    counts = {}
+    for snapshot in snapshots.values():
+        attributions, snapshot_counts = attribute_cases(snapshot)
+        all_attributions.extend(attributions)
+        for category, count in snapshot_counts.items():
+            counts[category] = counts.get(category, 0) + count
+    return all_attributions, counts
+
+
+async def _maybe_write_back(
+    bundle: InputBundle,
+    prompts: dict[str, str],
+    accepted: bool,
+    write_back: bool,
+) -> bool:
+    if not accepted or not write_back:
+        return False
+    target = TargetPrompt().add_path(PROMPT_KEY, str(bundle.prompt_path))
+    await target.write_all(prompts)
+    return True
+
+
+def _build_report(
+    bundle: InputBundle,
+    options: PipelineOptions,
+    baseline: dict[SplitName, EvaluationSnapshot],
+    candidate: dict[SplitName, EvaluationSnapshot],
+    deltas: dict[SplitName, Any],
+    attributions: list,
+    counts: dict,
+    optimization: OptimizationSummary,
+    cost: CostSummary,
+    decision,
+    duration: float,
+    source_updated: bool,
+) -> OptimizationReport:
+    finished = datetime.now().astimezone()
+    started = finished - timedelta(seconds=duration)
+    audit = AuditInfo(
+        seed=91,
+        input_hashes={
+            **bundle.hashes, "candidate_prompt": _prompt_hash(optimization.best_prompts)
+        },
+        model_name=options.model_name,
+        num_runs=options.num_runs,
+        case_parallelism=options.case_parallelism,
+        python_version=platform.python_version(),
+        sdk_version=_sdk_version(),
+        git_sha=_git_sha(),
+        started_at=started.isoformat(),
+        finished_at=finished.isoformat(),
+        stage_durations={"pipeline": duration},
+    )
+    status = "ACCEPTED" if decision.accepted else "REJECTED"
+    return OptimizationReport(
+        status=status,
+        baseline=baseline,
+        candidate=candidate,
+        delta=deltas,
+        attributions=attributions,
+        attribution_counts=counts,
+        optimization=optimization,
+        gate=decision,
+        cost=cost,
+        audit=audit,
+        source_updated=source_updated,
+    )
+
+
+def _prompt_hash(prompts: dict[str, str]) -> str:
+    value = prompts.get(PROMPT_KEY, "")
+    return hashlib.sha256(value.encode("utf-8")).hexdigest()
+
+
+def _git_sha() -> str:
+    try:
+        return subprocess.run(
+            ["git", "rev-parse", "HEAD"],
+            capture_output=True,
+            text=True,
+            check=True,
+        ).stdout.strip()
+    except (OSError, subprocess.CalledProcessError):
+        return "unknown"
+
+
+def _sdk_version() -> str:
+    from trpc_agent_sdk.version import __version__
+    return str(__version__)
diff --git a/examples/optimization/eval_optimize_loop/loop/reporting.py b/examples/optimization/eval_optimize_loop/loop/reporting.py
new file mode 100644
index 000000000..dabe4d13b
--- /dev/null
+++ b/examples/optimization/eval_optimize_loop/loop/reporting.py
@@ -0,0 +1,133 @@
+"""Stable JSON and Markdown report writers."""
+
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+from pathlib import Path
+
+from .models import OptimizationReport
+
+REPORT_JSON_NAME = "optimization_report.json"
+REPORT_MARKDOWN_NAME = "optimization_report.md"
+
+
+def write_reports(report: OptimizationReport, output_dir: Path) -> tuple[Path, Path]:
+    """Write JSON and Markdown reports with restrictive local permissions."""
+    output_dir.mkdir(parents=True, exist_ok=True)
+    json_path = output_dir / REPORT_JSON_NAME
+    markdown_path = output_dir / REPORT_MARKDOWN_NAME
+    payload = report.model_dump(mode="json")
+    contents = {
+        json_path: json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
+        markdown_path: render_markdown(report),
+    }
+    _publish_reports(contents)
+    return json_path, markdown_path
+
+
+def _publish_reports(contents: dict[Path, str]) -> None:
+    staged: dict[Path, Path] = {}
+    try:
+        for path, content in contents.items():
+            staged[path] = _stage_report(path, content)
+    except Exception:
+        for staged_path in staged.values():
+            staged_path.unlink(missing_ok=True)
+        raise
+    previous = {path: path.read_bytes() if path.exists() else None for path in contents}
+    replaced: list[Path] = []
+    try:
+        for path, staged_path in staged.items():
+            os.replace(staged_path, path)
+            replaced.append(path)
+    except Exception:
+        _restore_reports(replaced, previous)
+        raise
+    finally:
+        for staged_path in staged.values():
+            staged_path.unlink(missing_ok=True)
+
+
+def _stage_report(path: Path, content: str) -> Path:
+    return _stage_bytes(path, content.encode("utf-8"))
+
+
+def _stage_bytes(path: Path, content: bytes) -> Path:
+    descriptor, temporary_name = tempfile.mkstemp(
+        dir=path.parent,
+        prefix=f".{path.name}.",
+        suffix=".tmp",
+    )
+    os.close(descriptor)
+    staged_path = Path(temporary_name)
+    staged_path.write_bytes(content)
+    _restrict_permissions(staged_path)
+    return staged_path
+
+
+def _restore_reports(replaced: list[Path], previous: dict[Path, bytes | None]) -> None:
+    for path in reversed(replaced):
+        content = previous[path]
+        if content is None:
+            path.unlink(missing_ok=True)
+            continue
+        staged_path = _stage_bytes(path, content)
+        try:
+            os.replace(staged_path, path)
+        finally:
+            staged_path.unlink(missing_ok=True)
+
+
+def render_markdown(report: OptimizationReport) -> str:
+    """Render the human-readable acceptance summary."""
+    gate = "ACCEPT" if report.gate.accepted else "REJECT"
+    lines = [
+        "# Optimization Report",
+        "",
+        f"- Status: `{report.status}`",
+        f"- Gate: **{gate}**",
+        f"- Overfitting: `{report.gate.overfitting}`",
+        "",
+        "## Scores",
+        "",
+        "| Split | Baseline | Candidate | Delta | Pass rate |",
+        "| --- | ---: | ---: | ---: | ---: |",
+    ]
+    for split, delta in report.delta.items():
+        lines.append(f"| {split.value} | {_format_score(delta.baseline_score)} | "
+                     f"{_format_score(delta.candidate_score)} | {_format_score(delta.score_delta)} | "
+                     f"{_pass_rate(report, split)} |")
+    lines.extend(["", "## Gate checks", ""])
+    lines.extend(f"- {'PASS' if check.passed else 'FAIL'} `{check.name}`: {check.reason}"
+                 for check in report.gate.checks)
+    lines.extend(["", "## Failure attribution", ""])
+    if report.attribution_counts:
+        lines.extend(f"- `{category.value}`: {count}" for category, count in report.attribution_counts.items())
+    else:
+        lines.append("- None")
+    lines.extend(["", "## Reasons", ""])
+    if report.gate.reasons:
+        lines.extend(f"- {reason}" for reason in report.gate.reasons)
+    else:
+        lines.append("- Accepted")
+    lines.extend(["", "## Audit", ""])
+    lines.append(f"- Seed: `{report.audit.seed}`")
+    lines.append(f"- Cost complete: `{report.cost.cost_complete}`")
+    lines.append(f"- Total duration: `{sum(report.audit.stage_durations.values()):.3f}s`")
+    return "\n".join(lines) + "\n"
+
+
+def _pass_rate(report: OptimizationReport, split) -> str:
+    snapshot = report.candidate.get(split)
+    return f"{snapshot.pass_rate:.3f}" if snapshot else "-"
+
+
+def _format_score(value: float | None) -> str:
+    return "-" if value is None else f"{value:.6f}"
+
+
+def _restrict_permissions(path: Path) -> None:
+    if os.name != "nt":
+        path.chmod(0o600)
diff --git a/examples/optimization/eval_optimize_loop/optimization_report.json b/examples/optimization/eval_optimize_loop/optimization_report.json
new file mode 100644
index 000000000..bab16b012
--- /dev/null
+++ b/examples/optimization/eval_optimize_loop/optimization_report.json
@@ -0,0 +1,545 @@
+{
+  "schema_version": "v1",
+  "status": "ACCEPTED",
+  "baseline": {
+    "train": {
+      "split": "train",
+      "primary_metric": "final_response_avg_score",
+      "primary_score": 1.0,
+      "pass_rate": 1.0,
+      "metric_scores": {
+        "final_response_avg_score": 1.0
+      },
+      "cases": [
+        {
+          "case_id": "train_account_recall",
+          "split": "train",
+          "passed": true,
+          "hard_failure": false,
+          "metric_scores": {
+            "final_response_avg_score": 1.0
+          },
+          "metric_statuses": {
+            "final_response_avg_score": "PASSED"
+          },
+          "reasons": {},
+          "error_message": "",
+          "actual": [
+            {
+              "final_text": "{\"queue\":\"account\"}",
+              "tool_calls": []
+            }
+          ],
+          "expected": [
+            {
+              "final_text": "{\"queue\":\"account\"}",
+              "tool_calls": []
+            }
+          ]
+        },
+        {
+          "case_id": "train_billing_format",
+          "split": "train",
+          "passed": true,
+          "hard_failure": false,
+          "metric_scores": {
+            "final_response_avg_score": 1.0
+          },
+          "metric_statuses": {
+            "final_response_avg_score": "PASSED"
+          },
+          "reasons": {},
+          "error_message": "",
+          "actual": [
+            {
+              "final_text": "{\"queue\":\"billing\"}",
+              "tool_calls": []
+            }
+          ],
+          "expected": [
+            {
+              "final_text": "{\"queue\":\"billing\"}",
+              "tool_calls": []
+            }
+          ]
+        },
+        {
+          "case_id": "train_technical_router",
+          "split": "train",
+          "passed": true,
+          "hard_failure": false,
+          "metric_scores": {
+            "final_response_avg_score": 1.0
+          },
+          "metric_statuses": {
+            "final_response_avg_score": "PASSED"
+          },
+          "reasons": {},
+          "error_message": "",
+          "actual": [
+            {
+              "final_text": "{\"queue\":\"technical\"}",
+              "tool_calls": []
+            }
+          ],
+          "expected": [
+            {
+              "final_text": "{\"queue\":\"technical\"}",
+              "tool_calls": []
+            }
+          ]
+        }
+      ],
+      "duration_seconds": 0.010187200001382735
+    },
+    "validation": {
+      "split": "validation",
+      "primary_metric": "final_response_avg_score",
+      "primary_score": 0.3333333333333333,
+      "pass_rate": 0.3333333333333333,
+      "metric_scores": {
+        "final_response_avg_score": 0.3333333333333333
+      },
+      "cases": [
+        {
+          "case_id": "val_candidate_improves",
+          "split": "validation",
+          "passed": false,
+          "hard_failure": false,
+          "metric_scores": {
+            "final_response_avg_score": 0.0
+          },
+          "metric_statuses": {
+            "final_response_avg_score": "FAILED"
+          },
+          "reasons": {},
+          "error_message": "",
+          "actual": [
+            {
+              "final_text": "{\"queue\":\"unknown\"}",
+              "tool_calls": []
+            }
+          ],
+          "expected": [
+            {
+              "final_text": "{\"queue\":\"billing\"}",
+              "tool_calls": []
+            }
+          ]
+        },
+        {
+          "case_id": "val_candidate_no_effect",
+          "split": "validation",
+          "passed": false,
+          "hard_failure": false,
+          "metric_scores": {
+            "final_response_avg_score": 0.0
+          },
+          "metric_statuses": {
+            "final_response_avg_score": "FAILED"
+          },
+          "reasons": {},
+          "error_message": "",
+          "actual": [
+            {
+              "final_text": "{\"queue\":\"unknown\"}",
+              "tool_calls": []
+            }
+          ],
+          "expected": [
+            {
+              "final_text": "{\"queue\":\"billing\"}",
+              "tool_calls": []
+            }
+          ]
+        },
+        {
+          "case_id": "val_candidate_regresses",
+          "split": "validation",
+          "passed": true,
+          "hard_failure": false,
+          "metric_scores": {
+            "final_response_avg_score": 1.0
+          },
+          "metric_statuses": {
+            "final_response_avg_score": "PASSED"
+          },
+          "reasons": {},
+          "error_message": "",
+          "actual": [
+            {
+              "final_text": "{\"queue\":\"technical\"}",
+              "tool_calls": []
+            }
+          ],
+          "expected": [
+            {
+              "final_text": "{\"queue\":\"technical\"}",
+              "tool_calls": []
+            }
+          ]
+        }
+      ],
+      "duration_seconds": 0.005737899999076035
+    }
+  },
+  "candidate": {
+    "train": {
+      "split": "train",
+      "primary_metric": "final_response_avg_score",
+      "primary_score": 1.0,
+      "pass_rate": 1.0,
+      "metric_scores": {
+        "final_response_avg_score": 1.0
+      },
+      "cases": [
+        {
+          "case_id": "train_account_recall",
+          "split": "train",
+          "passed": true,
+          "hard_failure": false,
+          "metric_scores": {
+            "final_response_avg_score": 1.0
+          },
+          "metric_statuses": {
+            "final_response_avg_score": "PASSED"
+          },
+          "reasons": {},
+          "error_message": "",
+          "actual": [
+            {
+              "final_text": "{\"queue\":\"account\"}",
+              "tool_calls": []
+            }
+          ],
+          "expected": [
+            {
+              "final_text": "{\"queue\":\"account\"}",
+              "tool_calls": []
+            }
+          ]
+        },
+        {
+          "case_id": "train_billing_format",
+          "split": "train",
+          "passed": true,
+          "hard_failure": false,
+          "metric_scores": {
+            "final_response_avg_score": 1.0
+          },
+          "metric_statuses": {
+            "final_response_avg_score": "PASSED"
+          },
+          "reasons": {},
+          "error_message": "",
+          "actual": [
+            {
+              "final_text": "{\"queue\":\"billing\"}",
+              "tool_calls": []
+            }
+          ],
+          "expected": [
+            {
+              "final_text": "{\"queue\":\"billing\"}",
+              "tool_calls": []
+            }
+          ]
+        },
+        {
+          "case_id": "train_technical_router",
+          "split": "train",
+          "passed": true,
+          "hard_failure": false,
+          "metric_scores": {
+            "final_response_avg_score": 1.0
+          },
+          "metric_statuses": {
+            "final_response_avg_score": "PASSED"
+          },
+          "reasons": {},
+          "error_message": "",
+          "actual": [
+            {
+              "final_text": "{\"queue\":\"technical\"}",
+              "tool_calls": []
+            }
+          ],
+          "expected": [
+            {
+              "final_text": "{\"queue\":\"technical\"}",
+              "tool_calls": []
+            }
+          ]
+        }
+      ],
+      "duration_seconds": 0.00834469999972498
+    },
+    "validation": {
+      "split": "validation",
+      "primary_metric": "final_response_avg_score",
+      "primary_score": 0.6666666666666666,
+      "pass_rate": 0.6666666666666666,
+      "metric_scores": {
+        "final_response_avg_score": 0.6666666666666666
+      },
+      "cases": [
+        {
+          "case_id": "val_candidate_improves",
+          "split": "validation",
+          "passed": true,
+          "hard_failure": false,
+          "metric_scores": {
+            "final_response_avg_score": 1.0
+          },
+          "metric_statuses": {
+            "final_response_avg_score": "PASSED"
+          },
+          "reasons": {},
+          "error_message": "",
+          "actual": [
+            {
+              "final_text": "{\"queue\":\"billing\"}",
+              "tool_calls": []
+            }
+          ],
+          "expected": [
+            {
+              "final_text": "{\"queue\":\"billing\"}",
+              "tool_calls": []
+            }
+          ]
+        },
+        {
+          "case_id": "val_candidate_no_effect",
+          "split": "validation",
+          "passed": true,
+          "hard_failure": false,
+          "metric_scores": {
+            "final_response_avg_score": 1.0
+          },
+          "metric_statuses": {
+            "final_response_avg_score": "PASSED"
+          },
+          "reasons": {},
+          "error_message": "",
+          "actual": [
+            {
+              "final_text": "{\"queue\":\"billing\"}",
+              "tool_calls": []
+            }
+          ],
+          "expected": [
+            {
+              "final_text": "{\"queue\":\"billing\"}",
+              "tool_calls": []
+            }
+          ]
+        },
+        {
+          "case_id": "val_candidate_regresses",
+          "split": "validation",
+          "passed": false,
+          "hard_failure": false,
+          "metric_scores": {
+            "final_response_avg_score": 0.0
+          },
+          "metric_statuses": {
+            "final_response_avg_score": "FAILED"
+          },
+          "reasons": {},
+          "error_message": "",
+          "actual": [
+            {
+              "final_text": "{\"queue\":\"billing\"}",
+              "tool_calls": []
+            }
+          ],
+          "expected": [
+            {
+              "final_text": "{\"queue\":\"technical\"}",
+              "tool_calls": []
+            }
+          ]
+        }
+      ],
+      "duration_seconds": 0.00605869999708375
+    }
+  },
+  "delta": {
+    "train": {
+      "split": "train",
+      "baseline_score": 1.0,
+      "candidate_score": 1.0,
+      "score_delta": 0.0,
+      "cases": [
+        {
+          "case_id": "train_account_recall",
+          "baseline_passed": true,
+          "candidate_passed": true,
+          "baseline_score": 1.0,
+          "candidate_score": 1.0,
+          "score_delta": 0.0,
+          "change": "unchanged",
+          "hard_failure_added": false
+        },
+        {
+          "case_id": "train_billing_format",
+          "baseline_passed": true,
+          "candidate_passed": true,
+          "baseline_score": 1.0,
+          "candidate_score": 1.0,
+          "score_delta": 0.0,
+          "change": "unchanged",
+          "hard_failure_added": false
+        },
+        {
+          "case_id": "train_technical_router",
+          "baseline_passed": true,
+          "candidate_passed": true,
+          "baseline_score": 1.0,
+          "candidate_score": 1.0,
+          "score_delta": 0.0,
+          "change": "unchanged",
+          "hard_failure_added": false
+        }
+      ]
+    },
+    "validation": {
+      "split": "validation",
+      "baseline_score": 0.3333333333333333,
+      "candidate_score": 0.6666666666666666,
+      "score_delta": 0.3333333333333333,
+      "cases": [
+        {
+          "case_id": "val_candidate_improves",
+          "baseline_passed": false,
+          "candidate_passed": true,
+          "baseline_score": 0.0,
+          "candidate_score": 1.0,
+          "score_delta": 1.0,
+          "change": "new_pass",
+          "hard_failure_added": false
+        },
+        {
+          "case_id": "val_candidate_no_effect",
+          "baseline_passed": false,
+          "candidate_passed": true,
+          "baseline_score": 0.0,
+          "candidate_score": 1.0,
+          "score_delta": 1.0,
+          "change": "new_pass",
+          "hard_failure_added": false
+        },
+        {
+          "case_id": "val_candidate_regresses",
+          "baseline_passed": true,
+          "candidate_passed": false,
+          "baseline_score": 1.0,
+          "candidate_score": 0.0,
+          "score_delta": -1.0,
+          "change": "new_fail",
+          "hard_failure_added": false
+        }
+      ]
+    }
+  },
+  "attributions": [
+    {
+      "case_id": "val_candidate_improves",
+      "category": "final_response_mismatch",
+      "rule_id": "response.mismatch",
+      "evidence": "final response metric did not meet threshold"
+    },
+    {
+      "case_id": "val_candidate_no_effect",
+      "category": "final_response_mismatch",
+      "rule_id": "response.mismatch",
+      "evidence": "final response metric did not meet threshold"
+    }
+  ],
+  "attribution_counts": {
+    "final_response_mismatch": 2
+  },
+  "optimization": {
+    "status": "SUCCEEDED",
+    "finish_reason": "fake_candidate",
+    "best_prompts": {
+      "system_prompt": "Classify support requests.\n\nReturn a short answer containing the selected queue.\n\n\nOPTIMIZED_CANDIDATE\n"
+    },
+    "rounds": [],
+    "total_cost": 0.0,
+    "error_message": ""
+  },
+  "gate": {
+    "accepted": true,
+    "overfitting": false,
+    "checks": [
+      {
+        "name": "validation_score",
+        "passed": true,
+        "reason": "validation delta=0.3333333333333333, required>=0.1"
+      },
+      {
+        "name": "validation_regression",
+        "passed": true,
+        "reason": "validation did not regress"
+      },
+      {
+        "name": "hard_failures",
+        "passed": true,
+        "reason": "no hard failures"
+      },
+      {
+        "name": "critical_cases",
+        "passed": true,
+        "reason": "critical cases preserved"
+      },
+      {
+        "name": "cost",
+        "passed": true,
+        "reason": "cost budget disabled"
+      },
+      {
+        "name": "duration",
+        "passed": true,
+        "reason": "duration=0.155s, limit=180.0s"
+      },
+      {
+        "name": "overfitting",
+        "passed": true,
+        "reason": "no overfitting signal"
+      }
+    ],
+    "reasons": []
+  },
+  "cost": {
+    "optimizer_cost": 0.0,
+    "external_cost": 0.0,
+    "total_cost": 0.0,
+    "cost_complete": true
+  },
+  "audit": {
+    "seed": 91,
+    "input_hashes": {
+      "system.md": "ba0381d590139a9ac700c0b232e7aa71133419cbba1a26e53e2ad4f0fa5a70c8",
+      "train.evalset.json": "7731436db1444b2b606069724fb1ad0829f2894f733b85488782049254215ead",
+      "val.evalset.json": "fca00392951499c1314b4096f45d31d46be4de155bccf8f7c3ffc4f1527494bc",
+      "optimizer.json": "9399c11247f0dc8c983d83df3d9078bdc87ffa22c37e96bb11a618eece9fdd50",
+      "gate.json": "70fc2ca79c30a2c7e87d4eaaae1e99e093891220b8d339bec78885664e1f217b",
+      "candidate_prompt": "e545989315f5f7ad3b1bed61161cd9118565699f3228fc2976cd454039dd89b5"
+    },
+    "model_name": "fake-model",
+    "num_runs": 1,
+    "case_parallelism": 1,
+    "python_version": "3.14.5",
+    "sdk_version": "unknown",
+    "git_sha": "6a2f7f9bdbe8c697990a6198d1cc28ddcc2552c2",
+    "started_at": "2026-07-27T00:58:58.991468+08:00",
+    "finished_at": "2026-07-27T00:58:59.145985+08:00",
+    "stage_durations": {
+      "pipeline": 0.1545168000011472
+    }
+  },
+  "source_updated": false,
+  "failures": []
+}
diff --git a/examples/optimization/eval_optimize_loop/optimizer.json b/examples/optimization/eval_optimize_loop/optimizer.json
new file mode 100644
index 000000000..2114fd0af
--- /dev/null
+++ b/examples/optimization/eval_optimize_loop/optimizer.json
@@ -0,0 +1,46 @@
+{
+  "evaluate": {
+    "metrics": [
+      {
+        "metric_name": "final_response_avg_score",
+        "threshold": 1.0,
+        "criterion": {
+          "final_response": {
+            "text": {
+              "match": "exact",
+              "case_insensitive": false
+            }
+          }
+        }
+      }
+    ],
+    "num_runs": 1
+  },
+  "optimize": {
+    "eval_case_parallelism": 1,
+    "stop": {
+      "required_metrics": []
+    },
+    "algorithm": {
+      "name": "gepa_reflective",
+      "seed": 91,
+      "reflection_lm": {
+        "model_name": "${TRPC_AGENT_MODEL_NAME}",
+        "base_url": "${TRPC_AGENT_BASE_URL}",
+        "api_key": "${TRPC_AGENT_API_KEY}",
+        "generation_config": {
+          "max_tokens": 1024,
+          "temperature": 0.0
+        }
+      },
+      "module_selector": "round_robin",
+      "reflection_minibatch_size": 2,
+      "skip_perfect_score": false,
+      "max_metric_calls": 12,
+      "max_iterations_without_improvement": 2,
+      "timeout_seconds": 120.0,
+      "max_candidate_proposals": 4,
+      "max_tracked_candidates": 4
+    }
+  }
+}
diff --git a/examples/optimization/eval_optimize_loop/run_pipeline.py b/examples/optimization/eval_optimize_loop/run_pipeline.py
new file mode 100644
index 000000000..aa112df26
--- /dev/null
+++ b/examples/optimization/eval_optimize_loop/run_pipeline.py
@@ -0,0 +1,74 @@
+"""Command-line entry point for the evaluation optimization loop."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import os
+import sys
+from pathlib import Path
+
+if __package__ in (None, ""):  # pragma: no cover
+    sys.path.insert(0, str(Path(__file__).resolve().parents[3]))
+
+from examples.optimization.eval_optimize_loop.loop.models import InputPaths
+from examples.optimization.eval_optimize_loop.loop.models import PipelineOptions
+from examples.optimization.eval_optimize_loop.loop.pipeline import run_pipeline
+
+DEFAULT_ROOT = Path(__file__).parent
+DEFAULT_OUTPUT = DEFAULT_ROOT / "artifacts"
+REAL_MODE = "real"
+DEFAULT_REAL_MODEL_NAME = "real"
+
+
+def _parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--prompt", type=Path, default=DEFAULT_ROOT / "agent/prompts/system.md")
+    parser.add_argument("--train", type=Path, default=DEFAULT_ROOT / "data/train.evalset.json")
+    parser.add_argument("--validation", type=Path, default=DEFAULT_ROOT / "data/val.evalset.json")
+    parser.add_argument("--optimizer", type=Path, default=DEFAULT_ROOT / "optimizer.json")
+    parser.add_argument("--gate", type=Path, default=DEFAULT_ROOT / "gate.json")
+    parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
+    parser.add_argument("--mode", choices=("fake-model", "real", "trace"), default="fake-model")
+    parser.add_argument("--model-name")
+    parser.add_argument("--trace-file", type=Path)
+    parser.add_argument("--write-back", action="store_true")
+    parser.add_argument("--fake-judge", action="store_true")
+    return parser
+
+
+def _model_name(mode: str, explicit_name: str | None) -> str:
+    if explicit_name:
+        return explicit_name
+    if mode == REAL_MODE:
+        return os.getenv("TRPC_AGENT_MODEL_NAME", DEFAULT_REAL_MODEL_NAME)
+    return "fake-model"
+
+
+def main() -> int:
+    args = _parser().parse_args()
+    paths = InputPaths(
+        prompt_path=args.prompt,
+        train_path=args.train,
+        validation_path=args.validation,
+        optimizer_path=args.optimizer,
+        gate_path=args.gate,
+    )
+    result = asyncio.run(
+        run_pipeline(
+            PipelineOptions(
+                paths=paths,
+                output_dir=args.output,
+                mode=args.mode,
+                model_name=_model_name(args.mode, args.model_name),
+                trace_file=args.trace_file,
+                fake_judge=args.fake_judge,
+                write_back=args.write_back,
+            )))
+    print(f"{result.report.status}: {result.json_path}")
+    print(result.markdown_path)
+    return 0 if result.report.gate.accepted and not result.report.failures else 1
+
+
+if __name__ == "__main__":  # pragma: no cover
+    raise SystemExit(main())
diff --git a/tests/evaluation/test_eval_optimize_loop_analysis.py b/tests/evaluation/test_eval_optimize_loop_analysis.py
new file mode 100644
index 000000000..66965720a
--- /dev/null
+++ b/tests/evaluation/test_eval_optimize_loop_analysis.py
@@ -0,0 +1,155 @@
+"""Unit tests for attribution, deltas, and acceptance gates."""
+
+from examples.optimization.eval_optimize_loop.loop.analysis import GateInput
+from examples.optimization.eval_optimize_loop.loop.analysis import attribute_case
+from examples.optimization.eval_optimize_loop.loop.analysis import compare_snapshots
+from examples.optimization.eval_optimize_loop.loop.analysis import evaluate_gate
+from examples.optimization.eval_optimize_loop.loop.models import Attribution
+from examples.optimization.eval_optimize_loop.loop.models import CaseSnapshot
+from examples.optimization.eval_optimize_loop.loop.models import CostSummary
+from examples.optimization.eval_optimize_loop.loop.models import EvaluationSnapshot
+from examples.optimization.eval_optimize_loop.loop.models import FailureCategory
+from examples.optimization.eval_optimize_loop.loop.models import GateConfig
+from examples.optimization.eval_optimize_loop.loop.models import InvocationSnapshot
+from examples.optimization.eval_optimize_loop.loop.models import SplitName
+
+
+def _case(case_id: str, passed: bool, score: float | None) -> CaseSnapshot:
+    return CaseSnapshot(
+        case_id=case_id,
+        split=SplitName.VALIDATION,
+        passed=passed,
+        hard_failure=score is None,
+        metric_scores={"final_response_avg_score": score},
+        metric_statuses={"final_response_avg_score": "PASSED" if passed else "FAILED"},
+        actual=[InvocationSnapshot(final_text="actual")],
+        expected=[InvocationSnapshot(final_text="expected")],
+    )
+
+
+def _snapshot(cases: list[CaseSnapshot], score: float | None) -> EvaluationSnapshot:
+    return EvaluationSnapshot(
+        split=SplitName.VALIDATION,
+        primary_metric="final_response_avg_score",
+        primary_score=score,
+        pass_rate=sum(case.passed for case in cases) / len(cases) if cases else 0.0,
+        metric_scores={"final_response_avg_score": score},
+        cases=cases,
+        duration_seconds=0.1,
+    )
+
+
+def _gate(**overrides) -> GateConfig:
+    values = {
+        "primary_metric": "final_response_avg_score",
+        "min_score_delta": 0.1,
+    }
+    values.update(overrides)
+    return GateConfig(**values)
+
+
+def _gate_input(
+    config: GateConfig,
+    baseline: EvaluationSnapshot,
+    candidate: EvaluationSnapshot,
+    cost: CostSummary | None = None,
+) -> GateInput:
+    return GateInput(
+        config=config,
+        train_delta=compare_snapshots(baseline, candidate),
+        validation_delta=compare_snapshots(baseline, candidate),
+        baseline_validation=baseline,
+        candidate_validation=candidate,
+        cost=cost or CostSummary(optimizer_cost=0.0, total_cost=0.0, cost_complete=True),
+        duration_seconds=1.0,
+    )
+
+
+def test_compare_snapshots_reports_new_pass_and_new_fail():
+    baseline = _snapshot([_case("a", False, 0.0), _case("b", True, 1.0)], 0.5)
+    candidate = _snapshot([_case("a", True, 1.0), _case("b", False, 0.0)], 0.5)
+
+    delta = compare_snapshots(baseline, candidate)
+
+    assert {item.change.value for item in delta.cases} == {"new_pass", "new_fail"}
+
+
+def test_missing_candidate_case_is_hard_failure():
+    baseline = _snapshot([_case("a", True, 1.0)], 1.0)
+    candidate = _snapshot([], None)
+
+    delta = compare_snapshots(baseline, candidate)
+
+    assert delta.cases[0].hard_failure_added is True
+
+
+def test_attribution_prefers_tool_argument_category():
+    case = _case("tool", False, 0.0)
+    case = case.model_copy(
+        update={
+            "actual": [InvocationSnapshot(tool_calls=[{
+                "name": "lookup",
+                "args": {
+                    "id": 1
+                }
+            }])],
+            "expected": [InvocationSnapshot(tool_calls=[{
+                "name": "lookup",
+                "args": {
+                    "id": 2
+                }
+            }])],
+        })
+
+    result = attribute_case(case, "final_response_avg_score")
+
+    assert isinstance(result, Attribution)
+    assert result.category == FailureCategory.TOOL_ARGUMENT
+
+
+def test_gate_rejects_score_regression_and_overfitting():
+    baseline = _snapshot([_case("a", False, 0.0)], 0.0)
+    candidate = _snapshot([_case("a", False, 0.0)], 0.0)
+    config = _gate(min_score_delta=0.1)
+
+    decision = evaluate_gate(_gate_input(config, baseline, candidate))
+
+    assert decision.accepted is False
+    assert "validation delta" in decision.reasons[0]
+
+
+def test_gate_rejects_regression_even_with_negative_minimum_delta():
+    baseline = _snapshot([_case("a", True, 1.0)], 1.0)
+    candidate = _snapshot([_case("a", False, 0.0)], 0.0)
+
+    decision = evaluate_gate(_gate_input(_gate(min_score_delta=-1.0), baseline, candidate))
+
+    assert any(check.name == "validation_regression" and not check.passed for check in decision.checks)
+
+
+def test_gate_rejects_configured_hard_case_and_incomplete_cost():
+    baseline = _snapshot([_case("a", True, 1.0)], 1.0)
+    candidate = _snapshot([_case("a", False, 0.0)], 0.0)
+    config = _gate(min_score_delta=-1.0, hard_case_ids=["a"], max_total_cost=1.0)
+
+    decision = evaluate_gate(
+        _gate_input(
+            config,
+            baseline,
+            candidate,
+            CostSummary(optimizer_cost=0.1, total_cost=0.1, cost_complete=False),
+        ))
+
+    assert decision.accepted is False
+    assert any(check.name == "hard_failures" and not check.passed for check in decision.checks)
+    assert any(check.name == "cost" and not check.passed for check in decision.checks)
+
+
+def test_gate_rejects_missing_critical_case():
+    baseline = _snapshot([_case("a", True, 1.0)], 1.0)
+    candidate = _snapshot([], None)
+    config = _gate(min_score_delta=-1.0, critical_case_ids=["a"])
+
+    decision = evaluate_gate(_gate_input(config, baseline, candidate))
+
+    assert any(check.name == "critical_cases" and not check.passed for check in decision.checks)
diff --git a/tests/evaluation/test_eval_optimize_loop_cli.py b/tests/evaluation/test_eval_optimize_loop_cli.py
new file mode 100644
index 000000000..cf8e8ff02
--- /dev/null
+++ b/tests/evaluation/test_eval_optimize_loop_cli.py
@@ -0,0 +1,106 @@
+"""CLI smoke tests for the offline optimization loop."""
+
+import subprocess
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+
+from examples.optimization.eval_optimize_loop import run_pipeline
+
+
+def test_cli_fake_mode_runs_without_api_key(tmp_path):
+    command = [
+        sys.executable,
+        "examples/optimization/eval_optimize_loop/run_pipeline.py",
+        "--output",
+        str(tmp_path),
+        "--fake-judge",
+    ]
+
+    completed = subprocess.run(command, capture_output=True, text=True, check=True)
+
+    assert "optimization_report.json" in completed.stdout
+    assert (tmp_path / "optimization_report.json").is_file()
+
+
+def test_real_model_name_uses_environment(monkeypatch):
+    monkeypatch.setenv("TRPC_AGENT_MODEL_NAME", "deepseek-chat")
+
+    assert run_pipeline._model_name("real", None) == "deepseek-chat"
+    assert run_pipeline._model_name("real", "explicit-model") == "explicit-model"
+
+
+def test_main_accepts_completed_report(monkeypatch, tmp_path, capsys):
+    report = SimpleNamespace(
+        gate=SimpleNamespace(accepted=True),
+        failures=[],
+        status="ACCEPTED",
+    )
+    result = SimpleNamespace(
+        report=report,
+        json_path=tmp_path / "optimization_report.json",
+        markdown_path=tmp_path / "optimization_report.md",
+    )
+
+    async def fake_run(options):
+        assert options.model_name == "fake-model"
+        return result
+
+    monkeypatch.setattr(sys, "argv", ["run_pipeline.py", "--output", str(tmp_path)])
+    monkeypatch.setattr(run_pipeline, "run_pipeline", fake_run)
+
+    assert run_pipeline.main() == 0
+    assert "ACCEPTED" in capsys.readouterr().out
+
+
+def test_main_rejects_gate_failure(monkeypatch, tmp_path):
+    report = SimpleNamespace(
+        gate=SimpleNamespace(accepted=False),
+        failures=[],
+        status="REJECTED",
+    )
+    result = SimpleNamespace(
+        report=report,
+        json_path=Path(tmp_path) / "optimization_report.json",
+        markdown_path=Path(tmp_path) / "optimization_report.md",
+    )
+
+    async def fake_run(options):
+        return result
+
+    monkeypatch.setattr(sys, "argv", ["run_pipeline.py"])
+    monkeypatch.setattr(run_pipeline, "run_pipeline", fake_run)
+
+    assert run_pipeline.main() == 1
+
+
+def test_cli_pipeline_failure_returns_nonzero(tmp_path):
+    command = [
+        sys.executable,
+        "examples/optimization/eval_optimize_loop/run_pipeline.py",
+        "--output",
+        str(tmp_path),
+        "--train",
+        str(tmp_path / "missing.evalset.json"),
+    ]
+
+    completed = subprocess.run(command, capture_output=True, text=True)
+
+    assert completed.returncode != 0
+
+
+def test_cli_gate_rejection_returns_nonzero(tmp_path):
+    gate = tmp_path / "gate.json"
+    gate.write_text('{"primary_metric":"final_response_avg_score","min_score_delta":2.0}', encoding="utf-8")
+    command = [
+        sys.executable,
+        "examples/optimization/eval_optimize_loop/run_pipeline.py",
+        "--output",
+        str(tmp_path),
+        "--gate",
+        str(gate),
+    ]
+
+    completed = subprocess.run(command, capture_output=True, text=True)
+
+    assert completed.returncode != 0
diff --git a/tests/evaluation/test_eval_optimize_loop_evaluation.py b/tests/evaluation/test_eval_optimize_loop_evaluation.py
new file mode 100644
index 000000000..799ffd57d
--- /dev/null
+++ b/tests/evaluation/test_eval_optimize_loop_evaluation.py
@@ -0,0 +1,178 @@
+"""Tests for the evaluation adapter and leakage checks."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+from trpc_agent_sdk.evaluation import EvalStatus
+from trpc_agent_sdk.evaluation._eval_case import Invocation
+from trpc_agent_sdk.evaluation._eval_result import EvalCaseResult
+from trpc_agent_sdk.evaluation._eval_result import EvalMetricResult
+from trpc_agent_sdk.evaluation._eval_result import EvalMetricResultPerInvocation
+from trpc_agent_sdk.types import Content
+from trpc_agent_sdk.types import Part
+
+from examples.optimization.eval_optimize_loop.loop.evaluation import EvaluationRequest
+from examples.optimization.eval_optimize_loop.loop.evaluation import evaluate_split
+from examples.optimization.eval_optimize_loop.loop.evaluation import load_eval_set
+from examples.optimization.eval_optimize_loop.loop.evaluation import _snapshot_case
+from examples.optimization.eval_optimize_loop.loop.evaluation import _dataset_for_sdk
+from examples.optimization.eval_optimize_loop.loop.evaluation import validate_inputs
+from examples.optimization.eval_optimize_loop.loop import evaluation as evaluation_module
+from examples.optimization.eval_optimize_loop.loop.models import InputPaths
+from examples.optimization.eval_optimize_loop.loop.models import SplitName
+
+ROOT = Path("examples/optimization/eval_optimize_loop")
+PROMPT = ROOT / "agent" / "prompts" / "system.md"
+TRAIN = ROOT / "data" / "train.evalset.json"
+VALIDATION = ROOT / "data" / "val.evalset.json"
+OPTIMIZER = ROOT / "optimizer.json"
+GATE = ROOT / "gate.json"
+
+
+def _paths() -> InputPaths:
+    return InputPaths(
+        prompt_path=PROMPT,
+        train_path=TRAIN,
+        validation_path=VALIDATION,
+        optimizer_path=OPTIMIZER,
+        gate_path=GATE,
+    )
+
+
+def _expected_response(query: str) -> str:
+    if "invoice" in query or "payment" in query:
+        return '{"queue":"billing"}'
+    if "password" in query:
+        return '{"queue":"account"}'
+    return '{"queue":"technical"}'
+
+
+async def _passing_agent(query: str) -> str:
+    return _expected_response(query)
+
+
+async def _failing_agent(query: str) -> str:
+    del query
+    return '{"queue":"unknown"}'
+
+
+def test_validate_inputs_rejects_split_content_leakage(tmp_path):
+    train = load_eval_set(TRAIN)
+    validation = load_eval_set(VALIDATION)
+    duplicate = train.eval_cases[0].model_copy(update={"eval_id": "duplicate"})
+    duplicate_path = tmp_path / "duplicate.evalset.json"
+    duplicate_path.write_text(
+        validation.model_copy(update={
+            "eval_cases": [duplicate]
+        }).model_dump_json(),
+        encoding="utf-8",
+    )
+    paths = _paths().model_copy(update={"validation_path": duplicate_path})
+
+    with pytest.raises(ValueError, match="overlap"):
+        validate_inputs(paths)
+
+
+def test_validate_inputs_returns_hashes_and_gate():
+    bundle, optimizer, gate = validate_inputs(_paths())
+
+    assert bundle.prompt_path == PROMPT.resolve()
+    assert set(bundle.hashes) == {
+        "prompt",
+        "train",
+        "validation",
+        "optimizer",
+        "gate",
+    }
+    assert optimizer.evaluate.get_eval_metrics()[0].metric_name == gate.primary_metric
+
+
+def test_dataset_outside_cwd_is_copied_to_managed_temp(tmp_path):
+    source = tmp_path / "outside.evalset.json"
+    source.write_text("{}", encoding="utf-8")
+    temp_dir = tmp_path / "temp"
+    temp_dir.mkdir()
+
+    dataset = _dataset_for_sdk(source, temp_dir)
+
+    assert Path(dataset).resolve() == (temp_dir / source.name).resolve()
+    assert (temp_dir / source.name).read_text(encoding="utf-8") == "{}"
+
+
+@pytest.mark.asyncio
+async def test_evaluate_split_keeps_passing_result():
+    snapshot = await evaluate_split(EvaluationRequest(TRAIN, OPTIMIZER, SplitName.TRAIN, _passing_agent))
+
+    assert snapshot.primary_score == pytest.approx(1.0)
+    assert snapshot.pass_rate == pytest.approx(1.0)
+    assert all(case.passed for case in snapshot.cases)
+
+
+@pytest.mark.asyncio
+async def test_evaluate_split_keeps_failed_case_result():
+    snapshot = await evaluate_split(EvaluationRequest(VALIDATION, OPTIMIZER, SplitName.VALIDATION, _failing_agent))
+
+    assert snapshot.primary_score == pytest.approx(0.0)
+    assert snapshot.pass_rate == pytest.approx(0.0)
+    assert all(not case.passed for case in snapshot.cases)
+    assert all(not case.hard_failure for case in snapshot.cases)
+
+
+@pytest.mark.asyncio
+async def test_evaluate_split_reraises_unrelated_assertion(monkeypatch):
+
+    class _BuggyExecutor:
+
+        async def evaluate(self):
+            raise AssertionError("unexpected SDK assertion")
+
+        def get_result(self):
+            return object()
+
+    monkeypatch.setattr(
+        evaluation_module.AgentEvaluator,
+        "get_executer",
+        lambda *args, **kwargs: _BuggyExecutor(),
+    )
+
+    with pytest.raises(AssertionError, match="unexpected SDK assertion"):
+        await evaluate_split(EvaluationRequest(TRAIN, OPTIMIZER, SplitName.TRAIN, _passing_agent))
+
+
+def test_validate_inputs_rejects_unknown_critical_case(tmp_path):
+    gate = tmp_path / "gate.json"
+    gate.write_text(
+        '{"primary_metric":"final_response_avg_score","critical_case_ids":["missing"]}',
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="belong to validation"):
+        validate_inputs(_paths().model_copy(update={"gate_path": gate}))
+
+
+def test_metric_not_evaluated_is_a_hard_failure():
+    invocation = Invocation(
+        user_content=Content(role="user", parts=[Part.from_text(text="query")]),
+        final_response=Content(role="model", parts=[Part.from_text(text="answer")]),
+    )
+    metric = EvalMetricResult(
+        metric_name="final_response_avg_score",
+        threshold=1.0,
+        score=None,
+        eval_status=EvalStatus.NOT_EVALUATED,
+    )
+    case = EvalCaseResult(
+        eval_set_id="set",
+        eval_id="case",
+        final_eval_status=EvalStatus.PASSED,
+        overall_eval_metric_results=[metric],
+        eval_metric_result_per_invocation=[EvalMetricResultPerInvocation(actual_invocation=invocation)],
+        session_id="session",
+    )
+
+    snapshot = _snapshot_case("case", SplitName.VALIDATION, [case], metric.metric_name)
+
+    assert snapshot.hard_failure is True
+    assert snapshot.passed is False
diff --git a/tests/evaluation/test_eval_optimize_loop_pipeline.py b/tests/evaluation/test_eval_optimize_loop_pipeline.py
new file mode 100644
index 000000000..c5fdf1346
--- /dev/null
+++ b/tests/evaluation/test_eval_optimize_loop_pipeline.py
@@ -0,0 +1,238 @@
+"""Integration tests for the fake optimization loop."""
+
+import asyncio
+import json
+import os
+import time
+from pathlib import Path
+
+import pytest
+from trpc_agent_sdk.evaluation._target_prompt import _RollbackError
+
+from examples.optimization.eval_optimize_loop.loop.models import InputPaths
+from examples.optimization.eval_optimize_loop.loop.models import PipelineOptions
+from examples.optimization.eval_optimize_loop.loop.pipeline import _write_back_and_report
+from examples.optimization.eval_optimize_loop.loop.pipeline import _failure_result
+from examples.optimization.eval_optimize_loop.loop.pipeline import _sdk_version
+from examples.optimization.eval_optimize_loop.loop.pipeline import run_pipeline
+from examples.optimization.eval_optimize_loop.loop import pipeline as pipeline_module
+from examples.optimization.eval_optimize_loop.loop import reporting as reporting_module
+from examples.optimization.eval_optimize_loop.loop.evaluation import validate_inputs
+from examples.optimization.eval_optimize_loop.loop.reporting import write_reports
+
+ROOT = Path("examples/optimization/eval_optimize_loop")
+
+
+def _options(tmp_path: Path, **overrides) -> PipelineOptions:
+    values = {
+        "paths":
+        InputPaths(
+            prompt_path=ROOT / "agent/prompts/system.md",
+            train_path=ROOT / "data/train.evalset.json",
+            validation_path=ROOT / "data/val.evalset.json",
+            optimizer_path=ROOT / "optimizer.json",
+            gate_path=ROOT / "gate.json",
+        ),
+        "output_dir":
+        tmp_path,
+        "mode":
+        "fake-model",
+    }
+    values.update(overrides)
+    return PipelineOptions(**values)
+
+
+def test_fake_pipeline_writes_auditable_json_and_markdown(tmp_path):
+    result = asyncio.run(run_pipeline(_options(tmp_path, fake_judge=True)))
+
+    assert result.json_path.is_file()
+    assert result.markdown_path.is_file()
+    payload = json.loads(result.json_path.read_text(encoding="utf-8"))
+    assert payload["gate"]["checks"]
+    assert payload["audit"]["input_hashes"]
+    assert "Optimization" in result.markdown_path.read_text(encoding="utf-8")
+
+
+def test_candidate_prompt_is_restored_after_replay(tmp_path):
+    work_prompt = tmp_path / "work/prompts/system.md"
+    source = (ROOT / "agent/prompts/system.md").read_text(encoding="utf-8")
+
+    asyncio.run(run_pipeline(_options(tmp_path)))
+
+    assert work_prompt.read_text(encoding="utf-8") == source
+
+
+def test_trace_mode_replays_recorded_cases(tmp_path):
+    options = _options(
+        tmp_path,
+        mode="trace",
+        trace_file=ROOT / "data/fake_trace.json",
+    )
+
+    result = asyncio.run(run_pipeline(options))
+
+    assert result.report.baseline
+    assert result.report.candidate
+    assert (tmp_path / "work/trace/baseline/train.evalset.json").is_file()
+    assert (tmp_path / "work/trace/baseline/validation.evalset.json").is_file()
+    assert (tmp_path / "work/trace/candidate/train.evalset.json").is_file()
+    assert (tmp_path / "work/trace/candidate/validation.evalset.json").is_file()
+
+
+def test_pipeline_failure_is_reported_without_prompt_write(tmp_path):
+    options = _options(tmp_path,
+                       paths=InputPaths(
+                           prompt_path=ROOT / "agent/prompts/system.md",
+                           train_path=ROOT / "missing.evalset.json",
+                           validation_path=ROOT / "data/val.evalset.json",
+                           optimizer_path=ROOT / "optimizer.json",
+                           gate_path=ROOT / "gate.json",
+                       ))
+
+    result = asyncio.run(run_pipeline(options))
+
+    assert result.report.status == "REJECTED"
+    assert result.report.failures
+    assert result.json_path.is_file()
+
+
+def test_failure_report_keeps_validated_audit_context(tmp_path, monkeypatch):
+
+    async def fail_after_validation(*args, **kwargs):
+        raise RuntimeError("evaluation failed")
+
+    monkeypatch.setattr(pipeline_module, "_evaluate_pair", fail_after_validation)
+    optimizer = json.loads((ROOT / "optimizer.json").read_text(encoding="utf-8"))
+    optimizer["evaluate"]["num_runs"] = 2
+    optimizer_path = tmp_path / "optimizer.json"
+    optimizer_path.write_text(json.dumps(optimizer), encoding="utf-8")
+    options = _options(
+        tmp_path,
+        paths=_options(tmp_path).paths.model_copy(update={"optimizer_path": optimizer_path}),
+    )
+
+    result = asyncio.run(run_pipeline(options))
+
+    assert result.report.status == "REJECTED"
+    assert result.report.audit.input_hashes
+    assert result.report.audit.num_runs == 2
+    assert result.report.audit.case_parallelism == 1
+
+
+def test_pipeline_timeout_is_reported(tmp_path, monkeypatch):
+
+    async def stall_evaluation(*args, **kwargs):
+        await asyncio.sleep(1)
+
+    monkeypatch.setattr(pipeline_module, "_evaluate_pair", stall_evaluation)
+    gate = json.loads((ROOT / "gate.json").read_text(encoding="utf-8"))
+    gate["max_duration_seconds"] = 0.01
+    gate_path = tmp_path / "gate.json"
+    gate_path.write_text(json.dumps(gate), encoding="utf-8")
+    options = _options(
+        tmp_path,
+        paths=_options(tmp_path).paths.model_copy(update={"gate_path": gate_path}),
+    )
+
+    started = time.monotonic()
+    result = asyncio.run(run_pipeline(options))
+
+    assert time.monotonic() - started < 1
+    assert result.report.status == "REJECTED"
+    assert result.report.failures[0].startswith("TimeoutError:")
+
+
+def test_rollback_failure_details_are_audited(tmp_path):
+    error = _RollbackError([("system_prompt", RuntimeError("rollback failed"))])
+
+    result = _failure_result(_options(tmp_path), time.monotonic(), error)
+
+    assert "rollback failed" in result.report.failures[0]
+
+
+def test_sdk_version_is_available():
+    assert _sdk_version() != "unknown"
+
+
+def test_write_back_report_matches_updated_prompt(tmp_path):
+    prompt_path = tmp_path / "system.md"
+    original = (ROOT / "agent/prompts/system.md").read_text(encoding="utf-8")
+    prompt_path.write_text(original, encoding="utf-8")
+    options = _options(
+        tmp_path / "output",
+        paths=_options(tmp_path).paths.model_copy(update={"prompt_path": prompt_path}),
+    )
+
+    result = asyncio.run(run_pipeline(options))
+    bundle, _, _ = validate_inputs(options.paths)
+    asyncio.run(
+        _write_back_and_report(
+            result.report,
+            bundle,
+            {"system_prompt": original + "\n\nOPTIMIZED_CANDIDATE\n"},
+            options.model_copy(update={
+                "write_back": True,
+                "mode": "real"
+            }),
+        ))
+    payload = json.loads(result.json_path.read_text(encoding="utf-8"))
+
+    assert result.report.gate.accepted is True
+    assert payload["source_updated"] is True
+    assert "OPTIMIZED_CANDIDATE" in prompt_path.read_text(encoding="utf-8")
+
+
+def test_fake_mode_write_back_does_not_touch_prompt(tmp_path):
+    prompt_path = tmp_path / "system.md"
+    original = (ROOT / "agent/prompts/system.md").read_text(encoding="utf-8")
+    prompt_path.write_text(original, encoding="utf-8")
+    options = _options(
+        tmp_path / "output",
+        paths=_options(tmp_path).paths.model_copy(update={"prompt_path": prompt_path}),
+        write_back=True,
+    )
+
+    result = asyncio.run(run_pipeline(options))
+
+    assert result.report.status == "REJECTED"
+    assert result.report.failures[0].startswith("ValueError:")
+    assert prompt_path.read_text(encoding="utf-8") == original
+
+
+def test_report_audit_uses_configured_num_runs(tmp_path):
+    optimizer = json.loads((ROOT / "optimizer.json").read_text(encoding="utf-8"))
+    optimizer["evaluate"]["num_runs"] = 2
+    optimizer_path = tmp_path / "optimizer.json"
+    optimizer_path.write_text(json.dumps(optimizer), encoding="utf-8")
+    options = _options(
+        tmp_path / "output",
+        paths=_options(tmp_path).paths.model_copy(update={"optimizer_path": optimizer_path}),
+    )
+
+    result = asyncio.run(run_pipeline(options))
+
+    assert result.report.audit.num_runs == 2
+
+
+def test_report_pair_is_rolled_back_when_publish_fails(tmp_path, monkeypatch):
+    result = asyncio.run(run_pipeline(_options(tmp_path, fake_judge=True)))
+    json_before = result.json_path.read_text(encoding="utf-8")
+    markdown_before = result.markdown_path.read_text(encoding="utf-8")
+    real_replace = os.replace
+    failed = False
+
+    def fail_markdown_once(source, destination):
+        nonlocal failed
+        if Path(destination) == result.markdown_path and not failed:
+            failed = True
+            raise OSError("markdown publish failed")
+        real_replace(source, destination)
+
+    monkeypatch.setattr(reporting_module.os, "replace", fail_markdown_once)
+    result.report.status = "REJECTED"
+
+    with pytest.raises(OSError, match="markdown publish failed"):
+        write_reports(result.report, tmp_path)
+
+    assert result.json_path.read_text(encoding="utf-8") == json_before
+    assert result.markdown_path.read_text(encoding="utf-8") == markdown_before