feat(examples): add counterfactual trace diagnosis optimization loop

examples/optimization/counterfactual_trace_loop/DESIGN.md

@@ -0,0 +1,20 @@
+# 方案设计
+
+本方案将评测优化闭环的核心从“根据失败文本猜测原因”改为“通过同一评测器验证最小行为反事实”。系统先检查训练集、验证集、提示词和预算契约，并将样例标记为 trusted、suspect 或 invalid。只有 trusted 且具有修复证据的失败能够驱动优化。
+
+对于失败轨迹，系统深拷贝实际对话，分别替换最终回复、工具名称、工具参数或受限组合，保持预期对话不变，再调用 `AgentEvaluator.get_executer()` 评分。单一干预修复失败时形成强归因；只有组合有效时判定为复合失败。数据、评测器和基础设施异常不进入优化摘要。若局部替换与原工具响应不一致，证据会降低置信度。
+
+归因结果映射到 router、skill 和 system 三类 `TargetPrompt`。真实优化入口固定使用 `AgentOptimizer.optimize(update_source=False)`，候选必须在完整验证集重新评测，新增退化再次执行反事实诊断。Gate 采用 all-must-pass，检查验证增益、可信子集、hard fail、关键样例、严重度、成本、耗时和证据。只有 gate 接受且显式传入 `--apply` 时才调用 `TargetPrompt.write_all()`，并记录写前写后哈希。报告保存轨迹证据、候选差异、拒绝原因、随机种子及复现命令。
+
+## 技术附录
+
+### 与现有方案的边界
+
+PR #159 主要依据失败文本和预期/实际轨迹的静态差异归因；PR #161 重点在优化候选回放、bootstrap 与 Pareto 决策。本方案不把静态差异直接视为原因，而是把它转化为局部反事实干预，再由同一个 `AgentEvaluator` 重评。只有可重复观察到 metric 修复的干预才形成归因证据，且该机制同时用于 baseline 失败和 candidate 新增退化。
+
+- 反事实归因：结论来自真实 metric delta，不依赖 case ID、failure reason 或人工标签。
+- Prompt actionability：只有 agent behavior failure 能够选择优化表面。
+- Gate：关键检查必须全部通过，证据不足时以 `NEEDS_REVIEW` 拒绝。
+- 防过拟合：训练失败驱动优化，完整 validation 决策，新退化独立诊断。
+- 审计：记录输入和提示词哈希、seed、耗时、成本、候选及每项 gate 证据。
+- 限制：局部轨迹替换可能形成现实中不可执行的状态，因此显式记录一致性并降低置信度。

examples/optimization/counterfactual_trace_loop/README.md

@@ -0,0 +1,39 @@
+# Trust-Aware Counterfactual Trace Diagnosis Loop
+
+This example closes evaluation and prompt optimization with evidence from trace interventions. Unlike loops that classify a failure from reason text or sample metadata, it deep-copies the actual trace, changes one execution surface, and sends the counterfactual `EvalCase` through the same public `AgentEvaluator` metrics. Evaluation-data, evaluator, and infrastructure failures are excluded from prompt optimization.
+
+## Distinction from other Issue #91 proposals
+
+PR #159 uses rule-first attribution from failure text and static expected/actual trace differences. PR #161 replays optimizer candidates and combines static trace/rubric attribution with bootstrap and Pareto selection. This example addresses a different question: which minimal trace intervention causally repairs the failing metric when the same `AgentEvaluator` evaluates it again?
+
+The attribution evidence is therefore an observed before/after metric delta from `replace_final_response`, `replace_tool_name`, `replace_tool_arguments`, or a bounded combination. Static trace differences can propose an intervention, but cannot by themselves establish the diagnosis. The same mechanism is reused after candidate validation to locate regressions. No case ID, expected failure label, attribution hint, or hand-authored metric score participates in the decision.
+
+## Quick start
+
+```bash
+python examples/optimization/counterfactual_trace_loop/run_counterfactual_probe.py
+python examples/optimization/counterfactual_trace_loop/run_pipeline.py --mode fake
+python examples/optimization/counterfactual_trace_loop/run_pipeline.py --mode trace
+python examples/optimization/counterfactual_trace_loop/run_pipeline.py --mode fake --candidate-profile accepted
+python examples/optimization/counterfactual_trace_loop/run_pipeline.py --mode fake --candidate-profile ineffective
+```
+
+Both fake and trace modes need no API key. Trace mode replays `actual_conversation`; `conversation` remains the expected trace. The fake optimizer is prompt-sensitive and introduces an intentionally broad billing rule without reading case IDs.
+
+## Real optimizer
+
+Fake and trace modes are fully runnable and verified end to end. `pipeline.optimizer.run_real_optimizer()` wires the public `AgentOptimizer.optimize(..., update_source=False)` API to selected `TargetPrompt` files and is verified with a mock/spy, including actionable-case filtering and unified optimization fields. It has not been run against a real model in this example. Production use requires a business `call_agent`, model credentials, and a trace-capture adapter for candidate regression diagnosis.
+
+## Write-back
+
+`--apply` is necessary but not sufficient. Files are written only after every gate check passes. The implementation records baseline hashes, calls `TargetPrompt.write_all()` once, verifies changed hashes, and restores the baseline on failure. The bundled candidate is rejected, so running with `--apply` leaves all source prompts unchanged.
+
+## Outputs
+
+- `prototype_output/counterfactual_probe.{json,md}`: feasibility evidence.
+- `sample_output/optimizer_failure_digest.json`: actionable and excluded failures.
+- `sample_output/optimization_report.{json,md}`: baseline, reliability, attribution, candidate deltas, regression diagnosis, gate, cost, duration, hashes, and reproduction command.
+
+## Known limits
+
+Counterfactual traces can create states that a real agent could not produce, especially when a tool name changes but its response does not. The loop therefore restricts combinations, validates trace shape, exposes invalid interventions, and rejects insufficient evidence. Exact metrics are deterministic; LLM judges require repeated sampling before a case can be trusted. Windows absolute evalset paths are converted to relative paths because the current SDK interprets a drive-letter colon as a case selector.

examples/optimization/counterfactual_trace_loop/__init__.py

@@ -0,0 +1 @@
+"""Trust-aware counterfactual trace optimization example."""

examples/optimization/counterfactual_trace_loop/fake/__init__.py

@@ -0,0 +1 @@
+"""Deterministic semantic fake components."""

examples/optimization/counterfactual_trace_loop/fake/model.py

@@ -0,0 +1,64 @@
+"""Prompt-sensitive fake behavior generator; never reads eval IDs."""
+
+from __future__ import annotations
+
+import re
+
+from trpc_agent_sdk.evaluation import EvalCase, Invocation
+
+
+def user_text(case: EvalCase) -> str:
+    return " ".join(part.text or "" for part in case.conversation[0].user_content.parts).lower()
+
+
+def generate_trace(case: EvalCase, prompts: dict[str, str]) -> EvalCase:
+    """Generate behavior from user semantics and prompt rules."""
+    changed = case.model_copy(deep=True)
+    text = user_text(case)
+    router = prompts.get("router_prompt", "")
+    skill = prompts.get("skill_prompt", "")
+    system = prompts.get("system_prompt", "")
+    tool_name = None
+    tool_args = None
+    if "shipping" in text or "shipment" in text:
+        tracking = re.search(r"t-\d+", text).group(0).upper()
+        tool_name, tool_args = "track_shipment", {"tracking_id": tracking}
+        response = f'{{"tracking_id":"{tracking}","status":"in_transit"}}'
+    elif "billing" in text or "invoice" in text:
+        invoice = re.search(r"i-\d+", text).group(0).upper()
+        tool_name = "create_refund" if "BILLING_TO_REFUND=ON" in router else "get_invoice"
+        tool_args = {"invoice_id": invoice}
+        response = f'{{"invoice_id":"{invoice}","status":"open"}}'
+    else:
+        order_match = re.search(r"o-\d+", text)
+        if order_match is None:
+            response = '{"status":"unsupported"}'
+            raw = {
+                "user_content": case.conversation[0].user_content.model_dump(mode="json"),
+                "final_response": {"role": "model", "parts": [{"text": response}]},
+            }
+            changed.actual_conversation = [Invocation.model_validate(raw)]
+            return changed
+        order = order_match.group(0).upper()
+        if "summarize" in text:
+            response = (
+                f'{{"status":"pending","order_id":"{order}"}}'
+                if "JSON_STATUS=ALWAYS_REQUIRED" in system
+                else f'{{"order_id":"{order}"}}'
+            )
+        else:
+            reason = "duplicate" if "duplicate" in text else "damaged"
+            strict = "REFUND_ROUTE=STRICT" in router
+            tool_name = "create_refund" if strict or reason == "duplicate" else "get_invoice"
+            tool_args = {"order_id": order}
+            if "REFUND_REASON=REQUIRED" in skill or reason == "damaged":
+                tool_args["reason"] = reason
+            response = f'{{"status":"submitted","order_id":"{order}"}}'
+    raw = {
+        "user_content": case.conversation[0].user_content.model_dump(mode="json"),
+        "final_response": {"role": "model", "parts": [{"text": response}]},
+    }
+    if tool_name:
+        raw["intermediate_data"] = {"tool_uses": [{"name": tool_name, "args": tool_args}]}
+    changed.actual_conversation = [Invocation.model_validate(raw)]
+    return changed

examples/optimization/counterfactual_trace_loop/fake/optimizer.py

@@ -0,0 +1,56 @@
+"""Fake optimizer with the same candidate contract as the real adapter."""
+
+from __future__ import annotations
+
+
+async def optimize(prompts: dict[str, str], targets: list[str], profile: str = "overfit") -> dict:
+    candidate = dict(prompts)
+    if profile == "ineffective":
+        return {
+            "total_rounds": 1,
+            "rounds": [
+                {
+                    "round": 1,
+                    "profile": profile,
+                    "targets": [],
+                    "candidate_prompts": candidate,
+                    "cost": 0.0,
+                    "tokens": 0,
+                    "duration_seconds": 0.0,
+                }
+            ],
+            "best_prompts": candidate,
+            "cost": 0.0,
+            "tokens": 0,
+            "seed": 42,
+            "candidate_profile": profile,
+        }
+    if profile not in ("accepted", "overfit"):
+        raise ValueError(f"unknown candidate profile: {profile}")
+    if "router_prompt" in targets:
+        candidate["router_prompt"] += "\nREFUND_ROUTE=STRICT\n"
+        if profile == "overfit":
+            candidate["router_prompt"] += "BILLING_TO_REFUND=ON\n"
+    if "skill_prompt" in targets:
+        candidate["skill_prompt"] += "\nREFUND_REASON=REQUIRED\n"
+    if "system_prompt" in targets:
+        candidate["system_prompt"] += "\nJSON_STATUS=ALWAYS_REQUIRED\n"
+    return {
+        "total_rounds": 1,
+        "rounds": [
+            {
+                "round": 1,
+                "profile": profile,
+                "targets": list(targets),
+                "candidate_prompts": candidate,
+                "cost": 0.0,
+                "tokens": 0,
+                "duration_seconds": 0.0,
+            }
+        ],
+        "best_prompts": candidate,
+        "cost": 0.0,
+        "tokens": 0,
+        "seed": 42,
+        "candidate_profile": profile,
+    }

examples/optimization/counterfactual_trace_loop/gate.json

@@ -0,0 +1 @@
+{"min_validation_delta":0.01,"max_cost":1.0,"max_latency_seconds":180,"max_counterfactual_evaluations_per_case":7,"protected_cases":["val_billing"]}

examples/optimization/counterfactual_trace_loop/optimizer.json

@@ -0,0 +1 @@
+{"evaluate":{"metrics":[{"metric_name":"tool_trajectory_avg_score","threshold":1.0},{"metric_name":"final_response_avg_score","threshold":1.0}],"num_runs":1},"optimize":{"algorithm":{"name":"gepa_reflective","seed":42,"reflection_lm":{"model_name":"${TRPC_AGENT_MODEL_NAME}","base_url":"${TRPC_AGENT_BASE_URL}","api_key":"${TRPC_AGENT_API_KEY}"},"max_metric_calls":20}}}

examples/optimization/counterfactual_trace_loop/pipeline/__init__.py

@@ -0,0 +1 @@
+"""Pipeline components for the counterfactual trace optimization loop."""

examples/optimization/counterfactual_trace_loop/pipeline/counterfactual.py

@@ -0,0 +1,70 @@
+"""Budgeted counterfactual evaluation using the official evaluator."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from trpc_agent_sdk.evaluation import EvalCase
+
+from .diagnosis import attribute_from_evidence
+from .interventions import InterventionKind, build_counterfactual
+from .models import CounterfactualEvidence, FailureAttribution
+from .probe import evaluate_trace_cases
+
+
+async def diagnose_trace_case(case: EvalCase, workspace: Path, max_evaluations: int = 7) -> FailureAttribution:
+    baseline = (await evaluate_trace_cases([case], workspace))[case.eval_id]
+    evidence = []
+    singles = [
+        InterventionKind.REPLACE_FINAL_RESPONSE,
+        InterventionKind.REPLACE_TOOL_NAME,
+        InterventionKind.REPLACE_TOOL_ARGUMENTS,
+        InterventionKind.NORMALIZE_FORMAT,
+    ]
+    combinations = [
+        InterventionKind.REPLACE_TOOL_NAME_AND_ARGUMENTS,
+        InterventionKind.REPLACE_TOOL_NAME_AND_FINAL_RESPONSE,
+        InterventionKind.REPLACE_TOOL_ARGUMENTS_AND_FINAL_RESPONSE,
+    ]
+    for kind in singles + combinations:
+        if len(evidence) >= max_evaluations:
+            break
+        built = build_counterfactual(case, kind)
+        if not built.valid or built.eval_case is None:
+            evidence.append(
+                CounterfactualEvidence(
+                    kind.value,
+                    False,
+                    built.status,
+                    False,
+                    [],
+                    [],
+                    baseline,
+                    {},
+                    structurally_valid=built.structurally_valid,
+                    semantically_coherent=built.semantically_coherent,
+                    coherence_warnings=list(built.coherence_warnings),
+                )
+            )
+            continue
+        after = (await evaluate_trace_cases([built.eval_case], workspace))[built.eval_case.eval_id]
+        repaired = sorted(k for k, v in baseline.items() if v < 1 and after.get(k, v) >= 1)
+        unchanged = sorted(k for k, v in baseline.items() if after.get(k) == v)
+        evidence.append(
+            CounterfactualEvidence(
+                kind.value,
+                True,
+                built.status,
+                all(v >= 1 for v in after.values()),
+                repaired,
+                unchanged,
+                baseline,
+                after,
+                structurally_valid=built.structurally_valid,
+                semantically_coherent=built.semantically_coherent,
+                coherence_warnings=list(built.coherence_warnings),
+            )
+        )
+        if evidence[-1].changed_fail_to_pass and kind in singles:
+            break
+    return attribute_from_evidence(case.eval_id, evidence, evaluations_used=len(evidence), budget=max_evaluations)