refactor(examples): define safe optimization contracts

Author: YAO-001

examples/optimization/eval_optimize_loop/eval_loop/config.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import math
 from dataclasses import dataclass
 from dataclasses import field
 from pathlib import Path
@@ -16,7 +17,7 @@ class GateConfig:
     allow_new_hard_fail: bool = False
     protected_case_ids: list[str] = field(default_factory=list)
     max_score_drop_per_case: float = 0.0
-    max_total_cost: float = 1.0
+    max_total_cost: float | None = 1.0
     extras: dict[str, Any] = field(default_factory=dict)
 
     def to_dict(self) -> dict[str, Any]:
@@ -125,9 +126,12 @@ def _parse_gate_config(payload: dict[str, Any], *, path: str) -> GateConfig:
     }
     extras = {key: value for key, value in payload.items() if key not in allowed}
 
-    min_val = payload.get("min_val_score_improvement", 0.01)
-    if not isinstance(min_val, (int, float)) or min_val < 0 or min_val > 1:
-        raise ValueError(f"{path}: field 'gate.min_val_score_improvement' must be a number between 0 and 1")
+    min_val = _finite_number(
+        payload.get("min_val_score_improvement", 0.01),
+        f"{path}: field 'gate.min_val_score_improvement'",
+        0.0,
+        1.0,
+    )
 
     allow_new_hard_fail = payload.get("allow_new_hard_fail", False)
     if not isinstance(allow_new_hard_fail, bool):
@@ -137,24 +141,56 @@ def _parse_gate_config(payload: dict[str, Any], *, path: str) -> GateConfig:
     if not isinstance(protected_case_ids, list) or not all(isinstance(item, str) for item in protected_case_ids):
         raise ValueError(f"{path}: field 'gate.protected_case_ids' must be a list of strings")
 
-    max_drop = payload.get("max_score_drop_per_case", 0.0)
-    if not isinstance(max_drop, (int, float)) or max_drop < 0:
-        raise ValueError(f"{path}: field 'gate.max_score_drop_per_case' must be a non-negative number")
+    max_drop = _finite_number(
+        payload.get("max_score_drop_per_case", 0.0),
+        f"{path}: field 'gate.max_score_drop_per_case'",
+        0.0,
+    )
 
-    max_total_cost = payload.get("max_total_cost", 1.0)
-    if not isinstance(max_total_cost, (int, float)) or max_total_cost < 0:
-        raise ValueError(f"{path}: field 'gate.max_total_cost' must be a non-negative number")
+    max_total_cost_value = payload.get("max_total_cost", 1.0)
+    max_total_cost = (
+        None
+        if max_total_cost_value is None
+        else _finite_number(
+            max_total_cost_value,
+            f"{path}: field 'gate.max_total_cost'",
+            0.0,
+        )
+    )
 
     return GateConfig(
-        min_val_score_improvement=float(min_val),
+        min_val_score_improvement=min_val,
         allow_new_hard_fail=allow_new_hard_fail,
         protected_case_ids=list(protected_case_ids),
-        max_score_drop_per_case=float(max_drop),
-        max_total_cost=float(max_total_cost),
+        max_score_drop_per_case=max_drop,
+        max_total_cost=max_total_cost,
         extras=extras,
     )
 
 
+def _finite_number(
+    value: Any,
+    field_name: str,
+    minimum: float,
+    maximum: float | None = None,
+) -> float:
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        raise ValueError(f"{field_name} must be a finite number")
+    try:
+        number = float(value)
+    except OverflowError as exc:
+        raise ValueError(f"{field_name} must be a finite number") from exc
+    if not math.isfinite(number):
+        raise ValueError(f"{field_name} must be a finite number")
+    if number < minimum:
+        raise ValueError(f"{field_name} must be a finite number greater than or equal to {minimum:g}")
+    if maximum is not None and number > maximum:
+        raise ValueError(
+            f"{field_name} must be a finite number between {minimum:g} and {maximum:g}"
+        )
+    return number
+
+
 def _validate_cases(cases: list[EvalCase], *, split: str, path: str | Path) -> None:
     seen: set[str] = set()
     for case in cases:

examples/optimization/eval_optimize_loop/eval_loop/loader.py

@@ -13,20 +13,33 @@
 
 def read_json(path: str | Path) -> dict[str, Any]:
     resolved = Path(path)
-    with resolved.open("r", encoding="utf-8") as file:
-        payload = json.load(file)
+    try:
+        with resolved.open("r", encoding="utf-8") as file:
+            payload = json.load(file, parse_constant=_reject_non_standard_json_constant)
+    except (json.JSONDecodeError, ValueError) as exc:
+        raise ValueError(f"{resolved}: invalid JSON: {exc}") from exc
     if not isinstance(payload, dict):
         raise ValueError(f"expected JSON object in {resolved}")
     return payload
 
 
+def _reject_non_standard_json_constant(constant: str) -> None:
+    raise ValueError(f"non-standard JSON constant {constant!r}")
+
+
 def load_eval_cases(path: str | Path, split: str | None = None) -> list[EvalCase]:
     payload = read_json(path)
     cases = payload.get("cases")
-    if not isinstance(cases, list):
-        raise ValueError(f"evalset {path} must contain a cases list")
     effective_split = split or payload.get("split") or Path(path).name.split(".", 1)[0]
-    return [EvalCase.from_dict(case, str(effective_split)) for case in cases]
+    if isinstance(cases, list):
+        return [EvalCase.from_dict(case, str(effective_split)) for case in cases]
+
+    sdk_cases = payload.get("eval_cases") or payload.get("evalCases")
+    if isinstance(sdk_cases, list):
+        _validate_sdk_evalset(payload, path)
+        return [_eval_case_from_sdk_dict(case, str(effective_split)) for case in sdk_cases]
+
+    raise ValueError(f"evalset {path} must contain a cases or evalCases list")
 
 
 def load_optimizer_config(path: str | Path) -> OptimizerConfig:
@@ -53,3 +66,84 @@ def sha256_file(path: str | Path) -> str:
         for chunk in iter(lambda: file.read(1024 * 1024), b""):
             digest.update(chunk)
     return digest.hexdigest()
+
+
+def _validate_sdk_evalset(payload: dict[str, Any], path: str | Path) -> None:
+    try:
+        from trpc_agent_sdk.evaluation._eval_set import EvalSet
+
+        EvalSet.model_validate(payload)
+    except Exception as exc:
+        raise ValueError(f"evalset {path} is not a valid SDK EvalSet: {exc}") from exc
+
+
+def _eval_case_from_sdk_dict(payload: dict[str, Any], split: str) -> EvalCase:
+    case_id = payload.get("eval_id") or payload.get("evalId")
+    if not case_id:
+        raise ValueError(f"SDK eval case is missing evalId/eval_id: {payload!r}")
+
+    session_input = payload.get("session_input") or payload.get("sessionInput") or {}
+    state = session_input.get("state") if isinstance(session_input, dict) else {}
+    state = state if isinstance(state, dict) else {}
+    expectation = state.get("eval_optimize_expectation")
+    if not isinstance(expectation, dict):
+        expectation = _infer_expectation_from_sdk_case(payload)
+
+    return EvalCase(
+        case_id=str(case_id),
+        split=split,
+        input=_first_user_text(payload),
+        expectation=dict(expectation),
+        tags=[str(item) for item in state.get("eval_optimize_tags", [])],
+        protected=bool(state.get("eval_optimize_protected", False)),
+        simulated_outputs=dict(state.get("eval_optimize_simulated_outputs") or expectation.get("simulated_outputs") or {}),
+        expected_failure_category=state.get("eval_optimize_expected_failure_category")
+        or expectation.get("expected_failure_category"),
+    )
+
+
+def _infer_expectation_from_sdk_case(payload: dict[str, Any]) -> dict[str, Any]:
+    expected = _first_final_response_text(payload)
+    if expected:
+        return {
+            "type": "exact",
+            "expected": expected,
+            "expected_failure_category": "final_response_mismatch",
+        }
+    raise ValueError(
+        "SDK eval case must put fake-mode metadata in sessionInput.state.eval_optimize_expectation "
+        f"or provide a finalResponse that can be treated as an exact expectation: {payload!r}"
+    )
+
+
+def _first_user_text(payload: dict[str, Any]) -> str:
+    for invocation in _conversation(payload):
+        content = invocation.get("user_content") or invocation.get("userContent") or {}
+        text = _content_text(content)
+        if text:
+            return text
+    return ""
+
+
+def _first_final_response_text(payload: dict[str, Any]) -> str:
+    for invocation in _conversation(payload):
+        content = invocation.get("final_response") or invocation.get("finalResponse") or {}
+        text = _content_text(content)
+        if text:
+            return text
+    return ""
+
+
+def _conversation(payload: dict[str, Any]) -> list[dict[str, Any]]:
+    conversation = payload.get("conversation") or []
+    return [item for item in conversation if isinstance(item, dict)]
+
+
+def _content_text(content: Any) -> str:
+    if not isinstance(content, dict):
+        return ""
+    texts = []
+    for part in content.get("parts") or []:
+        if isinstance(part, dict) and part.get("text") is not None:
+            texts.append(str(part["text"]))
+    return "\n".join(texts)

examples/optimization/eval_optimize_loop/eval_loop/schemas.py

@@ -7,6 +7,7 @@
 from dataclasses import field
 from dataclasses import is_dataclass
 from typing import Any
+from typing import Literal
 
 
 @dataclass(frozen=True)
@@ -27,12 +28,16 @@ def from_dict(cls, payload: dict[str, Any], split: str) -> "EvalCase":
         case_id = payload.get("case_id") or payload.get("id")
         if not case_id:
             raise ValueError(f"eval case is missing id/case_id: {payload!r}")
+        if "split" in payload and str(payload["split"]) != str(split):
+            raise ValueError(
+                f"eval case {case_id!r} split mismatch: payload has {payload['split']!r}, expected {split!r}"
+            )
         expectation = payload.get("expectation")
         if not isinstance(expectation, dict):
             raise ValueError(f"eval case {case_id!r} is missing expectation object")
         return cls(
             case_id=str(case_id),
-            split=str(payload.get("split") or split),
+            split=str(split),
             input=str(payload.get("input") or payload.get("user_input") or ""),
             expectation=dict(expectation),
             tags=list(payload.get("tags") or []),
@@ -52,7 +57,9 @@ class CaseResult:
     score: float
     passed: bool
     output: str
+    metrics: dict[str, float] = field(default_factory=dict)
     trace: dict[str, Any] = field(default_factory=dict)
+    trace_available: bool = False
     failure_category: str | None = None
     failure_reason: str | None = None
     evidence: str | None = None
@@ -84,6 +91,67 @@ class CandidatePrompt:
     prompt: str
     rationale: str
     prompt_diff: str
+    prompt_fields: dict[str, str] = field(default_factory=dict)
+
+    def bundle(self) -> dict[str, str]:
+        """Return this candidate's complete prompt bundle."""
+
+        if self.prompt_fields:
+            return dict(self.prompt_fields)
+        return {"system_prompt": self.prompt}
+
+
+@dataclass(frozen=True)
+class CostSummary:
+    """Cost attribution for an optimization run."""
+
+    optimizer: float = 0.0
+    evaluator: float = 0.0
+    agent: float = 0.0
+    total: float = 0.0
+    complete: bool = True
+
+
+@dataclass(frozen=True)
+class OptimizationRound:
+    """One auditable optimizer round."""
+
+    round_id: int
+    candidate_id: str
+    prompts: dict[str, str]
+    rationale: str
+    metrics: dict[str, float]
+    cost: CostSummary
+    duration_seconds: float
+
+
+WritebackStatus = Literal[
+    "rejected",
+    "not_requested",
+    "applied",
+    "rolled_back",
+    "rollback_failed",
+]
+
+
+@dataclass(frozen=True)
+class WritebackResult:
+    """Outcome of an optional source prompt writeback."""
+
+    status: WritebackStatus
+    before_hashes: dict[str, str] = field(default_factory=dict)
+    after_hashes: dict[str, str] = field(default_factory=dict)
+    error: str | None = None
+
+
+@dataclass(frozen=True)
+class OptimizationResult:
+    """Backend-neutral optimization output."""
+
+    candidates: list[CandidatePrompt]
+    rounds: list[OptimizationRound]
+    cost: CostSummary
+    raw_summary: dict[str, Any] = field(default_factory=dict)
 
 
 @dataclass(frozen=True)
@@ -141,6 +209,11 @@ class OptimizationReport:
     gate_decisions: list[GateDecision]
     selected_candidate: str | None
     audit: dict[str, Any]
+    rounds: list[OptimizationRound] = field(default_factory=list)
+    cost_summary: CostSummary = field(default_factory=CostSummary)
+    writeback: WritebackResult = field(
+        default_factory=lambda: WritebackResult(status="not_requested")
+    )
 
 
 def to_jsonable(value: Any) -> Any: