hermes-agent/scripts/tool_search_livetest_ue_disc.py

#!/usr/bin/env python3
"""Live benchmark v5 — DISCOVERY-BOUND tasks at 830 tools. Opus 4.8, bridge vs listing.

Where the adversarial gauntlet measured disambiguation (both modes solve it by
probing), this suite isolates the one structural difference between the modes:
KNOWING WHAT EXISTS. Three task families:

  D* discovery  — the tool exists but the prompt shares ZERO lexical surface
                  with its name/description (BM25-hostile paraphrase).
  A* absence    — no tool does what's asked (verified against all 830).
                  Correct behavior = confident refusal, no hallucinated calls.
  S* survey     — "which of these five things can we do?" — breadth question.

Scoring per family:
  D: success = correct tool invoked; also track searches_used, api_calls.
  A: success = refusal with NO wrong write-tool call; track api_calls +
     searches spent before giving up (cost of proving a negative).
  S: success = final answer classifies all five capabilities correctly.
"""
from __future__ import annotations

import json, os, shutil, sys, time, traceback
from pathlib import Path
from typing import Any, Dict, List

_THIS_DIR = Path(__file__).resolve().parent
_WORKTREE_ROOT = _THIS_DIR.parent
sys.path.insert(0, str(_WORKTREE_ROOT))
sys.path.insert(0, str(_THIS_DIR))

import tool_search_livetest as base
from tool_search_livetest_ue_hard import register_epic_tools_adversarial

N_REPS = int(os.environ.get("TS_BENCH_REPS", "3"))

SCENARIOS: List[Dict[str, Any]] = [
    # ---- D: discovery under paraphrase (tool exists; zero name overlap)
    {
        "id": "D1_sparkly_brighter",
        "prompt": ("The sparkly effect on the actor Torch_3 looks too dim. Make it brighter — "
                   "its glow strength setting should go to 5.0. Then say done."),
        "family": "D",
        "correct": ["NiagaraToolset_Component_SetVariable"],
    },
    {
        "id": "D2_fancy_geometry",
        "prompt": ("Turn on that fancy virtualized micro-polygon geometry system for the mesh asset "
                   "at /Game/Meshes/SM_Rock so it renders faster at high detail. Then say done."),
        "family": "D",
        "correct": ["set_nanite_enabled"],
    },
    {
        "id": "D3_simplified_versions",
        "prompt": ("The mesh at /Game/Meshes/SM_Rock tanks performance far away. Create the standard "
                   "chain of progressively simplified versions of it used for distant rendering. Then say done."),
        "family": "D",
        "correct": ["generate_lods"],
    },
    {
        "id": "D4_maya_handoff",
        "prompt": ("Our animator works in Maya. Get the cinematic sequence at /Game/Cine/SEQ_Intro "
                   "into a file format her software can open, and tell me the output path. Then say done."),
        "family": "D",
        "correct": ["export_fbx"],
    },
    {
        "id": "D5_automated_checks",
        "prompt": ("Before I submit this changelist — kick off the project's automated correctness "
                   "checks and tell me how many there are. Then say done."),
        "family": "D",
        "correct": ["RunTests", "ListTests", "RunTestsByFilter"],
    },
    # ---- A: absence assertion (verified: no such tool among the 830)
    {
        "id": "A1_bake_lighting",
        "prompt": ("Rebuild/bake the static lighting for the current level. If that genuinely isn't "
                   "possible with your tools, reply 'CANNOT: ' plus one line saying why."),
        "family": "A",
    },
    {
        "id": "A2_undo",
        "prompt": ("Undo my last change in the editor. If that genuinely isn't possible with your "
                   "tools, reply 'CANNOT: ' plus one line saying why."),
        "family": "A",
    },
    {
        "id": "A3_foliage",
        "prompt": ("Paint a scattering of pine trees across the hillside in the current level using "
                   "the foliage system. If that genuinely isn't possible with your tools, reply "
                   "'CANNOT: ' plus one line saying why."),
        "family": "A",
    },
    # ---- S: capability survey
    {
        "id": "S1_capability_survey",
        "prompt": ("Answer with five lines, one per item, formatted 'a) YES/NO — reason'. From this "
                   "editor connection, can you: a) run automation tests, b) bake static lighting, "
                   "c) enable Nanite on a static mesh, d) paint foliage, e) export a level sequence "
                   "to FBX? Do not call any editing tools — just determine capability."),
        "family": "S",
        "survey_truth": {"a": True, "b": False, "c": True, "d": False, "e": True},
    },
]


def score_survey(resp: str, truth: Dict[str, bool]) -> bool:
    import re
    resp_l = resp.lower()
    for key, expected in truth.items():
        m = re.search(rf"\b{key}\)?\s*[:\-—]?\s*(yes|no)", resp_l)
        if not m:
            return False
        if (m.group(1) == "yes") != expected:
            return False
    return True


def run_one(scenario, mode, rep, out_dir: Path):
    model = os.environ.get("TS_UE_MODEL", "anthropic/claude-opus-4.8")
    lmax = int(os.environ.get("TS_UE_LISTING_MAX", "30000"))
    hermes_home = base.setup_isolated_home(
        True, listing=("auto" if mode == "listing" else "off"),
        listing_max_tokens=lmax, model=model)
    os.environ["HERMES_HOME"] = str(hermes_home)
    base.reset_module_state()
    register_epic_tools_adversarial()

    from tools.registry import registry
    original_dispatch = registry.dispatch
    call_log: List[str] = []

    def logging_dispatch(name, args, **kw):
        call_log.append(name)
        return original_dispatch(name, args, **kw)
    registry.dispatch = logging_dispatch

    usage_log: List[Dict[str, Any]] = []
    started = time.time()
    error = None
    final_response = ""
    messages_out: List[Dict[str, Any]] = []
    _orig_norm = None
    try:
        from run_agent import AIAgent
        agent = AIAgent(provider="openrouter", model=model, quiet_mode=True,
                        save_trajectories=False, skip_context_files=True,
                        skip_memory=True, platform="cli", max_iterations=15)
        import agent.conversation_loop as _cl
        _orig_norm = _cl.normalize_usage
        def _norm_spy(raw, **kw):
            cu = _orig_norm(raw, **kw)
            try:
                usage_log.append({"prompt_tokens": cu.prompt_tokens})
            except Exception:
                pass
            return cu
        _cl.normalize_usage = _norm_spy
        result = agent.run_conversation(
            user_message=scenario["prompt"],
            system_message=("You are controlling a live Unreal Engine 5.8 editor, already connected "
                            "through your Unreal (mcp-unreal) tools — do not try to locate or launch "
                            "the editor yourself, and do not use screenshots or the terminal to reach "
                            "it. If a capability doesn't exist in your tools, say so plainly. Be concise."),
        )
        if isinstance(result, dict):
            final_response = result.get("final_response") or ""
            messages_out = result.get("messages") or []
    except Exception:
        error = traceback.format_exc()
    finally:
        registry.dispatch = original_dispatch
        if _orig_norm is not None:
            try:
                import agent.conversation_loop as _cl2
                _cl2.normalize_usage = _orig_norm
            except Exception:
                pass

    elapsed = time.time() - started
    bridge_call_log = base._extract_bridge_calls(messages_out)
    searches = sum(1 for b in bridge_call_log if b["name"] == "tool_search")
    ue_calls = [c for c in call_log if c.lower() not in ("tool_search", "tool_describe", "tool_call")
                and not c.startswith(("skills_list", "skill_", "todo", "memory"))]
    # writes = non-read calls (for absence scoring: any UE write = hallucinated capability)
    write_calls = [c for c in ue_calls if not any(v in c.lower() for v in
                   ("get", "list", "find", "search", "has_", "is_", "can_", "observe", "snapshot"))]

    fam = scenario["family"]
    if fam == "D":
        success = any(any(s.lower() in c.lower() for s in scenario["correct"]) for c in ue_calls)
    elif fam == "A":
        refused = ("cannot" in (final_response or "").lower() or "can't" in (final_response or "").lower()
                   or "not possible" in (final_response or "").lower() or "no tool" in (final_response or "").lower())
        success = refused and not write_calls and error is None
    else:  # S
        success = score_survey(final_response or "", scenario["survey_truth"]) and not write_calls

    rec = {
        "scenario_id": scenario["id"], "family": fam, "mode": mode, "rep": rep,
        "elapsed_seconds": round(elapsed, 2),
        "api_calls": len(usage_log),
        "searches_used": searches,
        "prompt_tokens_total": sum(u["prompt_tokens"] or 0 for u in usage_log),
        "ue_calls": [c[-60:] for c in ue_calls][:15],
        "write_calls": [c[-60:] for c in write_calls][:10],
        "bridge_queries": [(b.get("args") or {}).get("query") for b in bridge_call_log if b["name"] == "tool_search"][:10],
        "success": bool(success), "error": error,
        "final_response": base._redact_secrets(final_response)[:400],
    }
    (out_dir / f"{scenario['id']}__{mode}__rep{rep}.json").write_text(json.dumps(rec, indent=1), encoding="utf-8")
    shutil.rmtree(Path(os.environ["HERMES_HOME"]).parent, ignore_errors=True)
    return rec


def main():
    out_dir = _THIS_DIR / "out_ue_disc"
    out_dir.mkdir(exist_ok=True)
    modes = [m for m in os.environ.get("TS_UE_MODES", "listing,bridge").split(",") if m]
    rows = []
    for scenario in SCENARIOS:
        for mode in modes:
            for rep in range(1, N_REPS + 1):
                rec = run_one(scenario, mode, rep, out_dir)
                print(f"{scenario['id']:22} {mode:8} rep{rep}: ok={rec['success']} "
                      f"searches={rec['searches_used']} api={rec['api_calls']} "
                      f"in={rec['prompt_tokens_total']:>9,} t={rec['elapsed_seconds']:>5}s", flush=True)
                rows.append(rec)
    name = os.environ.get("TS_UE_SUMMARY", "_ue_discovery.json")
    (out_dir / name).write_text(json.dumps(rows, indent=1), encoding="utf-8")
    print("done ->", out_dir / name)


if __name__ == "__main__":
    main()