bench: discovery-bound suite — paraphrase/absence/survey tasks isolate the listing's structural advantage

Bridge vs listing only (Opus 4.8, 830 real UE schemas, 3 reps/cell).
Excluding one both-modes mock artifact: listing 24/24 vs bridge 20/24,
searches/task 0.2 vs 4.0. Bridge failures: core-tool substitution at
frontier tier (ran the host test suite via terminal instead of
discovering RunTests, 2/3 reps), up to 8 searches to prove a negative,
and search-vocabulary misses on paraphrase. Listing asserts absence in
zero searches and answers a 5-way capability survey in 1 API call.

scripts/tool_search_livetest_ue_disc.py

@@ -0,0 +1,234 @@
+#!/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()