run_evaluation.py

#!/usr/bin/env python3
"""Evaluation runner for OmniNPC.

Usage:
    python run_evaluation.py --trials audio --n 50
    python run_evaluation.py --trials visual --n 50
    python run_evaluation.py --trials hierarchy --n 20
    python run_evaluation.py --trials realplay --session <session_id> --n 50
    python run_evaluation.py --trials all
"""

from __future__ import annotations

import argparse
import asyncio
import logging
from datetime import datetime
from pathlib import Path

from dotenv import load_dotenv

load_dotenv()
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)


def _detect_latest_eval_session(eval_frames_dir: Path = Path("data/eval_frames")) -> str:
    if not eval_frames_dir.exists():
        raise ValueError(f"eval frames directory not found: {eval_frames_dir}")

    session_dirs = [p for p in eval_frames_dir.iterdir() if p.is_dir()]
    if not session_dirs:
        raise ValueError(f"no session folders found in {eval_frames_dir}")

    latest = max(session_dirs, key=lambda p: p.stat().st_mtime)
    return latest.name


async def run_eval(args) -> None:
    from config import load_config
    config = load_config(args.config)

    ablation = args.ablation
    environment = args.environment
    # grid_off/model_swap: still genuinely unwireable -- neither has ANY trial
    # type that calls Super live (hierarchy uses hardcoded scripted directive
    # strings, see evaluation/trial_runner.py lines 702-710). This is a
    # protocol gap, not something Nano-side wiring (like blackboard's) can fix.
    # See Context/memory-2.md, 2026-07-17 "architectural fact" entry.
    _UNWIREABLE = {
        "grid_off": "no trial type calls SuperAgent.get_directive() live -- hierarchy uses hardcoded scripted directives",
        "model_swap": "TrialRunner never calls the Super model at all -- same protocol gap as grid_off",
    }
    if ablation in _UNWIREABLE:
        logger.warning(
            "Ablation '%s' requested, but no TrialRunner trial type can exercise it "
            "yet (%s) -- see Context/memory-2.md, 2026-07-17 entry. Refusing to run "
            "a batch that would silently produce misleading (unchanged) results.",
            ablation, _UNWIREABLE[ablation],
        )
        return
    # env2's trial-method frame-generation wiring is done for "visual",
    # "hierarchy", and "role" (as of 2026-07-23 -- see Context/memory-3.md and
    # docs/second-environment/open-decisions.md #1). "audio" and "adversarial"
    # remain unwired, and "all" still refuses since it would silently run a
    # partial batch mixing wired and unwired trial types under one label.
    # Same discipline as the _UNWIREABLE guard above.
    _ENV2_WIRED_TRIALS = {"visual", "hierarchy", "role"}
    if environment == "env2" and args.trials not in _ENV2_WIRED_TRIALS:
        logger.warning(
            "environment='env2' requested with --trials=%s, but only %s "
            "have env2 frame-generation wired in so far (audio/adversarial/"
            "all are not yet wired). Refusing to run a batch that would "
            "silently produce env1 data, or a misleading partial batch, "
            "under an env2 label. See Context/progress-tracker-3.md Stage 2.",
            args.trials, sorted(_ENV2_WIRED_TRIALS),
        )
        return
    if ablation == "memory_off":
        config.orchestrator.memory_enabled = False
    if ablation == "blackboard_off":
        config.orchestrator.blackboard_enabled = False

    from evaluation.trial_runner import TrialRunner
    runner = TrialRunner(config, environment=environment)

    audio_r: dict = {}
    visual_r: dict = {}
    hierarchy_r: dict = {}
    adversarial_r: dict = {}
    role_r: dict = {}
    # realplay removed

    if args.trials in ("audio", "all"):
        audio_r = await runner.run_audio_trials(n=args.n)
        logger.info("Audio results: %s", audio_r)

    if args.trials in ("visual", "all"):
        visual_r = await runner.run_visual_trials(n=args.n)
        logger.info("Visual results: %s", visual_r)

    if args.trials in ("hierarchy", "all"):
        hierarchy_r = await runner.run_hierarchy_comparison(n=min(args.n, 35))
        logger.info("Hierarchy results: %s", hierarchy_r)

    if args.trials in ("adversarial", "all"):
        adversarial_r = await runner.run_adversarial_trials(n=min(args.n, 35))
        logger.info("Adversarial results: %s", adversarial_r)

    if args.trials in ("role", "all"):
        role_r = await runner.run_role_distinctiveness_trials(n=min(args.n, 35))
        logger.info("Role results: %s", role_r)

    # realplay trials removed

    # Build combined results in the format generate_paper_table() expects
    vtrt_source = visual_r
    acrl = audio_r.get("acrl_mean_ms", 0)
    total_frames = sum(r.get("total_frames", 0) for r in (audio_r, visual_r, hierarchy_r, adversarial_r))
    inferred_frames = sum(r.get("inferred_frames", 0) for r in (audio_r, visual_r, hierarchy_r, adversarial_r))
    skip_rate = 1 - (inferred_frames / total_frames) if total_frames > 0 else 0.0
    combined = {
        "timestamp": datetime.now().isoformat(),
        # ACRL
        "acrl_mean_ms": acrl,
        "acrl_std_ms": audio_r.get("acrl_std_ms", 0),
        "acrl_p95_ms": audio_r.get("acrl_p95_ms", 0),
        # VTRT
        "vtrt_mean_ms": vtrt_source.get("vtrt_mean_ms", 0),
        "vtrt_std_ms": vtrt_source.get("vtrt_std_ms", 0),
        "vtrt_p95_ms": vtrt_source.get("vtrt_p95_ms", 0),
        "vtrt_evasion_rate": vtrt_source.get("pass_rate", 0),
        "vtrt_evasion_ci_lo": vtrt_source.get("vtrt_evasion_ci_lo"),
        "vtrt_evasion_ci_hi": vtrt_source.get("vtrt_evasion_ci_hi"),
        "vtrt_nodirective_evasion_rate": vtrt_source.get("vtrt_nodirective_evasion_rate", 0),
        # CMSS
        "cmss": audio_r.get("pass_rate", 0),
        "cmss_ci_lo": audio_r.get("cmss_ci_lo"),
        "cmss_ci_hi": audio_r.get("cmss_ci_hi"),
        # Directive adherence
        "survive_directive_adherence": hierarchy_r.get("survive_directive_adherence_pct", 0),
        "survive_ci_lo": hierarchy_r.get("survive_ci_lo"),
        "survive_ci_hi": hierarchy_r.get("survive_ci_hi"),
        "collect_directive_adherence": hierarchy_r.get("collect_directive_adherence_pct", 0),
        "collect_ci_lo": hierarchy_r.get("collect_ci_lo"),
        "collect_ci_hi": hierarchy_r.get("collect_ci_hi"),
        "vague_baseline_rate": round(
            (hierarchy_r.get("vague_baseline_evasive_pct", 0) +
             hierarchy_r.get("vague_baseline_collect_pct", 0)) / 2, 1
        ),
        "directive_lift_pp": round(
            (hierarchy_r.get("directive_lift_survive_pp", 0) +
             hierarchy_r.get("directive_lift_collect_pp", 0)) / 2, 1
        ),
        # Legacy fields
        "nano_completion": hierarchy_r.get("without_super_rate", 0),
        "super_completion": hierarchy_r.get("with_super_rate", 0),
        "fps": round(1000.0 / acrl, 2) if acrl else 0,
        "n_audio": audio_r.get("n", 0),
        "n_visual": vtrt_source.get("n", 0),
        "n_hierarchy": hierarchy_r.get("n", 0),
        "audio_detail": audio_r,
        "visual_detail": vtrt_source,
        "hierarchy_detail": hierarchy_r,
        "adversarial_detail": adversarial_r,
        "role_detail": role_r,
        # realplay_detail removed
        "inference_efficiency": {
            "skip_rate": skip_rate,
            "total_frames": total_frames,
            "inferred_frames": inferred_frames,
        },
    }

    if args.output_dir:
        out_dir = Path(args.output_dir)
    elif environment == "env2":
        out_dir = (
            Path("data/results/env2") if ablation == "none"
            else Path("data/results/env2/ablations") / ablation
        )
    elif ablation == "none":
        out_dir = Path("data/results")
    else:
        out_dir = Path("data/results/ablations") / ablation
    out_dir.mkdir(parents=True, exist_ok=True)
    out_path = out_dir / "trial_results.json"

    from evaluation.result_io import merge_and_write_results
    merge_and_write_results(out_path, combined, args.trials)
    logger.info("Trial results saved to %s", out_path)
    logger.info("Evaluation complete. Run `python run_analysis.py` to generate paper table.")


def main() -> None:
    parser = argparse.ArgumentParser(description="OmniNPC Evaluation Runner")
    parser.add_argument("--trials", choices=["audio", "visual", "hierarchy", "adversarial", "role", "all"], required=True)
    parser.add_argument("--n", type=int, default=50)
    parser.add_argument("--config", type=str, default="config/config.yaml")
    parser.add_argument(
        "--environment",
        choices=["env1", "env2"],
        default="env1",
        help=(
            "Which environment to run trials in. env1 is the existing arena "
            "(default, completely unaffected by this flag). env2 is the "
            "occluded synthetic scenario (Context/architecture-3.md) -- "
            "currently plumbing-only: refuses to run rather than silently "
            "reuse env1 frames, since no trial method has env2 frame "
            "generation wired in yet."
        ),
    )
    parser.add_argument(
        "--ablation",
        choices=["none", "memory_off", "blackboard_off", "grid_off", "model_swap"],
        default="none",
        help=(
            "Run under an ablation condition. memory_off and blackboard_off have "
            "real effects (both only touch Nano's prompt, which every trial type "
            "calls live); grid_off/model_swap are accepted but refuse to run with "
            "a warning, since no trial type calls Super live -- see "
            "Context/memory-2.md, 2026-07-17 entry."
        ),
    )
    parser.add_argument(
        "--output-dir",
        type=str,
        default=None,
        help=(
            "Override the output directory, taking priority over the --ablation-based "
            "default. Use this for re-runs of the full/baseline system at a new trial "
            "count so results don't land in the protected data/results/trial_results.json "
            "(e.g. --ablation none --output-dir data/results/stage2_full)."
        ),
    )
    args = parser.parse_args()
    asyncio.run(run_eval(args))


if __name__ == "__main__":
    main()