Initial EviScene release
c10636a 15 days ago
44.1 kB
from __future__ import annotations

import argparse
import asyncio
import hashlib
import json
import os
import statistics
import subprocess
from dataclasses import asdict
from collections import Counter
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, Iterable, List

from eviscene.workflow import (
    AGENT_DECISION_POLICY_VERSION,
    VerificationWorkflow,
    WorkflowConfig,
)
from eviscene.provider_profiles import PROFILE_IDS
from eviscene.storage import default_eval_root
from eviscene.eval.release_adapter import (
    image_only_case_from_runtime_row,
    load_runtime_release,
    resolve_runtime_image_path,
)
from eviscene.eval.gemini_run_guard import GeminiRunGuard
from eviscene.eval.result_records import classification_prediction
from eviscene.eval.run_artifacts import private_index
from eviscene.eval.scoring_release_adapter import (
    SCORING_PACKAGE_SCHEMA_VERSION,
    ScoringRuntimeRelease,
    adapt_scoring_gold_for_process,
    load_scoring_release,
    resolve_scoring_image_path,
)
from eviscene.orchestrator.runtime_case import verify_case_image
from eviscene.orchestrator.source_access import SourceAccessPolicy
from eviscene.redaction import sanitize_for_persistence
from eviscene.trajectory.exporter import (
    export_trajectory_sft_example,
    trajectory_policy_step_ids,
)
from eviscene.trajectory.perception_exporter import export_perception_example
from eviscene.trajectory.reference_chain import score_reference_chain_trace
from eviscene.trajectory.scoring import score_process_trace
from scripts.audit_real_trace import audit_trace


RUN_SCHEMA_VERSION = "ifv-eval-run-v2"
ROLLOUT_MEMBER_SCHEMA_VERSION = "ifv-rollout-group-member-v1"
ACTIVE_POLICY_STAGES = (
    "UNIFIED_REACT",
    "UNIFIED_JUDGMENT",
)


def _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Run EviScene benchmark evaluation."
    )
    parser.add_argument(
        "--benchmark",
        required=True,
        help=(
            "Path to runtime_input/cases.jsonl in a v0.3 benchmark or "
            "reviewed ifv-scoring-gold-v1 package."
        ),
    )
    parser.add_argument(
        "--profile",
        choices=PROFILE_IDS,
        default=None,
        help="Explicit teacher/student provider profile.",
    )
    parser.add_argument(
        "--provider",
        default=None,
        help="LLM provider for the agent.",
    )
    parser.add_argument(
        "--model",
        default=None,
        help="Model name for the agent.",
    )
    parser.add_argument(
        "--vlm-provider",
        default=None,
        help="Optional VLM provider override.",
    )
    parser.add_argument(
        "--vlm-model",
        default=None,
        help="Optional VLM model override.",
    )
    parser.add_argument(
        "--image-access-mode",
        choices=["direct_multimodal", "separate_vlm"],
        default="direct_multimodal",
        help=(
            "Whether the main LLM receives the image directly or only receives "
            "structured VLM observations."
        ),
    )
    parser.add_argument(
        "--agent-decision-policy-version",
        choices=["unified-react-v1"],
        default=AGENT_DECISION_POLICY_VERSION,
        help=(
            "Agent orchestration policy. The current runtime uses unified-react-v1."
        ),
    )
    parser.add_argument(
        "--llm-wire-api",
        default=None,
        choices=["interactions", "responses", "chat_completions"],
        help="Optional wire API override. Gemini accepts interactions only.",
    )
    parser.add_argument(
        "--vlm-wire-api",
        default=None,
        choices=["interactions", "responses", "chat_completions"],
        help="Optional VLM wire API override. Gemini production runs require interactions.",
    )
    parser.add_argument(
        "--output-dir",
        default=None,
        help="Directory for traces, predictions, and summary files.",
    )
    parser.add_argument(
        "--concurrency",
        type=int,
        default=1,
        help="Concurrent complete rollout episodes.",
    )
    parser.add_argument(
        "--rollouts-per-case",
        type=int,
        default=1,
        help="Independent complete episodes per case (recommended for RL: 4).",
    )
    parser.add_argument(
        "--base-sampling-seed",
        type=int,
        default=1729,
        help="Base seed for reproducible per-episode Qwen sampling seeds.",
    )
    parser.add_argument(
        "--timeout",
        type=float,
        default=1800.0,
        help="Per-image timeout budget in seconds.",
    )
    parser.add_argument(
        "--limit",
        type=int,
        default=None,
        help="Optional sample limit for debugging.",
    )
    parser.add_argument(
        "--case-id",
        action="append",
        default=None,
        help=(
            "Run only this case_id. Repeat to select an explicit ordered canary "
            "set without exposing labels to the Agent."
        ),
    )
    parser.add_argument(
        "--shard-count",
        type=int,
        default=1,
        help="Deterministically split sorted case IDs into this many shards.",
    )
    parser.add_argument(
        "--shard-index",
        type=int,
        default=0,
        help="Zero-based shard index selected after explicit case filtering.",
    )
    parser.add_argument(
        "--training-prohibited",
        action="store_true",
        help=(
            "Mark every derived trajectory as ineligible for training; required "
            "for frozen external evaluation releases."
        ),
    )
    parser.add_argument(
        "--source-access-policy",
        default=None,
        help=(
            "Evaluation-only JSON policy for blocking benchmark-origin sources. "
            "The policy is never added to model context or traces."
        ),
    )
    parser.add_argument(
        "--disable-web-search",
        action="store_true",
        help="Ablation: remove text_search from the formal ReAct runtime.",
    )
    parser.add_argument(
        "--disable-image-retrieval",
        action="store_true",
        help=(
            "Ablation: remove text_image_search and reverse_image_search."
        ),
    )
    parser.add_argument(
        "--disable-evidence-inspection",
        action="store_true",
        help=(
            "Ablation: remove visit and compare_with_reference."
        ),
    )
    return parser.parse_args()


def _load_benchmark(path: Path) -> List[Dict[str, Any]]:
    if path.suffix.lower() == ".jsonl":
        rows = []
        for line in path.read_text(encoding="utf-8").splitlines():
            if line.strip():
                rows.append(json.loads(line))
        return rows
    payload = json.loads(path.read_text(encoding="utf-8"))
    if isinstance(payload, list):
        return payload
    raise TypeError("Benchmark must be a JSON list or JSONL file.")


def _safe_mean(values: List[float]) -> float:
    if not values:
        return 0.0
    return float(statistics.mean(values))


def _positive_int(value: Any, *, name: str) -> int:
    try:
        parsed = int(value)
    except (TypeError, ValueError) as exc:
        raise ValueError(f"{name} must be an integer") from exc
    if parsed < 1:
        raise ValueError(f"{name} must be at least 1")
    return parsed


def _qwen_stage_thinking_config(
    *,
    provider: str,
    model_name: str,
) -> Dict[str, str]:
    """Record the actual Qwen thinking switch used by the runtime.
    Gemini stage settings are irrelevant for local Qwen serving. For the
    active unified stages, Qwen3.5 and Qwen3-VL both default to visible
    thinking in ``Orchestrator._stage_generation_config``; an explicit
    environment override remains authoritative.
    """

    active_provider = str(provider).strip().lower()
    local_qwen = active_provider in {"qwen_local", "lmdeploy"}
    result: Dict[str, str] = {}
    for stage in ACTIVE_POLICY_STAGES:
        raw = os.getenv(
            f"QWEN_{stage}_ENABLE_THINKING",
            "true" if local_qwen else "false",
        ).strip().lower()
        if raw not in {"true", "false"}:
            raise ValueError(
                f"QWEN_{stage}_ENABLE_THINKING must be true or false."
            )
        result[stage.lower()] = raw
    return result


def _now_iso() -> str:
    return datetime.now().astimezone().isoformat(timespec="seconds")


def _sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def _git_commit() -> str:
    configured = os.getenv("GIT_COMMIT", "").strip()
    try:
        completed = subprocess.run(
            ["git", "rev-parse", "HEAD"],
            cwd=Path(__file__).resolve().parents[2],
            check=True,
            capture_output=True,
            text=True,
            timeout=5,
        )
    except (OSError, subprocess.SubprocessError):
        return configured
    actual = completed.stdout.strip()
    if configured and actual and configured != actual:
        raise RuntimeError(
            "GIT_COMMIT does not match the runtime checkout: "
            f"configured={configured}, actual={actual}"
        )
    return actual or configured


def _write_json(path: Path, payload: Dict[str, Any]) -> None:
    serialized = json.dumps(
        sanitize_for_persistence(payload),
        ensure_ascii=False,
        indent=2,
    )
    _write_text(path, serialized + "\n")


def _write_jsonl(path: Path, rows: Iterable[Dict[str, Any]]) -> None:
    lines = [
        json.dumps(
            sanitize_for_persistence(row),
            ensure_ascii=False,
            separators=(",", ":"),
        )
        for row in rows
    ]
    _write_text(path, "\n".join(lines) + ("\n" if lines else ""))


def _write_text(path: Path, text: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    pending = path.with_name(f".{path.name}.tmp")
    pending.write_text(text, encoding="utf-8")
    pending.replace(path)


def _run_result_record(
    sample: Dict[str, Any],
    result: Dict[str, Any],
) -> Dict[str, Any]:
    state = result.get("state") or {}
    judgment = result.get("judgment")
    verdict_basis = result.get("verdict_basis")
    if verdict_basis is None and isinstance(judgment, dict):
        verdict_basis = judgment.get("verdict_basis")
    return {
        "case_id": sample.get("case_id"),
        "verdict": result.get("verdict"),
        "confidence": result.get("confidence"),
        "verdict_basis": verdict_basis,
        "termination": result.get("termination"),
        "time_taken": result.get("time_taken"),
        "total_tool_calls": result.get("total_tool_calls"),
        "llm_api_calls": result.get("llm_api_calls"),
        "token_usage": result.get("token_usage"),
        "error": result.get("error"),
        "stage_timings": state.get("stage_timings", {}),
        "trace_path": None,
    }


def _classification_prediction(
    sample: Dict[str, Any],
    result: Dict[str, Any],
) -> Dict[str, Any] | None:
    """Return a scorer row only for a completed factual judgment."""

    return classification_prediction(sample, result)


def _compute_summary(run_results: List[Dict[str, Any]]) -> Dict[str, Any]:
    total = len(run_results)
    verdict_counter: Counter[str] = Counter()
    times: List[float] = []
    tool_calls: List[float] = []
    llm_calls: List[float] = []
    errors = 0

    for row in run_results:
        pred = str(row.get("verdict", ""))
        verdict_counter[pred] += 1
        if (
            pred == "error"
            or str(row.get("termination") or "") == "error"
            or bool(str(row.get("error") or "").strip())
        ):
            errors += 1
        if isinstance(row.get("time_taken"), (int, float)):
            times.append(float(row["time_taken"]))
        if isinstance(row.get("total_tool_calls"), (int, float)):
            tool_calls.append(float(row["total_tool_calls"]))
        if isinstance(row.get("llm_api_calls"), (int, float)):
            llm_calls.append(float(row["llm_api_calls"]))

    return {
        "num_samples": total,
        "num_errors": errors,
        "predicted_verdict_distribution": dict(verdict_counter),
        "avg_time_taken_sec": round(_safe_mean(times), 3),
        "avg_tool_calls": round(_safe_mean(tool_calls), 3),
        "avg_llm_api_calls": round(_safe_mean(llm_calls), 3),
    }


def _private_index(
    rows: List[Dict[str, Any]],
    *,
    key: str,
    name: str,
) -> Dict[str, Dict[str, Any]]:
    return private_index(rows, key=key, name=name)


def _safe_trace_filename(identifier: str) -> str:
    safe_id = "".join(
        character if character.isalnum() or character in "-_."
        else "_"
        for character in str(identifier)
    )
    return f"{safe_id}.json"


def _select_samples(
    samples: List[Dict[str, Any]],
    *,
    requested_case_ids: List[str] | None,
    limit: int | None,
    shard_count: int = 1,
    shard_index: int = 0,
) -> List[Dict[str, Any]]:
    indexed: Dict[str, Dict[str, Any]] = {}
    for sample in samples:
        case_id = str(sample.get("case_id", "")).strip()
        if not case_id:
            raise ValueError("runtime input row lacks case_id")
        if case_id in indexed:
            raise ValueError(
                f"image-only release contains duplicate case_id {case_id}"
            )
        indexed[case_id] = sample
    if requested_case_ids:
        requested = [str(item).strip() for item in requested_case_ids]
        if any(not item for item in requested):
            raise ValueError("--case-id values must be non-empty")
        if len(requested) != len(set(requested)):
            raise ValueError("--case-id values must be unique")
        missing = [case_id for case_id in requested if case_id not in indexed]
        if missing:
            raise ValueError(
                "requested case_id values are absent from release: "
                + ", ".join(missing)
            )
        samples = [indexed[case_id] for case_id in requested]
    if shard_count < 1:
        raise ValueError("--shard-count must be at least 1")
    if shard_index < 0 or shard_index >= shard_count:
        raise ValueError("--shard-index must be in [0, shard-count)")
    if shard_count > 1:
        samples = [
            sample
            for index, sample in enumerate(
                sorted(samples, key=lambda item: str(item["case_id"]))
            )
            if index % shard_count == shard_index
        ]
    if limit is not None:
        samples = samples[:limit]
    return samples


def _file_descriptor(path: Path | None) -> Dict[str, Any] | None:
    if path is None or not path.is_file():
        return None
    return {"path": str(path.resolve()), "sha256": _sha256(path)}


def _rollout_specs(
    samples: List[Dict[str, Any]],
    runtime_cases: List[Any],
    *,
    rollouts_per_case: int,
    base_sampling_seed: int,
    policy_revision: str,
    model: str,
) -> List[Dict[str, Any]]:
    """Expand public cases into isolated, reproducible rollout episodes."""

    if rollouts_per_case < 1:
        raise ValueError("--rollouts-per-case must be at least 1")
    specs: List[Dict[str, Any]] = []
    for sample, runtime_case in zip(samples, runtime_cases):
        case_id = str(runtime_case.case_id)
        group_material = {
            "case_id": case_id,
            "policy_revision": policy_revision,
            "model": model,
            "base_sampling_seed": int(base_sampling_seed),
            "group_size": rollouts_per_case,
        }
        prompt_group_id = "pg-" + hashlib.sha256(
            json.dumps(
                group_material,
                ensure_ascii=False,
                sort_keys=True,
                separators=(",", ":"),
            ).encode("utf-8")
        ).hexdigest()[:24]
        for rollout_index in range(rollouts_per_case):
            seed_material = (
                f"{base_sampling_seed}:{case_id}:{rollout_index}"
            ).encode("utf-8")
            sampling_seed = int.from_bytes(
                hashlib.sha256(seed_material).digest()[:4], "big"
            ) & 0x7FFFFFFF
            episode_id = case_id
            if rollouts_per_case > 1:
                episode_id = (
                    f"{case_id[:150]}--{prompt_group_id[3:11]}"
                    f"--r{rollout_index:03d}"
                )
            specs.append(
                {
                    "sample": sample,
                    "runtime_case": runtime_case,
                    "case_id": case_id,
                    "prompt_group_id": prompt_group_id,
                    "episode_id": episode_id,
                    "rollout_index": rollout_index,
                    "group_size": rollouts_per_case,
                    "sampling_seed": sampling_seed,
                }
            )
    return specs


async def _run_eval(args: argparse.Namespace) -> Dict[str, Any]:
    benchmark_path = Path(args.benchmark).expanduser().resolve()
    manifest_probe = json.loads(
        (benchmark_path.parent.parent / "manifest.json").read_text(
            encoding="utf-8"
        )
    )
    is_scoring_release = (
        str(manifest_probe.get("schema_version") or "")
        == SCORING_PACKAGE_SCHEMA_VERSION
    )
    release = (
        load_scoring_release(benchmark_path)
        if is_scoring_release
        else load_runtime_release(benchmark_path)
    )
    samples = _load_benchmark(benchmark_path)
    samples = _select_samples(
        samples,
        requested_case_ids=getattr(args, "case_id", None),
        limit=args.limit,
        shard_count=int(getattr(args, "shard_count", 1)),
        shard_index=int(getattr(args, "shard_index", 0)),
    )
    samples = [
        (
            resolve_scoring_image_path(sample, release)
            if isinstance(release, ScoringRuntimeRelease)
            else resolve_runtime_image_path(sample, benchmark_path)
        )
        for sample in samples
    ]
    runtime_cases = [
        image_only_case_from_runtime_row(sample) for sample in samples
    ]
    case_ids = [case.case_id for case in runtime_cases]
    if len(case_ids) != len(set(case_ids)):
        raise ValueError("image-only release contains duplicate case_id values")
    for runtime_case in runtime_cases:
        verify_case_image(runtime_case, runtime_case.image_path)

    runtime_commit = _git_commit()
    rollouts_per_case = _positive_int(
        getattr(args, "rollouts_per_case", 1),
        name="rollouts per case",
    )
    base_sampling_seed = int(getattr(args, "base_sampling_seed", 1729))
    config = WorkflowConfig(
        profile_id=getattr(args, "profile", None),
        provider=getattr(args, "provider", None),
        model_name=getattr(args, "model", None),
        vlm_provider=getattr(args, "vlm_provider", None),
        vlm_model=getattr(args, "vlm_model", None),
        image_access_mode=getattr(
            args,
            "image_access_mode",
            "direct_multimodal",
        ),
        llm_wire_api=getattr(args, "llm_wire_api", None),
        vlm_wire_api=getattr(args, "vlm_wire_api", None),
        timeout=args.timeout,
        enable_web_search=not bool(
            getattr(args, "disable_web_search", False)
        ),
        enable_image_retrieval=not bool(
            getattr(args, "disable_image_retrieval", False)
        ),
        enable_evidence_inspection=not bool(
            getattr(args, "disable_evidence_inspection", False)
        ),
        save_traces=True,
        decision_policy_version=getattr(
            args,
            "agent_decision_policy_version",
            AGENT_DECISION_POLICY_VERSION,
        ),
    )
    rollout_specs = _rollout_specs(
        samples,
        runtime_cases,
        rollouts_per_case=rollouts_per_case,
        base_sampling_seed=base_sampling_seed,
        policy_revision=runtime_commit,
        model=str(config.model_name),
    )

    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    run_dir = (
        Path(args.output_dir)
        if args.output_dir
        else default_eval_root()
        / (
            f"{timestamp}_{config.provider}_"
            f"{str(config.model_name).replace('/', '_')}"
        )
    )
    if run_dir.exists() and any(run_dir.iterdir()):
        raise FileExistsError(
            f"Evaluation output directory must be new or empty: {run_dir}"
        )
    stage_thinking_levels = {
        stage.lower(): os.getenv(
            f"GEMINI_{stage}_THINKING_LEVEL",
            os.getenv(
                "GEMINI_AGENT_THINKING_LEVEL",
                "high" if stage.lower() == "unified_react" else "low",
            ),
        ).strip().lower()
        for stage in ACTIVE_POLICY_STAGES
    }
    qwen_stage_thinking = _qwen_stage_thinking_config(
        provider=str(config.provider),
        model_name=str(config.model_name),
    )
    explicit_policy = (
        SourceAccessPolicy.load(args.source_access_policy)
        if args.source_access_policy
        else SourceAccessPolicy.load(release.artifacts.source_access_policy)
        if release.artifacts.source_access_policy is not None
        else None
    )
    config.source_access_policy = explicit_policy

    trace_dir = run_dir / "traces"
    trace_dir.mkdir(parents=True, exist_ok=True)
    manifest_path = run_dir / "run_manifest.json"
    manifest: Dict[str, Any] = {
        "schema_version": RUN_SCHEMA_VERSION,
        "run_id": run_dir.name,
        "status": "running",
        "started_at": _now_iso(),
        "completed_at": None,
        "git_commit": runtime_commit,
        "benchmark": {
            "path": str(benchmark_path),
            "sha256": _sha256(benchmark_path),
            "sample_count": len(samples),
            "episode_count": len(rollout_specs),
            "limit": args.limit,
            "shard_count": int(getattr(args, "shard_count", 1)),
            "shard_index": int(getattr(args, "shard_index", 0)),
            "training_prohibited": bool(
                getattr(args, "training_prohibited", False)
            ),
            "runtime_release": True,
            "release_id": release.release_id,
            "release_stage": release.release_stage,
            "release_manifest": _file_descriptor(release.manifest_path),
            "runtime_contract_version": release.runtime_contract_version,
            "input_mode": release.input_mode,
            "decision_policy_version": release.decision_policy_version,
            "classification_protocol": (
                _file_descriptor(release.artifacts.classification_protocol)
                if release.artifacts.classification_protocol is not None
                else None
            ),
            "process_reference_protocol": (
                _file_descriptor(release.artifacts.process_reference_protocol)
                if release.artifacts.process_reference_protocol is not None
                else None
            ),
            "evaluation_gold": None,
        },
        "agent": {
            "profile_id": config.profile_id,
            "decision_policy_version": config.decision_policy_version,
            "provider": config.provider,
            "model": config.model_name,
            "vlm_provider": config.vlm_provider,
            "vlm_model": config.vlm_model,
            "image_access_mode": config.image_access_mode,
            "base_url": config.llm_base_url,
            "vlm_base_url": config.vlm_base_url,
            "llm_wire_api": config.llm_wire_api,
            "vlm_wire_api": config.vlm_wire_api,
            "timeout_seconds": args.timeout,
            "max_tool_actions": 24,
            "stage_thinking_levels": stage_thinking_levels,
            "qwen_stage_enable_thinking": qwen_stage_thinking,
            "concurrency": max(1, args.concurrency),
            "gemini_max_inflight_requests": (
                int(
                    os.getenv(
                        "GEMINI_MAX_INFLIGHT_REQUESTS",
                        str(max(1, args.concurrency)),
                    )
                )
                if str(config.provider).lower() == "gemini"
                else None
            ),
            "gemini_eval_max_concurrency": (
                int(
                    os.getenv(
                        "GEMINI_EVAL_MAX_CONCURRENCY",
                        str(max(1, args.concurrency)),
                    )
                )
                if str(config.provider).lower() == "gemini"
                else None
            ),
            "perception_cache_enabled": os.getenv(
                "PERCEPTION_CACHE_ENABLED",
                "1",
            ).strip().lower()
            in {"1", "true", "yes"},
            "web_cache_enabled": os.getenv(
                "TOOL_CACHE_ENABLED",
                "0",
            ).strip().lower()
            in {"1", "true", "yes"},
            "tool_cache_ttl_seconds": os.getenv(
                "TOOL_CACHE_TTL_SECONDS",
                "3600",
            ),
            "perception_cache_version": os.getenv(
                "PERCEPTION_CACHE_VERSION",
                "perception-v1",
            ),
            "ocr_cache_version": os.getenv(
                "OCR_CACHE_VERSION",
                f"{os.getenv('OCR_BACKEND', 'baidu').strip().lower()}-v1",
            ),
            "rollouts_per_case": rollouts_per_case,
            "base_sampling_seed": base_sampling_seed,
            "tool_families": config.react_tool_family_config.to_manifest(),
            "sampling_seed_derivation": (
                "sha256(base_seed:case_id:rollout_index)-31bit-v1"
            ),
        },
        "source_access_policy": {
            "active": bool(explicit_policy and explicit_policy.active),
            "policy_id": explicit_policy.policy_id if explicit_policy else None,
            "cache_partition": explicit_policy.cache_partition if explicit_policy else None,
        },
        "artifacts": {
            "predictions": "predictions.jsonl",
            "episode_predictions": "episode_predictions.jsonl",
            "run_results": "run_results.jsonl",
            "process_metrics": "process_metrics.jsonl",
            "reference_chain_metrics": "reference_chain_metrics.jsonl",
            "trajectory_scores": "trajectory_scores.jsonl",
            "trajectory_sft": "trajectory_sft.jsonl",
            "perception_trajectories": "perception_trajectories.jsonl",
            "rollout_groups": "rollout_groups.jsonl",
            "post_rollout_rewards": "post_rollout_rewards.jsonl",
            "summary": "summary.json",
            "traces": "traces/",
        },
    }
    gemini_guard = GeminiRunGuard.acquire(
        provider=config.provider,
        concurrency=max(1, args.concurrency),
        run_id=manifest["run_id"],
    )
    try:
        _write_json(manifest_path, manifest)
        config.output_dir = str(trace_dir)
        workflow = VerificationWorkflow(config)
        image_paths = [
            str(spec["sample"]["image_path"]) for spec in rollout_specs
        ]
        image_ids = [str(spec["episode_id"]) for spec in rollout_specs]
        results = await workflow.run_batch(
            image_paths=image_paths,
            image_ids=image_ids,
            runtime_cases=[spec["runtime_case"] for spec in rollout_specs],
            sampling_seeds=[spec["sampling_seed"] for spec in rollout_specs],
            concurrency=max(1, args.concurrency),
        )

        evaluation_gold_index = _private_index(
            _load_benchmark(release.artifacts.evaluation_gold),
            key="case_id",
            name="evaluation-gold",
        )
        missing_gold = [
            case_id
            for case_id in case_ids
            if case_id not in evaluation_gold_index
        ]
        if missing_gold:
            raise RuntimeError(
                "evaluation-gold rows missing runtime case IDs: "
                + ", ".join(missing_gold[:10])
            )
        manifest["benchmark"]["evaluation_gold"] = _file_descriptor(
            release.artifacts.evaluation_gold
        )

        predictions: List[Dict[str, Any]] = []
        episode_predictions: List[Dict[str, Any]] = []
        run_results: List[Dict[str, Any]] = []
        process_metrics: List[Dict[str, Any]] = []
        reference_chain_metrics: List[Dict[str, Any]] = []
        trajectory_scores: List[Dict[str, Any]] = []
        trajectory_sft: List[Dict[str, Any]] = []
        perception_trajectories: List[Dict[str, Any]] = []
        rollout_members: List[Dict[str, Any]] = []
        post_rollout_rewards: List[Dict[str, Any]] = []
        for spec, result in zip(rollout_specs, results):
            sample = spec["sample"]
            identity = str(spec["case_id"])
            episode_id = str(spec["episode_id"])
            record = _run_result_record(sample, result)
            record.update(
                {
                    "prompt_group_id": spec["prompt_group_id"],
                    "episode_id": episode_id,
                    "rollout_index": spec["rollout_index"],
                    "sampling_seed": spec["sampling_seed"],
                }
            )
            trace_path = trace_dir / _safe_trace_filename(episode_id)
            member = {
                "schema_version": ROLLOUT_MEMBER_SCHEMA_VERSION,
                "prompt_group_id": spec["prompt_group_id"],
                "case_id": identity,
                "episode_id": episode_id,
                "rollout_index": spec["rollout_index"],
                "group_size": spec["group_size"],
                "sampling_seed": spec["sampling_seed"],
                "policy_revision": manifest["git_commit"],
                "trace_path": None,
                "training_prohibited": (
                    bool(getattr(args, "training_prohibited", False))
                    or release.release_stage == "development_subset"
                ),
                "training_eligible": False,
                "sft_structured_eligibility_required": isinstance(
                    release, ScoringRuntimeRelease
                ),
            }
            if trace_path.exists():
                record["trace_path"] = trace_path.relative_to(run_dir).as_posix()
                member["trace_path"] = record["trace_path"]
                trace = json.loads(trace_path.read_text(encoding="utf-8"))
                trace_state = trace.get("state")
                if isinstance(trace_state, dict) and isinstance(
                    trace_state.get("perception"), dict
                ):
                    perception_trajectories.append(
                        export_perception_example(
                            trace,
                            source_metadata={
                                "source_run_id": manifest["run_id"],
                                "runtime_commit": manifest["git_commit"],
                                "release_id": release.release_id,
                                "runtime_contract_version": (
                                    release.runtime_contract_version
                                ),
                            },
                        ).model_dump(mode="json")
                    )
                private_gold = evaluation_gold_index[identity]
                process_gold = (
                    adapt_scoring_gold_for_process(private_gold)
                    if isinstance(release, ScoringRuntimeRelease)
                    else private_gold
                )
                process_metadata = {
                    "evaluation_gold": _file_descriptor(
                        release.artifacts.evaluation_gold
                    ),
                }
                if release.artifacts.process_reference_protocol is not None:
                    process_metadata["process_reference_protocol"] = (
                        _file_descriptor(
                            release.artifacts.process_reference_protocol
                        )
                    )
                metrics, teacher_score = score_process_trace(
                    trace,
                    process_gold,
                    score_metadata=process_metadata,
                )
                metrics["episode_id"] = episode_id
                metrics["prompt_group_id"] = spec["prompt_group_id"]
                teacher_score["episode_id"] = episode_id
                teacher_score["prompt_group_id"] = spec["prompt_group_id"]
                audit_report = audit_trace(
                    trace_path,
                    enforce_source_access_policy=bool(
                        explicit_policy and explicit_policy.active
                    ),
                    source_access_policy=explicit_policy,
                )
                strict_failures = audit_report.failures(strict_scheduler=True)
                hard_failures = audit_report.failures(strict_scheduler=False)
                strict_trace_audit_pass = not strict_failures
                hard_trace_audit_pass = not hard_failures
                metrics["strict_trace_audit_pass"] = strict_trace_audit_pass
                metrics["hard_trace_audit_pass"] = hard_trace_audit_pass
                metrics["strict_trace_audit_failures"] = [
                    asdict(item) for item in strict_failures
                ]
                process_metrics.append(metrics)
                if isinstance(release, ScoringRuntimeRelease):
                    reference_score = {
                        "schema_version": "ifv-scoring-reference-chain-v1",
                        "case_id": identity,
                        "applicable": False,
                        "reason": (
                            "ifv-scoring-gold-v1 uses a structured relation "
                            "target gate instead of the v0.3 reference protocol"
                        ),
                    }
                else:
                    reference_score = await score_reference_chain_trace(
                        trace,
                        private_gold,
                        score_metadata=process_metadata,
                    )
                reference_score["episode_id"] = episode_id
                reference_score["prompt_group_id"] = spec["prompt_group_id"]
                reference_chain_metrics.append(reference_score)
                trajectory_scores.append(teacher_score)
                try:
                    exported_trajectory = export_trajectory_sft_example(
                        trace,
                        source_metadata={
                            "source_run_id": manifest["run_id"],
                            "runtime_commit": manifest["git_commit"],
                            "release_id": release.release_id,
                            "runtime_contract_version": (
                                release.runtime_contract_version
                            ),
                            "process_reference_protocol_version": (
                                release.manifest.get(
                                    "process_reference_protocol_version",
                                    "",
                                )
                            ),
                        },
                    )
                except ValueError as exc:
                    metrics["training_eligible"] = False
                    reasons = list(
                        metrics.get("training_exclusion_reasons", []) or []
                    )
                    reasons.append(f"trajectory_sft_export_rejected: {exc}")
                    metrics["training_exclusion_reasons"] = list(
                        dict.fromkeys(reasons)
                    )
                    teacher_score["training_eligible"] = False
                    teacher_score["training_exclusion_reasons"] = list(
                        metrics["training_exclusion_reasons"]
                    )
                    exported_trajectory = None
                if exported_trajectory is not None:
                    trajectory_sft.append(
                        exported_trajectory.model_dump(mode="json")
                    )
                policy_step_ids = trajectory_policy_step_ids(
                    trace,
                    episode_id=episode_id,
                )
                member["sft_training_eligible"] = bool(
                    strict_trace_audit_pass
                    and bool(metrics.get("training_eligible", False))
                    and exported_trajectory is not None
                    and not member["training_prohibited"]
                    and not isinstance(release, ScoringRuntimeRelease)
                )
                member["rl_reward_eligible"] = bool(
                    not metrics.get("engineering_error", False)
                    and bool(policy_step_ids)
                    and not member["training_prohibited"]
                )
                member["training_eligible"] = member["sft_training_eligible"]
                post_rollout_rewards.append(
                    {
                        "schema_version": "ifv-post-rollout-deterministic-v1",
                        "prompt_group_id": spec["prompt_group_id"],
                        "case_id": identity,
                        "episode_id": episode_id,
                        "classification_correct": bool(
                            metrics.get("result_correct", False)
                        ),
                        "fatal_engineering_error": bool(
                            metrics.get("engineering_error", False)
                        ),
                        "strict_trace_audit_pass": bool(
                            strict_trace_audit_pass
                        ),
                        "hard_trace_audit_pass": bool(hard_trace_audit_pass),
                        "strict_trace_audit_failure_codes": [
                            str(item.get("code", ""))
                            for item in metrics.get("strict_trace_audit_failures", [])
                            if str(item.get("code", ""))
                        ],
                        "training_prohibited": member["training_prohibited"],
                        "rl_reward_eligible": member["rl_reward_eligible"],
                        "step_ids": policy_step_ids,
                        "process_components": dict(
                            teacher_score.get("components") or {}
                        ),
                    }
                )
            else:
                process_metrics.append(
                    {
                        "schema_version": "ifv-process-metrics-v2",
                        "case_id": identity,
                        "engineering_error": True,
                        "runtime_verdict": result.get("verdict"),
                        "result_correct": False,
                        "first_error": {
                            "error": "canonical trace is missing",
                        },
                    }
                )
                reference_chain_metrics.append(
                    {
                        "schema_version": "ifv-reference-chain-metrics-v2",
                        "case_id": identity,
                        "engineering_error": True,
                        "error": "canonical trace is missing",
                        "metrics": {
                            "fact_recovery_recall": 0.0,
                            "chain_recovery_recall": 0.0,
                            "evidence_recovery_recall": 0.0,
                            "basis_reference_precision": 0.0,
                        },
                    }
                )
                trajectory_scores.append(
                    {
                        "schema_version": "ifv-trajectory-score-v2",
                        "case_id": identity,
                        "components": {
                            "result_reward": 0.0,
                            "grounded_finding_reward": 0.0,
                            "gap_coverage_reward": 0.0,
                            "bridge_reward": 0.0,
                            "stop_calibration_reward": 0.0,
                            "duplicate_action_penalty": 0.0,
                            "invalid_task_penalty": 0.0,
                            "normalized_cost_penalty": 0.0,
                        },
                        "total": 0.0,
                    }
                )
                post_rollout_rewards.append(
                    {
                        "schema_version": "ifv-post-rollout-deterministic-v1",
                        "prompt_group_id": spec["prompt_group_id"],
                        "case_id": identity,
                        "episode_id": episode_id,
                        "classification_correct": False,
                        "fatal_engineering_error": True,
                        "strict_trace_audit_pass": False,
                        "training_prohibited": member["training_prohibited"],
                        "step_ids": [],
                        "process_components": {},
                    }
                )
            run_results.append(record)
            rollout_members.append(member)
            prediction = _classification_prediction(sample, result)
            if prediction is not None:
                episode_predictions.append(
                    {
                        **prediction,
                        "episode_id": episode_id,
                        "prompt_group_id": spec["prompt_group_id"],
                        "rollout_index": spec["rollout_index"],
                    }
                )
                # predictions.jsonl stays consumable by the data-owned
                # classification scorer: one row per public case. Multi-rollout
                # analysis belongs in episode_predictions.jsonl.
                if spec["rollout_index"] == 0:
                    predictions.append(prediction)

        summary = _compute_summary(run_results)
        summary["run_id"] = manifest["run_id"]
        _write_jsonl(run_dir / "predictions.jsonl", predictions)
        _write_jsonl(
            run_dir / "episode_predictions.jsonl",
            episode_predictions,
        )
        _write_jsonl(run_dir / "run_results.jsonl", run_results)
        _write_jsonl(run_dir / "process_metrics.jsonl", process_metrics)
        _write_jsonl(
            run_dir / "reference_chain_metrics.jsonl",
            reference_chain_metrics,
        )
        _write_jsonl(run_dir / "trajectory_scores.jsonl", trajectory_scores)
        _write_jsonl(run_dir / "trajectory_sft.jsonl", trajectory_sft)
        _write_jsonl(
            run_dir / "perception_trajectories.jsonl",
            perception_trajectories,
        )
        _write_jsonl(run_dir / "rollout_groups.jsonl", rollout_members)
        _write_jsonl(
            run_dir / "post_rollout_rewards.jsonl",
            post_rollout_rewards,
        )
        _write_json(run_dir / "summary.json", summary)

        manifest["status"] = (
            "completed_with_errors" if summary["num_errors"] else "completed"
        )
        manifest["completed_at"] = _now_iso()
        manifest["result"] = {
            "num_samples": len(samples),
            "num_episodes": summary["num_samples"],
            "num_errors": summary["num_errors"],
        }
        _write_json(manifest_path, manifest)
        return summary
    except BaseException as exc:
        manifest["status"] = "failed"
        manifest["completed_at"] = _now_iso()
        manifest["error"] = f"{type(exc).__name__}: {exc}"
        _write_json(manifest_path, manifest)
        raise
    finally:
        gemini_guard.release()


def main() -> None:
    args = _parse_args()
    summary = asyncio.run(_run_eval(args))
    print(json.dumps(summary, ensure_ascii=False, indent=2))
    if int(summary.get("num_errors", 0) or 0) > 0:
        raise SystemExit(1)


if __name__ == "__main__":
    main()