Spaces:

aadi-joshi
/

omninpc

Sleeping

hf: initial deploy -- no binaries, github raw urls for images

ed18158 6 months ago

23.4 kB

"""Trial Runner - systematic evaluation trials for the paper.
Three protocols:
  1. Audio Trials (N=50): measure ACRL and direction match (CMSS)
  2. Visual Threat Trials (N=50): measure VTRT
  3. Super+Nano Hierarchy Comparison (N=20): task completion with/without Super
"""

from __future__ import annotations

import asyncio
import base64
import io
import json
import logging
import math
import os
import random
import re
import time
from dataclasses import dataclass, field
from pathlib import Path

logger = logging.getLogger(__name__)


def _wilson_ci(k: int, n: int, z: float = 1.96) -> tuple[float, float]:
    """Return (lo_pct, hi_pct) 95% Wilson score interval for k successes in n trials."""
    if n == 0:
        return 0.0, 100.0
    p = k / n
    denom = 1 + z * z / n
    centre = (p + z * z / (2 * n)) / denom
    half = z * math.sqrt(p * (1 - p) / n + z * z / (4 * n * n)) / denom
    return round(max(0.0, centre - half) * 100, 1), round(min(100.0, centre + half) * 100, 1)

VOICE_COMMANDS = [
    ("go left", "left"),
    ("go right", "right"),
    ("move forward", "forward"),
    ("move backward", "backward"),
    ("turn left", "left"),
    ("turn right", "right"),
    ("run forward", "forward"),
    ("back up", "backward"),
    ("stop", "none"),
    ("retreat", "backward"),
]

_EVASIVE_ACTIONS = {
    "evade_threat", "move_backward", "move_left", "move_right",
    "sprint", "jump", "stop",
}

# Maps expected voice-command direction to action types that satisfy it.
# Used for real CMSS calculation (direction alignment, not just "not idle").
_DIRECTION_ACTIONS: dict[str, set[str]] = {
    "left":     {"move_left", "rotate_left"},
    "right":    {"move_right", "rotate_right"},
    "forward":  {"move_forward", "navigate_to_objective", "sprint", "interact", "pick_up"},
    "backward": {"move_backward", "evade_threat", "stop"},
}


@dataclass
class TrialResult:
    trial_id: int
    latency_ms: float
    success: bool
    details: dict = field(default_factory=dict)


def _make_scene_frame(threat_side: str = None) -> str:
    """Create a minimal game-like test JPEG using PIL."""
    from PIL import Image, ImageDraw
    img = Image.new("RGB", (128, 128), color=(30, 80, 40))
    draw = ImageDraw.Draw(img)
    draw.rectangle([0, 0, 128, 50], fill=(70, 130, 180))   # sky
    draw.rectangle([0, 50, 128, 128], fill=(30, 80, 40))   # ground
    if threat_side == "left":
        draw.rectangle([5, 25, 32, 68], fill=(200, 50, 50))
    elif threat_side == "right":
        draw.rectangle([96, 25, 123, 68], fill=(200, 50, 50))
    elif threat_side == "center":
        draw.rectangle([54, 25, 74, 68], fill=(200, 50, 50))
    draw.ellipse([58, 52, 70, 64], fill=(255, 215, 0))      # gold coin ahead
    buf = io.BytesIO()
    img.save(buf, format="JPEG", quality=60)
    return base64.b64encode(buf.getvalue()).decode()


def _make_adversarial_frame(frame_type: str) -> str:
    """Create degenerate/adversarial test frames for robustness evaluation."""
    from PIL import Image
    w, h = 128, 128
    if frame_type == "black":
        img = Image.new("RGB", (w, h), (0, 0, 0))
    elif frame_type == "white":
        img = Image.new("RGB", (w, h), (255, 255, 255))
    elif frame_type == "overexposed":
        img = Image.new("RGB", (w, h), (245, 245, 245))
    elif frame_type == "noise":
        try:
            import numpy as np
            arr = np.random.randint(0, 256, (h, w, 3), dtype=np.uint8)
            img = Image.fromarray(arr, "RGB")
        except ImportError:
            import random as _r
            img = Image.frombytes("RGB", (w, h), bytes(_r.randint(0, 255) for _ in range(w * h * 3)))
    else:  # "normal" — standard scene frame
        return _make_scene_frame()
    buf = io.BytesIO()
    img.save(buf, format="JPEG", quality=60)
    return base64.b64encode(buf.getvalue()).decode()


def _extract_action_dir(raw: str) -> tuple[str, str]:
    """Return (action, direction) from raw model response."""
    cleaned = re.sub(r"<think>.*?</think>", "", raw, flags=re.DOTALL).strip()
    cleaned = re.sub(r"```[a-z]*", "", cleaned).replace("```", "").strip()
    match = re.search(r"\{.*\}", cleaned, re.DOTALL)
    if match:
        try:
            data = json.loads(match.group(0))
            return str(data.get("action", "idle")).lower(), str(data.get("direction", "none")).lower()
        except json.JSONDecodeError:
            pass
    return "idle", "none"


class TrialRunner:
    """Runs evaluation trials against the live OmniNPC pipeline."""

    def __init__(self, config, orchestrator=None) -> None:
        self._config = config
        self._orchestrator = orchestrator
        self._audio_cues_dir = Path("data/prompts/audio_cues")
        self._audio_cues_dir.mkdir(parents=True, exist_ok=True)
        self._nim = None

    def _get_nim(self):
        if self._nim is None:
            from inference.nim_client import NIMClient
            api_key = os.getenv("NVIDIA_NIM_API_KEY", "")
            self._nim = NIMClient(api_key=api_key, base_url=self._config.nvidia.base_url)
        return self._nim

    async def _call_nano(self, frame_b64: str, audio_text: str, directive: str) -> tuple[str, float]:
        """Call Nano VL with a test frame. Returns (raw_response, latency_ms)."""
        from inference.prompt_templates import build_nano_system_prompt, build_nano_user_message
        nim = self._get_nim()
        system_prompt = build_nano_system_prompt(directive)
        audio_ctx = (
            f"Voice command heard: '{audio_text}'. Respond accordingly."
            if audio_text else "No voice command. React to visual input only."
        )
        user_text = build_nano_user_message(audio_ctx)
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": [
                {"type": "text", "text": user_text},
                {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{frame_b64}"}},
            ]},
        ]
        t0 = time.perf_counter()
        raw = await nim.complete(
            model=self._config.nvidia.nano_omni_model,
            messages=messages,
            max_tokens=256,
            temperature=0.1,
        )
        latency_ms = (time.perf_counter() - t0) * 1000
        return raw or "", latency_ms

    async def run_audio_trials(self, n: int = 50) -> dict:
        """Send voice-command prompts to Nano, measure ACRL and direction match (CMSS)."""
        logger.info("Starting %d audio trials", n)
        results: list[TrialResult] = []
        neutral_frame = _make_scene_frame()

        for i in range(n):
            cmd_text, expected_dir = random.choice(VOICE_COMMANDS)
            raw, latency_ms = await self._call_nano(
                neutral_frame, cmd_text,
                "Compete to collect coins. Follow any voice commands immediately.",
            )
            if not raw:
                logger.warning("Audio trial %d: empty API response, skipping", i)
                continue
            action, direction = _extract_action_dir(raw)
            # CMSS: direction-aligned success.
            # "none" (stop) → any response valid.
            # Otherwise: action type must be in the expected direction set,
            # OR the model's direction field must match the expected direction.
            if expected_dir == "none":
                success = True
            else:
                expected_acts = _DIRECTION_ACTIONS.get(expected_dir, set())
                success = (action in expected_acts) or (direction == expected_dir)
            results.append(TrialResult(
                trial_id=i,
                latency_ms=latency_ms,
                success=success,
                details={"command": cmd_text, "expected_dir": expected_dir,
                         "action": action, "direction": direction},
            ))
            logger.info("Audio %d/%d: '%s' -> action=%s dir=%s lat=%.0fms pass=%s",
                        i + 1, n, cmd_text, action, direction, latency_ms, success)
            await asyncio.sleep(0.05)

        if not results:
            return {"error": "all audio trials failed (API errors)"}

        latencies = [r.latency_ms for r in results]
        passed = sum(1 for r in results if r.success)
        n_actual = len(results)
        sorted_l = sorted(latencies)
        ci_lo, ci_hi = _wilson_ci(passed, n_actual)
        mean_lat = sum(latencies) / n_actual
        std_lat = math.sqrt(sum((l - mean_lat) ** 2 for l in latencies) / n_actual)
        return {
            "n": n_actual,
            "passed": passed,
            "failed": n_actual - passed,
            "pass_rate": round(passed / n_actual * 100, 1),
            "cmss_ci_lo": ci_lo,
            "cmss_ci_hi": ci_hi,
            "acrl_mean_ms": round(mean_lat, 1),
            "acrl_std_ms": round(std_lat, 1),
            "acrl_min_ms": round(sorted_l[0], 1),
            "acrl_max_ms": round(sorted_l[-1], 1),
            "acrl_p95_ms": round(sorted_l[int(n_actual * 0.95)], 1),
        }

    async def run_visual_trials(self, n: int = 50) -> dict:
        """Present threat frames to Nano, measure VTRT and evasion rate.
        Two conditions (n//2 each):
          - With directive: "EVADE IMMEDIATELY" (mirrors what Super issues in real system)
          - No directive:   vague "React to the scene" (baseline behavior)
        VTRT measures latency from frame presentation to evasive action.
        Evasion rate differentiates directive vs no-directive to validate hierarchy necessity.
        """
        logger.info("Starting %d visual trials (two conditions: directive / no-directive)", n)
        n_each = n // 2
        threat_sides = ["left", "right", "center"]

        directive_results: list[TrialResult] = []
        nodirective_results: list[TrialResult] = []

        for i in range(n_each):
            threat_side = random.choice(threat_sides)
            frame = _make_scene_frame(threat_side=threat_side)

            # Condition A: strong survival directive (as Super would issue)
            raw_d, lat_d = await self._call_nano(
                frame, "",
                f"PRIORITY: SURVIVAL. Enemy approaching from the {threat_side}. EVADE IMMEDIATELY. Do NOT move toward coins.",
            )
            if raw_d:
                action_d, _ = _extract_action_dir(raw_d)
                directive_results.append(TrialResult(
                    trial_id=i, latency_ms=lat_d,
                    success=(action_d in _EVASIVE_ACTIONS),
                    details={"threat_side": threat_side, "action": action_d, "condition": "directive"},
                ))
            await asyncio.sleep(0.05)

            # Condition B: vague directive (no Super guidance)
            raw_n, lat_n = await self._call_nano(frame, "", "React to the scene.")
            if raw_n:
                action_n, _ = _extract_action_dir(raw_n)
                nodirective_results.append(TrialResult(
                    trial_id=i, latency_ms=lat_n,
                    success=(action_n in _EVASIVE_ACTIONS),
                    details={"threat_side": threat_side, "action": action_n, "condition": "no_directive"},
                ))
            logger.info("Visual %d/%d: directive=%s(%s) nodirective=%s",
                        i + 1, n_each,
                        directive_results[-1].details.get("action", "?") if directive_results else "?",
                        "pass" if directive_results and directive_results[-1].success else "fail",
                        nodirective_results[-1].details.get("action", "?") if nodirective_results else "?")
            await asyncio.sleep(0.05)

        if not directive_results:
            return {"error": "all visual trials failed (API errors)"}

        # Stats for directive condition (primary VTRT measurement)
        d_lats = [r.latency_ms for r in directive_results]
        d_passed = sum(1 for r in directive_results if r.success)
        nd = len(directive_results)
        d_sorted = sorted(d_lats)
        d_ci_lo, d_ci_hi = _wilson_ci(d_passed, nd)
        d_mean = sum(d_lats) / nd
        d_std = math.sqrt(sum((l - d_mean) ** 2 for l in d_lats) / nd)

        # Stats for no-directive baseline
        nn = len(nodirective_results)
        nd_passed = sum(1 for r in nodirective_results if r.success)
        nd_ci_lo, nd_ci_hi = _wilson_ci(nd_passed, nn) if nn else (0.0, 100.0)

        return {
            "n": nd + nn,
            "n_directive": nd,
            "n_nodirective": nn,
            # Primary metric: evasion with directive
            "passed": d_passed,
            "failed": nd - d_passed,
            "pass_rate": round(d_passed / nd * 100, 1),
            "vtrt_evasion_ci_lo": d_ci_lo,
            "vtrt_evasion_ci_hi": d_ci_hi,
            "vtrt_mean_ms": round(d_mean, 1),
            "vtrt_std_ms": round(d_std, 1),
            "vtrt_min_ms": round(d_sorted[0], 1),
            "vtrt_max_ms": round(d_sorted[-1], 1),
            "vtrt_p95_ms": round(d_sorted[int(nd * 0.95)], 1),
            # Baseline (no directive) — for hierarchy validation
            "vtrt_nodirective_evasion_rate": round(nd_passed / nn * 100, 1) if nn else 0.0,
            "vtrt_nodirective_ci_lo": nd_ci_lo,
            "vtrt_nodirective_ci_hi": nd_ci_hi,
        }

    async def run_hierarchy_comparison(self, n: int = 20) -> dict:
        """Measure directive adherence: conflict scene where survival vs collection compete.
        Three conditions on the same frame (threat left + coin ahead):
          A. Survive directive  -> expect evasive action
          B. Collect directive  -> expect approach/pick action
          C. Vague directive    -> baseline (no Super guidance)
        This tests whether Super directives actually steer Nano's behavior vs the vague baseline.
        """
        logger.info("Starting %d hierarchy comparison trials (conflict-frame adherence)", n)

        _EVASIVE = {"evade_threat", "move_backward", "move_left", "move_right",
                    "sprint", "jump", "stop"}
        _COLLECT = {"navigate_to_objective", "pick_up", "move_forward",
                    "sprint", "interact"}

        # Conflict frame: threat on left, coin straight ahead
        conflict_frame = _make_scene_frame(threat_side="left")

        survive_d = (
            "PRIORITY: SURVIVAL. An enemy is approaching from the left. "
            "EVADE IMMEDIATELY. Do NOT move toward coins."
        )
        collect_d = (
            "PRIORITY: COLLECT. There is a coin directly ahead. "
            "Move toward it and pick it up. Ignore threats."
        )
        vague_d = "React to the scene."

        survive_adherent = 0
        collect_adherent = 0
        vague_evasive = 0
        vague_collect = 0

        for i in range(n):
            raw_s, _ = await self._call_nano(conflict_frame, "", survive_d)
            action_s, _ = _extract_action_dir(raw_s) if raw_s else ("idle", "none")
            if action_s in _EVASIVE:
                survive_adherent += 1

            await asyncio.sleep(0.05)

            raw_c, _ = await self._call_nano(conflict_frame, "", collect_d)
            action_c, _ = _extract_action_dir(raw_c) if raw_c else ("idle", "none")
            if action_c in _COLLECT:
                collect_adherent += 1

            await asyncio.sleep(0.05)

            raw_v, _ = await self._call_nano(conflict_frame, "", vague_d)
            action_v, _ = _extract_action_dir(raw_v) if raw_v else ("idle", "none")
            if action_v in _EVASIVE:
                vague_evasive += 1
            if action_v in _COLLECT:
                vague_collect += 1

            logger.info("Hierarchy %d/%d: survive=%s collect=%s vague=%s",
                        i + 1, n, action_s, action_c, action_v)
            await asyncio.sleep(0.05)

        survive_rate = round(survive_adherent / n * 100, 1)
        collect_rate = round(collect_adherent / n * 100, 1)
        vague_e_rate = round(vague_evasive / n * 100, 1)
        vague_c_rate = round(vague_collect / n * 100, 1)
        lift_survive = round(survive_rate - vague_e_rate, 1)
        lift_collect = round(collect_rate - vague_c_rate, 1)
        s_ci_lo, s_ci_hi = _wilson_ci(survive_adherent, n)
        c_ci_lo, c_ci_hi = _wilson_ci(collect_adherent, n)
        vs_ci_lo, vs_ci_hi = _wilson_ci(vague_evasive, n)

        return {
            "n": n,
            "survive_directive_adherence_pct": survive_rate,
            "survive_ci_lo": s_ci_lo,
            "survive_ci_hi": s_ci_hi,
            "collect_directive_adherence_pct": collect_rate,
            "collect_ci_lo": c_ci_lo,
            "collect_ci_hi": c_ci_hi,
            "vague_baseline_evasive_pct": vague_e_rate,
            "vague_evasive_ci_lo": vs_ci_lo,
            "vague_evasive_ci_hi": vs_ci_hi,
            "vague_baseline_collect_pct": vague_c_rate,
            "directive_lift_survive_pp": lift_survive,
            "directive_lift_collect_pp": lift_collect,
            # Legacy aliases
            "with_super_rate": round((survive_rate + collect_rate) / 2, 1),
            "without_super_rate": round((vague_e_rate + vague_c_rate) / 2, 1),
        }

    async def run_adversarial_trials(self, n: int = 10) -> dict:
        """Test robustness to adversarial/degenerate input frames.
        Frame types: black, white, noise, overexposed, normal (baseline).
        Measures: valid-JSON rate, idle rate, latency per frame type.
        """
        logger.info("Starting adversarial robustness trials (5 types x %d)", n)
        frame_types = ["black", "white", "noise", "overexposed", "normal"]
        results: dict[str, dict] = {}

        for ftype in frame_types:
            frame = _make_adversarial_frame(ftype)
            valid_json = 0
            non_idle = 0
            latencies = []

            for i in range(n):
                raw, lat = await self._call_nano(
                    frame, "",
                    "React to the scene. Explore the environment.",
                )
                latencies.append(lat)
                if raw:
                    action, _ = _extract_action_dir(raw)
                    if action != "":
                        valid_json += 1
                    if action not in ("idle", "stop", ""):
                        non_idle += 1
                logger.debug("Adversarial %s trial %d: lat=%.0fms", ftype, i + 1, lat)
                await asyncio.sleep(0.05)

            results[ftype] = {
                "valid_json_rate": round(valid_json / n * 100, 1),
                "non_idle_rate": round(non_idle / n * 100, 1),
                "avg_latency_ms": round(sum(latencies) / len(latencies), 1) if latencies else 0,
            }
            logger.info("Adversarial %s: valid_json=%.0f%% non_idle=%.0f%%",
                        ftype, results[ftype]["valid_json_rate"], results[ftype]["non_idle_rate"])

        # Degradation = drop in non-idle rate vs normal baseline
        normal_non_idle = results["normal"]["non_idle_rate"]
        for ftype in frame_types:
            if ftype != "normal":
                results[ftype]["degradation_pp"] = round(
                    normal_non_idle - results[ftype]["non_idle_rate"], 1
                )

        return {"n_per_type": n, "frame_types": results}

    async def run_role_distinctiveness_trials(self, n: int = 30) -> dict:
        """Measure whether NPC roles produce statistically distinct action distributions.
        Three role directives (HOSTILE / ALLY / WANDERER) on the same conflict frame.
        Reports per-role action histograms and Jensen-Shannon divergence between pairs.
        This directly validates the multi-agent claim: three roles, three distinct behaviors.
        """
        logger.info("Starting %d role-distinctiveness trials per role (3 roles)", n)

        _ROLE_DIRECTIVES = {
            "HOSTILE": (
                "You are a hostile rival. PRIORITY: AGGRESSION. "
                "Charge forward, collect coins before opponents, intimidate threats."
            ),
            "ALLY": (
                "You are a cooperative ally. PRIORITY: SUPPORT. "
                "Follow the player, collect coins together, protect from threats."
            ),
            "WANDERER": (
                "You are a wandering explorer. PRIORITY: EXPLORE. "
                "Roam freely, look around, collect coins at your own pace."
            ),
        }

        conflict_frame = _make_scene_frame(threat_side="left")
        role_actions: dict[str, list[str]] = {r: [] for r in _ROLE_DIRECTIVES}

        for role, directive in _ROLE_DIRECTIVES.items():
            for i in range(n):
                raw, _ = await self._call_nano(conflict_frame, "", directive)
                action, _ = _extract_action_dir(raw) if raw else ("idle", "none")
                role_actions[role].append(action)
                logger.info("Role %s %d/%d: action=%s", role, i + 1, n, action)
                await asyncio.sleep(0.05)

        def _hist(actions: list[str]) -> dict[str, float]:
            total = len(actions) or 1
            counts: dict[str, int] = {}
            for a in actions:
                counts[a] = counts.get(a, 0) + 1
            return {a: round(c / total, 4) for a, c in sorted(counts.items(), key=lambda x: -x[1])}

        def _jsd(p: dict, q: dict) -> float:
            """Jensen-Shannon divergence (0=identical, 1=maximally different)."""
            all_keys = set(p) | set(q)
            pv = [p.get(k, 0) for k in all_keys]
            qv = [q.get(k, 0) for k in all_keys]
            mv = [(a + b) / 2 for a, b in zip(pv, qv)]
            def _kl(a, m):
                s = 0.0
                for ai, mi in zip(a, m):
                    if ai > 0 and mi > 0:
                        s += ai * math.log(ai / mi)
                return s
            return round((_kl(pv, mv) + _kl(qv, mv)) / 2, 4)

        hists = {r: _hist(role_actions[r]) for r in _ROLE_DIRECTIVES}
        roles = list(_ROLE_DIRECTIVES.keys())

        # Modal action per role (most frequent)
        modal = {r: max(hists[r], key=hists[r].get) for r in roles}

        # JSD between all role pairs (higher = more distinct)
        jsd_pairs = {}
        for i in range(len(roles)):
            for j in range(i + 1, len(roles)):
                key = f"{roles[i]}_vs_{roles[j]}"
                jsd_pairs[key] = _jsd(hists[roles[i]], hists[roles[j]])

        mean_jsd = round(sum(jsd_pairs.values()) / len(jsd_pairs), 4)

        # Evasive rate per role (HOSTILE should be lowest, ALLY/WANDERER higher)
        evasive_rate = {
            r: round(sum(1 for a in role_actions[r] if a in _EVASIVE_ACTIONS) / n * 100, 1)
            for r in roles
        }
        collect_rate_role = {
            r: round(sum(1 for a in role_actions[r]
                         if a in {"move_forward", "navigate_to_objective", "sprint", "pick_up", "interact"})
                     / n * 100, 1)
            for r in roles
        }

        logger.info("Role distinctiveness: modal=%s JSD_mean=%.4f", modal, mean_jsd)
        return {
            "n_per_role": n,
            "roles": roles,
            "histograms": hists,
            "modal_action": modal,
            "jsd_pairs": jsd_pairs,
            "mean_jsd": mean_jsd,
            "evasive_rate_pct": evasive_rate,
            "collect_rate_pct": collect_rate_role,
        }

	"""Trial Runner - systematic evaluation trials for the paper.

	Three protocols:
	1. Audio Trials (N=50): measure ACRL and direction match (CMSS)
	2. Visual Threat Trials (N=50): measure VTRT
	3. Super+Nano Hierarchy Comparison (N=20): task completion with/without Super
	"""

	from __future__ import annotations

	import asyncio
	import base64
	import io
	import json
	import logging
	import math
	import os
	import random
	import re
	import time
	from dataclasses import dataclass, field
	from pathlib import Path

	logger = logging.getLogger(__name__)


	def _wilson_ci(k: int, n: int, z: float = 1.96) -> tuple[float, float]:
	"""Return (lo_pct, hi_pct) 95% Wilson score interval for k successes in n trials."""
	if n == 0:
	return 0.0, 100.0
	p = k / n
	denom = 1 + z * z / n
	centre = (p + z * z / (2 * n)) / denom
	half = z * math.sqrt(p * (1 - p) / n + z * z / (4 * n * n)) / denom
	return round(max(0.0, centre - half) * 100, 1), round(min(100.0, centre + half) * 100, 1)

	VOICE_COMMANDS = [
	("go left", "left"),
	("go right", "right"),
	("move forward", "forward"),
	("move backward", "backward"),
	("turn left", "left"),
	("turn right", "right"),
	("run forward", "forward"),
	("back up", "backward"),
	("stop", "none"),
	("retreat", "backward"),
	]

	_EVASIVE_ACTIONS = {
	"evade_threat", "move_backward", "move_left", "move_right",
	"sprint", "jump", "stop",
	}

	# Maps expected voice-command direction to action types that satisfy it.
	# Used for real CMSS calculation (direction alignment, not just "not idle").
	_DIRECTION_ACTIONS: dict[str, set[str]] = {
	"left": {"move_left", "rotate_left"},
	"right": {"move_right", "rotate_right"},
	"forward": {"move_forward", "navigate_to_objective", "sprint", "interact", "pick_up"},
	"backward": {"move_backward", "evade_threat", "stop"},
	}


	@dataclass
	class TrialResult:
	trial_id: int
	latency_ms: float
	success: bool
	details: dict = field(default_factory=dict)


	def _make_scene_frame(threat_side: str = None) -> str:
	"""Create a minimal game-like test JPEG using PIL."""
	from PIL import Image, ImageDraw
	img = Image.new("RGB", (128, 128), color=(30, 80, 40))
	draw = ImageDraw.Draw(img)
	draw.rectangle([0, 0, 128, 50], fill=(70, 130, 180)) # sky
	draw.rectangle([0, 50, 128, 128], fill=(30, 80, 40)) # ground
	if threat_side == "left":
	draw.rectangle([5, 25, 32, 68], fill=(200, 50, 50))
	elif threat_side == "right":
	draw.rectangle([96, 25, 123, 68], fill=(200, 50, 50))
	elif threat_side == "center":
	draw.rectangle([54, 25, 74, 68], fill=(200, 50, 50))
	draw.ellipse([58, 52, 70, 64], fill=(255, 215, 0)) # gold coin ahead
	buf = io.BytesIO()
	img.save(buf, format="JPEG", quality=60)
	return base64.b64encode(buf.getvalue()).decode()


	def _make_adversarial_frame(frame_type: str) -> str:
	"""Create degenerate/adversarial test frames for robustness evaluation."""
	from PIL import Image
	w, h = 128, 128
	if frame_type == "black":
	img = Image.new("RGB", (w, h), (0, 0, 0))
	elif frame_type == "white":
	img = Image.new("RGB", (w, h), (255, 255, 255))
	elif frame_type == "overexposed":
	img = Image.new("RGB", (w, h), (245, 245, 245))
	elif frame_type == "noise":
	try:
	import numpy as np
	arr = np.random.randint(0, 256, (h, w, 3), dtype=np.uint8)
	img = Image.fromarray(arr, "RGB")
	except ImportError:
	import random as _r
	img = Image.frombytes("RGB", (w, h), bytes(_r.randint(0, 255) for _ in range(w * h * 3)))
	else: # "normal" — standard scene frame
	return _make_scene_frame()
	buf = io.BytesIO()
	img.save(buf, format="JPEG", quality=60)
	return base64.b64encode(buf.getvalue()).decode()


	def _extract_action_dir(raw: str) -> tuple[str, str]:
	"""Return (action, direction) from raw model response."""
	cleaned = re.sub(r"<think>.*?</think>", "", raw, flags=re.DOTALL).strip()
	cleaned = re.sub(r"```[a-z]*", "", cleaned).replace("```", "").strip()
	match = re.search(r"\{.*\}", cleaned, re.DOTALL)
	if match:
	try:
	data = json.loads(match.group(0))
	return str(data.get("action", "idle")).lower(), str(data.get("direction", "none")).lower()
	except json.JSONDecodeError:
	pass
	return "idle", "none"


	class TrialRunner:
	"""Runs evaluation trials against the live OmniNPC pipeline."""

	def __init__(self, config, orchestrator=None) -> None:
	self._config = config
	self._orchestrator = orchestrator
	self._audio_cues_dir = Path("data/prompts/audio_cues")
	self._audio_cues_dir.mkdir(parents=True, exist_ok=True)
	self._nim = None

	def _get_nim(self):
	if self._nim is None:
	from inference.nim_client import NIMClient
	api_key = os.getenv("NVIDIA_NIM_API_KEY", "")
	self._nim = NIMClient(api_key=api_key, base_url=self._config.nvidia.base_url)
	return self._nim

	async def _call_nano(self, frame_b64: str, audio_text: str, directive: str) -> tuple[str, float]:
	"""Call Nano VL with a test frame. Returns (raw_response, latency_ms)."""
	from inference.prompt_templates import build_nano_system_prompt, build_nano_user_message
	nim = self._get_nim()
	system_prompt = build_nano_system_prompt(directive)
	audio_ctx = (
	f"Voice command heard: '{audio_text}'. Respond accordingly."
	if audio_text else "No voice command. React to visual input only."
	)
	user_text = build_nano_user_message(audio_ctx)
	messages = [
	{"role": "system", "content": system_prompt},
	{"role": "user", "content": [
	{"type": "text", "text": user_text},
	{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{frame_b64}"}},
	]},
	]
	t0 = time.perf_counter()
	raw = await nim.complete(
	model=self._config.nvidia.nano_omni_model,
	messages=messages,
	max_tokens=256,
	temperature=0.1,
	)
	latency_ms = (time.perf_counter() - t0) * 1000
	return raw or "", latency_ms

	async def run_audio_trials(self, n: int = 50) -> dict:
	"""Send voice-command prompts to Nano, measure ACRL and direction match (CMSS)."""
	logger.info("Starting %d audio trials", n)
	results: list[TrialResult] = []
	neutral_frame = _make_scene_frame()

	for i in range(n):
	cmd_text, expected_dir = random.choice(VOICE_COMMANDS)
	raw, latency_ms = await self._call_nano(
	neutral_frame, cmd_text,
	"Compete to collect coins. Follow any voice commands immediately.",
	)
	if not raw:
	logger.warning("Audio trial %d: empty API response, skipping", i)
	continue
	action, direction = _extract_action_dir(raw)
	# CMSS: direction-aligned success.
	# "none" (stop) → any response valid.
	# Otherwise: action type must be in the expected direction set,
	# OR the model's direction field must match the expected direction.
	if expected_dir == "none":
	success = True
	else:
	expected_acts = _DIRECTION_ACTIONS.get(expected_dir, set())
	success = (action in expected_acts) or (direction == expected_dir)
	results.append(TrialResult(
	trial_id=i,
	latency_ms=latency_ms,
	success=success,
	details={"command": cmd_text, "expected_dir": expected_dir,
	"action": action, "direction": direction},
	))
	logger.info("Audio %d/%d: '%s' -> action=%s dir=%s lat=%.0fms pass=%s",
	i + 1, n, cmd_text, action, direction, latency_ms, success)
	await asyncio.sleep(0.05)

	if not results:
	return {"error": "all audio trials failed (API errors)"}

	latencies = [r.latency_ms for r in results]
	passed = sum(1 for r in results if r.success)
	n_actual = len(results)
	sorted_l = sorted(latencies)
	ci_lo, ci_hi = _wilson_ci(passed, n_actual)
	mean_lat = sum(latencies) / n_actual
	std_lat = math.sqrt(sum((l - mean_lat) ** 2 for l in latencies) / n_actual)
	return {
	"n": n_actual,
	"passed": passed,
	"failed": n_actual - passed,
	"pass_rate": round(passed / n_actual * 100, 1),
	"cmss_ci_lo": ci_lo,
	"cmss_ci_hi": ci_hi,
	"acrl_mean_ms": round(mean_lat, 1),
	"acrl_std_ms": round(std_lat, 1),
	"acrl_min_ms": round(sorted_l[0], 1),
	"acrl_max_ms": round(sorted_l[-1], 1),
	"acrl_p95_ms": round(sorted_l[int(n_actual * 0.95)], 1),
	}

	async def run_visual_trials(self, n: int = 50) -> dict:
	"""Present threat frames to Nano, measure VTRT and evasion rate.

	Two conditions (n//2 each):
	- With directive: "EVADE IMMEDIATELY" (mirrors what Super issues in real system)
	- No directive: vague "React to the scene" (baseline behavior)

	VTRT measures latency from frame presentation to evasive action.
	Evasion rate differentiates directive vs no-directive to validate hierarchy necessity.
	"""
	logger.info("Starting %d visual trials (two conditions: directive / no-directive)", n)
	n_each = n // 2
	threat_sides = ["left", "right", "center"]

	directive_results: list[TrialResult] = []
	nodirective_results: list[TrialResult] = []

	for i in range(n_each):
	threat_side = random.choice(threat_sides)
	frame = _make_scene_frame(threat_side=threat_side)

	# Condition A: strong survival directive (as Super would issue)
	raw_d, lat_d = await self._call_nano(
	frame, "",
	f"PRIORITY: SURVIVAL. Enemy approaching from the {threat_side}. EVADE IMMEDIATELY. Do NOT move toward coins.",
	)
	if raw_d:
	action_d, _ = _extract_action_dir(raw_d)
	directive_results.append(TrialResult(
	trial_id=i, latency_ms=lat_d,
	success=(action_d in _EVASIVE_ACTIONS),
	details={"threat_side": threat_side, "action": action_d, "condition": "directive"},
	))
	await asyncio.sleep(0.05)

	# Condition B: vague directive (no Super guidance)
	raw_n, lat_n = await self._call_nano(frame, "", "React to the scene.")
	if raw_n:
	action_n, _ = _extract_action_dir(raw_n)
	nodirective_results.append(TrialResult(
	trial_id=i, latency_ms=lat_n,
	success=(action_n in _EVASIVE_ACTIONS),
	details={"threat_side": threat_side, "action": action_n, "condition": "no_directive"},
	))
	logger.info("Visual %d/%d: directive=%s(%s) nodirective=%s",
	i + 1, n_each,
	directive_results[-1].details.get("action", "?") if directive_results else "?",
	"pass" if directive_results and directive_results[-1].success else "fail",
	nodirective_results[-1].details.get("action", "?") if nodirective_results else "?")
	await asyncio.sleep(0.05)

	if not directive_results:
	return {"error": "all visual trials failed (API errors)"}

	# Stats for directive condition (primary VTRT measurement)
	d_lats = [r.latency_ms for r in directive_results]
	d_passed = sum(1 for r in directive_results if r.success)
	nd = len(directive_results)
	d_sorted = sorted(d_lats)
	d_ci_lo, d_ci_hi = _wilson_ci(d_passed, nd)
	d_mean = sum(d_lats) / nd
	d_std = math.sqrt(sum((l - d_mean) ** 2 for l in d_lats) / nd)

	# Stats for no-directive baseline
	nn = len(nodirective_results)
	nd_passed = sum(1 for r in nodirective_results if r.success)
	nd_ci_lo, nd_ci_hi = _wilson_ci(nd_passed, nn) if nn else (0.0, 100.0)

	return {
	"n": nd + nn,
	"n_directive": nd,
	"n_nodirective": nn,
	# Primary metric: evasion with directive
	"passed": d_passed,
	"failed": nd - d_passed,
	"pass_rate": round(d_passed / nd * 100, 1),
	"vtrt_evasion_ci_lo": d_ci_lo,
	"vtrt_evasion_ci_hi": d_ci_hi,
	"vtrt_mean_ms": round(d_mean, 1),
	"vtrt_std_ms": round(d_std, 1),
	"vtrt_min_ms": round(d_sorted[0], 1),
	"vtrt_max_ms": round(d_sorted[-1], 1),
	"vtrt_p95_ms": round(d_sorted[int(nd * 0.95)], 1),
	# Baseline (no directive) — for hierarchy validation
	"vtrt_nodirective_evasion_rate": round(nd_passed / nn * 100, 1) if nn else 0.0,
	"vtrt_nodirective_ci_lo": nd_ci_lo,
	"vtrt_nodirective_ci_hi": nd_ci_hi,
	}

	async def run_hierarchy_comparison(self, n: int = 20) -> dict:
	"""Measure directive adherence: conflict scene where survival vs collection compete.

	Three conditions on the same frame (threat left + coin ahead):
	A. Survive directive -> expect evasive action
	B. Collect directive -> expect approach/pick action
	C. Vague directive -> baseline (no Super guidance)

	This tests whether Super directives actually steer Nano's behavior vs the vague baseline.
	"""
	logger.info("Starting %d hierarchy comparison trials (conflict-frame adherence)", n)

	_EVASIVE = {"evade_threat", "move_backward", "move_left", "move_right",
	"sprint", "jump", "stop"}
	_COLLECT = {"navigate_to_objective", "pick_up", "move_forward",
	"sprint", "interact"}

	# Conflict frame: threat on left, coin straight ahead
	conflict_frame = _make_scene_frame(threat_side="left")

	survive_d = (
	"PRIORITY: SURVIVAL. An enemy is approaching from the left. "
	"EVADE IMMEDIATELY. Do NOT move toward coins."
	)
	collect_d = (
	"PRIORITY: COLLECT. There is a coin directly ahead. "
	"Move toward it and pick it up. Ignore threats."
	)
	vague_d = "React to the scene."

	survive_adherent = 0
	collect_adherent = 0
	vague_evasive = 0
	vague_collect = 0

	for i in range(n):
	raw_s, _ = await self._call_nano(conflict_frame, "", survive_d)
	action_s, _ = _extract_action_dir(raw_s) if raw_s else ("idle", "none")
	if action_s in _EVASIVE:
	survive_adherent += 1

	await asyncio.sleep(0.05)

	raw_c, _ = await self._call_nano(conflict_frame, "", collect_d)
	action_c, _ = _extract_action_dir(raw_c) if raw_c else ("idle", "none")
	if action_c in _COLLECT:
	collect_adherent += 1

	await asyncio.sleep(0.05)

	raw_v, _ = await self._call_nano(conflict_frame, "", vague_d)
	action_v, _ = _extract_action_dir(raw_v) if raw_v else ("idle", "none")
	if action_v in _EVASIVE:
	vague_evasive += 1
	if action_v in _COLLECT:
	vague_collect += 1

	logger.info("Hierarchy %d/%d: survive=%s collect=%s vague=%s",
	i + 1, n, action_s, action_c, action_v)
	await asyncio.sleep(0.05)

	survive_rate = round(survive_adherent / n * 100, 1)
	collect_rate = round(collect_adherent / n * 100, 1)
	vague_e_rate = round(vague_evasive / n * 100, 1)
	vague_c_rate = round(vague_collect / n * 100, 1)
	lift_survive = round(survive_rate - vague_e_rate, 1)
	lift_collect = round(collect_rate - vague_c_rate, 1)
	s_ci_lo, s_ci_hi = _wilson_ci(survive_adherent, n)
	c_ci_lo, c_ci_hi = _wilson_ci(collect_adherent, n)
	vs_ci_lo, vs_ci_hi = _wilson_ci(vague_evasive, n)

	return {
	"n": n,
	"survive_directive_adherence_pct": survive_rate,
	"survive_ci_lo": s_ci_lo,
	"survive_ci_hi": s_ci_hi,
	"collect_directive_adherence_pct": collect_rate,
	"collect_ci_lo": c_ci_lo,
	"collect_ci_hi": c_ci_hi,
	"vague_baseline_evasive_pct": vague_e_rate,
	"vague_evasive_ci_lo": vs_ci_lo,
	"vague_evasive_ci_hi": vs_ci_hi,
	"vague_baseline_collect_pct": vague_c_rate,
	"directive_lift_survive_pp": lift_survive,
	"directive_lift_collect_pp": lift_collect,
	# Legacy aliases
	"with_super_rate": round((survive_rate + collect_rate) / 2, 1),
	"without_super_rate": round((vague_e_rate + vague_c_rate) / 2, 1),
	}

	async def run_adversarial_trials(self, n: int = 10) -> dict:
	"""Test robustness to adversarial/degenerate input frames.

	Frame types: black, white, noise, overexposed, normal (baseline).
	Measures: valid-JSON rate, idle rate, latency per frame type.
	"""
	logger.info("Starting adversarial robustness trials (5 types x %d)", n)
	frame_types = ["black", "white", "noise", "overexposed", "normal"]
	results: dict[str, dict] = {}

	for ftype in frame_types:
	frame = _make_adversarial_frame(ftype)
	valid_json = 0
	non_idle = 0
	latencies = []

	for i in range(n):
	raw, lat = await self._call_nano(
	frame, "",
	"React to the scene. Explore the environment.",
	)
	latencies.append(lat)
	if raw:
	action, _ = _extract_action_dir(raw)
	if action != "":
	valid_json += 1
	if action not in ("idle", "stop", ""):
	non_idle += 1
	logger.debug("Adversarial %s trial %d: lat=%.0fms", ftype, i + 1, lat)
	await asyncio.sleep(0.05)

	results[ftype] = {
	"valid_json_rate": round(valid_json / n * 100, 1),
	"non_idle_rate": round(non_idle / n * 100, 1),
	"avg_latency_ms": round(sum(latencies) / len(latencies), 1) if latencies else 0,
	}
	logger.info("Adversarial %s: valid_json=%.0f%% non_idle=%.0f%%",
	ftype, results[ftype]["valid_json_rate"], results[ftype]["non_idle_rate"])

	# Degradation = drop in non-idle rate vs normal baseline
	normal_non_idle = results["normal"]["non_idle_rate"]
	for ftype in frame_types:
	if ftype != "normal":
	results[ftype]["degradation_pp"] = round(
	normal_non_idle - results[ftype]["non_idle_rate"], 1
	)

	return {"n_per_type": n, "frame_types": results}

	async def run_role_distinctiveness_trials(self, n: int = 30) -> dict:
	"""Measure whether NPC roles produce statistically distinct action distributions.

	Three role directives (HOSTILE / ALLY / WANDERER) on the same conflict frame.
	Reports per-role action histograms and Jensen-Shannon divergence between pairs.
	This directly validates the multi-agent claim: three roles, three distinct behaviors.
	"""
	logger.info("Starting %d role-distinctiveness trials per role (3 roles)", n)

	_ROLE_DIRECTIVES = {
	"HOSTILE": (
	"You are a hostile rival. PRIORITY: AGGRESSION. "
	"Charge forward, collect coins before opponents, intimidate threats."
	),
	"ALLY": (
	"You are a cooperative ally. PRIORITY: SUPPORT. "
	"Follow the player, collect coins together, protect from threats."
	),
	"WANDERER": (
	"You are a wandering explorer. PRIORITY: EXPLORE. "
	"Roam freely, look around, collect coins at your own pace."
	),
	}

	conflict_frame = _make_scene_frame(threat_side="left")
	role_actions: dict[str, list[str]] = {r: [] for r in _ROLE_DIRECTIVES}

	for role, directive in _ROLE_DIRECTIVES.items():
	for i in range(n):
	raw, _ = await self._call_nano(conflict_frame, "", directive)
	action, _ = _extract_action_dir(raw) if raw else ("idle", "none")
	role_actions[role].append(action)
	logger.info("Role %s %d/%d: action=%s", role, i + 1, n, action)
	await asyncio.sleep(0.05)

	def _hist(actions: list[str]) -> dict[str, float]:
	total = len(actions) or 1
	counts: dict[str, int] = {}
	for a in actions:
	counts[a] = counts.get(a, 0) + 1
	return {a: round(c / total, 4) for a, c in sorted(counts.items(), key=lambda x: -x[1])}

	def _jsd(p: dict, q: dict) -> float:
	"""Jensen-Shannon divergence (0=identical, 1=maximally different)."""
	all_keys = set(p) \| set(q)
	pv = [p.get(k, 0) for k in all_keys]
	qv = [q.get(k, 0) for k in all_keys]
	mv = [(a + b) / 2 for a, b in zip(pv, qv)]
	def _kl(a, m):
	s = 0.0
	for ai, mi in zip(a, m):
	if ai > 0 and mi > 0:
	s += ai * math.log(ai / mi)
	return s
	return round((_kl(pv, mv) + _kl(qv, mv)) / 2, 4)

	hists = {r: _hist(role_actions[r]) for r in _ROLE_DIRECTIVES}
	roles = list(_ROLE_DIRECTIVES.keys())

	# Modal action per role (most frequent)
	modal = {r: max(hists[r], key=hists[r].get) for r in roles}

	# JSD between all role pairs (higher = more distinct)
	jsd_pairs = {}
	for i in range(len(roles)):
	for j in range(i + 1, len(roles)):
	key = f"{roles[i]}_vs_{roles[j]}"
	jsd_pairs[key] = _jsd(hists[roles[i]], hists[roles[j]])

	mean_jsd = round(sum(jsd_pairs.values()) / len(jsd_pairs), 4)

	# Evasive rate per role (HOSTILE should be lowest, ALLY/WANDERER higher)
	evasive_rate = {
	r: round(sum(1 for a in role_actions[r] if a in _EVASIVE_ACTIONS) / n * 100, 1)
	for r in roles
	}
	collect_rate_role = {
	r: round(sum(1 for a in role_actions[r]
	if a in {"move_forward", "navigate_to_objective", "sprint", "pick_up", "interact"})
	/ n * 100, 1)
	for r in roles
	}

	logger.info("Role distinctiveness: modal=%s JSD_mean=%.4f", modal, mean_jsd)
	return {
	"n_per_role": n,
	"roles": roles,
	"histograms": hists,
	"modal_action": modal,
	"jsd_pairs": jsd_pairs,
	"mean_jsd": mean_jsd,
	"evasive_rate_pct": evasive_rate,
	"collect_rate_pct": collect_rate_role,
	}