Merge origin/main into feat/hermes-relay-shared-metrics

Signed-off-by: Alex Fournier <afournier@nvidia.com>

scripts/add_contributor.py

@@ -25,7 +25,11 @@ REPO_ROOT = Path(__file__).resolve().parent.parent
 EMAILS_DIR = REPO_ROOT / "contributors" / "emails"
 
 _EMAIL_RE = re.compile(r"^[^/\\\s]+@[^/\\\s]+$")
-_LOGIN_RE = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9]|-(?=[A-Za-z0-9])){0,38}$")
+# GitHub's *current* signup rules forbid consecutive hyphens, but legacy
+# accounts with them exist and are valid (e.g. Roger--Han, verified via the
+# users API July 2026). Accept any alphanumeric/hyphen login that doesn't
+# start or end with a hyphen, max 39 chars.
+_LOGIN_RE = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9-]{0,37}[A-Za-z0-9])?$")
 
 
 def read_mapping_file(path: Path) -> str | None:

scripts/check-windows-footguns.py

@@ -324,6 +324,83 @@ FOOTGUNS: list[Footgun] = [
             "    pass  # Windows asyncio doesn't support signal handlers"
         ),
     ),
+    Footgun(
+        name="subprocess text=True without explicit encoding=",
+        # Match ``text=True`` (or ``text = True``) anywhere on a line. We
+        # rely on the post_filter to (a) skip lines that already pass
+        # ``encoding=`` on the same line, and (b) skip false positives like
+        # ``def text(self, ...)`` or string literals. ``text=True`` is
+        # overwhelmingly a subprocess kwarg, so a bare match + filter has a
+        # high signal-to-noise ratio and avoids the complexity of parsing
+        # multi-line subprocess calls (which the line-based scanner can't
+        # reliably attribute to a single line anyway).
+        pattern=re.compile(r"\btext\s*=\s*True\b"),
+        message=(
+            "subprocess text=True without explicit encoding= decodes "
+            "child output with locale.getpreferredencoding() — cp936 "
+            "(GBK) on Chinese Windows, cp1252 on Western Windows — "
+            "which crashes _readerthread with UnicodeDecodeError on "
+            "non-default-codepage bytes. Always pass encoding='utf-8' "
+            "(and errors='replace' for Windows-native CLIs that emit "
+            "non-UTF-8). See issues #47939, #53428, #57238."
+        ),
+        fix=(
+            "subprocess.run(..., text=True, encoding='utf-8', "
+            "errors='replace')\n"
+            "Both params are required: encoding alone still crashes on "
+            "non-UTF-8 bytes from Windows-native CLIs (tasklist, "
+            "schtasks)."
+        ),
+        post_filter=lambda m, line: (
+            # Skip if the same line already specifies encoding=.
+            "encoding=" not in line
+            and "encoding =" not in line
+            # Skip method definitions named ``text`` (def text(self, ...)).
+            and not line.lstrip().startswith("def ")
+            and not line.lstrip().startswith("async def ")
+            # Skip ``text=True`` inside string literals (heuristic: the
+            # substring appears between matching quotes that aren't part
+            # of an f-string expression). This is imperfect but catches
+            # the common case of docstrings mentioning text=True.
+            and not _looks_like_string_literal(line, m)
+            # Skip lines that are obviously not subprocess calls — e.g.
+            # DataFrame.rename(text=True) or similar. We can't know for
+            # sure without parsing, so we accept some false negatives by
+            # only flagging when ``subprocess`` or a known subprocess-
+            # shaped call (run/Popen/call/check_output/check_call/
+            # check_output) appears on the same line. This keeps the
+            # rule focused on the actual footgun.
+            and _is_likely_subprocess_call(line)
+        ),
+    ),
+    Footgun(
+        name="bare Path.read_text()/write_text() without encoding=",
+        # Match ``.read_text(`` / ``.write_text(`` when the same line does
+        # not pass ``encoding=``. Multi-line calls where encoding= sits on
+        # a later line are handled by the post_filter's lookahead-free
+        # heuristic accepting a small false-negative rate — the AST guard
+        # test in tests/gateway/test_gateway_utf8_encoding.py catches the
+        # gateway/adapters exactly, and this rule catches the common
+        # single-line form everywhere else.
+        pattern=re.compile(r"\.(read_text|write_text)\s*\("),
+        message=(
+            "Path.read_text()/write_text() without encoding= uses "
+            "locale.getpreferredencoding() — cp936/cp1252 on Windows — "
+            "so UTF-8 content (config JSON, session state, skills) "
+            "crashes with UnicodeDecodeError or writes mojibake. "
+            "See issue #37423 and the #71014 / read_text campaign."
+        ),
+        fix='path.read_text(encoding="utf-8") / path.write_text(data, encoding="utf-8")',
+        post_filter=lambda m, line: (
+            "encoding=" not in line
+            and "encoding =" not in line
+            and not _looks_like_string_literal(line, m)
+            # Skip calls that continue onto the next line — the closing
+            # paren isn't on this line, so encoding= may follow. AST-level
+            # enforcement for those lives in the gateway guard test.
+            and line.rstrip().endswith(")")
+        ),
+    ),
 ]
 
 
@@ -408,6 +485,66 @@ def _find_unquoted_hash(line: str) -> int | None:
     return None
 
 
+# Subprocess method names that accept ``text=`` and are affected by the
+# encoding-default footgun. Used by ``_is_likely_subprocess_call`` below to
+# keep the ``text=True`` rule focused on subprocess calls (and avoid flagging
+# unrelated APIs that happen to accept a ``text`` kwarg).
+_SUBPROCESS_METHODS = (
+    "subprocess.run",
+    "subprocess.Popen",
+    "subprocess.call",
+    "subprocess.check_output",
+    "subprocess.check_call",
+    "_sp.run",            # common alias
+    "_sp.Popen",
+    "_sp.check_output",
+    "_sp.check_call",
+    "_sp.call",
+    ".run(",              # bare .run( — usually subprocess.run
+    ".Popen(",
+    ".check_output(",
+    ".check_call(",
+    ".call(",
+)
+
+
+def _is_likely_subprocess_call(line: str) -> bool:
+    """Heuristic: does this line look like a subprocess invocation?
+
+    The ``text=True`` footgun rule only fires when the matched line also
+    contains a subprocess-shaped call site. This avoids false positives on
+    unrelated APIs that accept a ``text`` kwarg (e.g. DataFrame.rename,
+    custom library calls). Multi-line calls where the ``subprocess.X(``
+    prefix is on a previous line won't be flagged — that's an acceptable
+    false negative for a line-based scanner.
+    """
+    return any(token in line for token in _SUBPROCESS_METHODS)
+
+
+def _looks_like_string_literal(line: str, match: "re.Match") -> bool:
+    """Heuristic: is the ``text=True`` match inside a string literal?
+
+    Catches the common case of docstrings/comments that mention ``text=True``
+    as prose. Walks the line tracking single/double quote state and returns
+    True if the match start index falls inside a quoted region.
+    """
+    start = match.start()
+    in_s = False
+    in_d = False
+    i = 0
+    while i < start and i < len(line):
+        c = line[i]
+        if c == "\\" and (in_s or in_d) and i + 1 < len(line):
+            i += 2
+            continue
+        if not in_d and c == "'":
+            in_s = not in_s
+        elif not in_s and c == '"':
+            in_d = not in_d
+        i += 1
+    return in_s or in_d
+
+
 def scan_file(path: Path, footguns: list[Footgun]) -> list[tuple[int, str, Footgun]]:
     """Return a list of (line_number, line, footgun) for unsuppressed matches."""
     try:
@@ -492,7 +629,7 @@ def get_staged_files() -> list[Path]:
             ["git", "diff", "--cached", "--name-only", "--diff-filter=ACMR"],
             cwd=REPO_ROOT,
             stderr=subprocess.DEVNULL,
-            text=True,
+            text=True, encoding='utf-8', errors='replace',
         )
     except (subprocess.CalledProcessError, FileNotFoundError):
         return []
@@ -506,7 +643,7 @@ def get_diff_files(ref: str) -> list[Path]:
             ["git", "diff", f"{ref}...HEAD", "--name-only", "--diff-filter=ACMR"],
             cwd=REPO_ROOT,
             stderr=subprocess.DEVNULL,
-            text=True,
+            text=True, encoding='utf-8', errors='replace',
         )
     except (subprocess.CalledProcessError, FileNotFoundError):
         return []

scripts/check_subprocess_stdin.py

@@ -181,7 +181,7 @@ def main() -> int:
             if any(skip.rstrip("/") in parts for skip in SKIP_DIRS):
                 continue
 
-            content = py_file.read_text()
+            content = py_file.read_text(encoding="utf-8")
             violations = find_subprocess_calls(content, rel)
             all_violations.extend(violations)
 
@@ -205,7 +205,7 @@ def main() -> int:
                 continue
 
             try:
-                content = py_file.read_text()
+                content = py_file.read_text(encoding="utf-8")
             except Exception:
                 continue
             violations = find_subprocess_calls(content, rel)

scripts/ci/assemble_review_comment.py

@@ -419,7 +419,7 @@ def main() -> int:
         pending_jobs=pending,
     )
 
-    args.output.write_text(body)
+    args.output.write_text(body, encoding="utf-8")
     print(f"Wrote {len(body)} chars to {args.output}")
     return 0
 

scripts/ci/lockfile_diff.py

@@ -115,7 +115,7 @@ def _git_show(ref: str, path: str, repo_root: str) -> str | None:
     proc = subprocess.run(
         ["git", "show", f"{ref}:{path}"],
         capture_output=True,
-        text=True,
+        text=True, encoding="utf-8", errors="replace",
         cwd=repo_root,
     )
     return proc.stdout if proc.returncode == 0 else None
@@ -125,7 +125,7 @@ def _tracked_lockfiles(ref: str, repo_root: str) -> set[str]:
     proc = subprocess.run(
         ["git", "ls-tree", "-r", "--name-only", ref],
         capture_output=True,
-        text=True,
+        text=True, encoding="utf-8", errors="replace",
         cwd=repo_root,
         check=True,
     )

scripts/ci/publish_e2e_evidence.py

@@ -246,7 +246,7 @@ def upload_evidence(
                 ],
                 check=True,
                 capture_output=True,
-                text=True,
+                text=True, encoding="utf-8", errors="replace",
                 env=environment,
             )
         except subprocess.CalledProcessError as exc:

scripts/contributor_audit.py

@@ -87,7 +87,7 @@ def git(*args, cwd=None):
     result = subprocess.run(
         ["git"] + list(args),
         capture_output=True,
-        text=True,
+        text=True, encoding='utf-8', errors='replace',
         cwd=cwd or str(REPO_ROOT),
     )
     if result.returncode != 0:
@@ -112,7 +112,7 @@ def gh_pr_list():
                 "--limit", "300",
             ],
             capture_output=True,
-            text=True,
+            text=True, encoding='utf-8', errors='replace',
             timeout=60,
         )
         if result.returncode != 0:

scripts/generate_conformance_vectors.py

@@ -0,0 +1,266 @@
+#!/usr/bin/env python3
+"""Conformance-vector generator — the native adapters as executable spec.
+
+Renders a shared corpus (markdown grid + scar tissue + adversarial agent
+output) through the NATIVE platform renderers and dumps input→output JSON
+vectors, stamped with the oracle commit. The gateway-gateway connector
+commits these under conformance/vectors/ and its vitest runner asserts the
+CONNECTOR's constructed REST payloads against them — so cross-repo renderer
+drift breaks a test instead of a user's formatting.
+
+Oracles (all imported, never reimplemented):
+  telegram  plugins.platforms.telegram.adapter.TelegramAdapter.format_message
+            (standard markdown → Telegram MarkdownV2)
+  slack     plugins.platforms.slack.adapter.SlackAdapter.format_message
+            (standard markdown → Slack mrkdwn)
+  whatsapp  gateway.platforms.whatsapp_common.WhatsAppBehaviorMixin
+            .format_message (standard markdown → WhatsApp formatting)
+  discord   plugins.platforms.discord.adapter.DiscordAdapter.format_message
+            (GFM tables → bullet groups; otherwise identity)
+
+Expect semantics (consumed by the gg runner):
+  parity    connector render must BYTE-EQUAL native_output
+            (Slack / WhatsApp — same-dialect ports; most Discord).
+  semantic  connector renders a DIFFERENT representation on purpose
+            (Telegram: connector sends HTML, native sends MarkdownV2);
+            the runner asserts plain-text content equivalence instead.
+  divergent documented no-parity (e.g. Discord tables: native converts to
+            bullets, connector passes raw markdown through). The runner
+            asserts the DOCUMENTED connector behavior named in `note`.
+
+Run:  python scripts/generate_conformance_vectors.py [--out DIR]
+Determinism is covered by tests/conformance/test_vector_generator.py.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any, Callable, Dict, List
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO_ROOT))
+
+GENERATOR_VERSION = 1
+
+# ── corpus ───────────────────────────────────────────────────────────────
+# Every entry: (id, category, input). Categories: grid | scar | adversarial.
+# Ids are STABLE API — the runner and divergence notes key on them.
+
+GRID: List[tuple] = [
+    ("plain-text", "Just a plain sentence."),
+    ("bold", "This is **bold** text."),
+    ("italic", "This is *italic* text."),
+    ("bold-italic", "Mix of **bold** and *italic* in one line."),
+    ("strikethrough", "This is ~~struck~~ text."),
+    ("inline-code", "Run `pip install hermes` to start."),
+    ("fenced-code", "```\nprint('hello')\n```"),
+    ("fenced-code-lang", "```python\ndef f(x):\n    return x * 2\n```"),
+    ("link", "See [the docs](https://example.com/docs) for more."),
+    ("link-parens-url", "See [spec](https://example.com/a_(b)) here."),
+    ("header-h1", "# Big Title\nBody follows."),
+    ("header-h2", "## Section\nBody follows."),
+    ("header-h3", "### Sub-section\nBody follows."),
+    ("ul-list", "- first\n- second\n- third"),
+    ("ol-list", "1. first\n2. second\n3. third"),
+    ("nested-list", "- outer\n  - inner one\n  - inner two\n- outer two"),
+    ("blockquote", "> quoted wisdom\nregular line"),
+    ("hrule", "above\n\n---\n\nbelow"),
+    (
+        "table-simple",
+        "| name | value |\n|------|-------|\n| a    | 1     |\n| b    | 2     |",
+    ),
+    ("emoji", "Done ✅ with 🎉 emoji 👀 test."),
+    ("cjk", "中文测试：**粗体** 和 `代码` 混排。"),
+    ("bare-url", "Visit https://example.com/path?q=1&r=2 today."),
+    (
+        "mixed-document",
+        "## Report\n\nStatus: **green**. Details in `runbook.md`.\n\n"
+        "- item *one*\n- item **two**\n\n```sh\nmake deploy\n```\n\n"
+        "See [dashboard](https://grafana.example.com/d/x).",
+    ),
+]
+
+SCAR: List[tuple] = [
+    # MarkdownV2 reserved characters in prose — the classic Telegram 400.
+    ("mdv2-reserved-chars", "Price is 3.50 (was 4.00) — save ~12%! #deal +tax = win."),
+    ("mdv2-underscores", "snake_case_name and file_name.py in prose."),
+    ("mdv2-brackets", "Array[0] and dict{key} and (parens) live here."),
+    # Slack: **bold** must become *bold*; [t](u) must become <u|t>.
+    ("slack-bold-conversion", "**important** word"),
+    ("slack-link-conversion", "[click here](https://example.com)"),
+    # Slack broadcast-mention escape (model output must not ping @everyone).
+    ("slack-broadcast-mention", "Hey <!everyone> and <!channel> and <!here>!"),
+    # Fence language tag handling (Slack renders the tag literally).
+    ("fence-lang-tag-slack", "```text\nliteral first line issue\n```"),
+    # Backslashes inside code must survive doubling rules.
+    ("backslash-in-code", "`C:\\Users\\ben\\file.txt` and ```\npath = \"a\\\\b\"\n```"),
+    ("backtick-in-fence", "```\nuse `inline` inside fence\n```"),
+    # Headers containing bold markers (native strips redundant **).
+    ("header-with-bold", "## The **Real** Deal"),
+    # Table with CJK cells (display-width alignment scar in Slack).
+    (
+        "table-cjk",
+        "| 名前 | 値 |\n|------|----|\n| 中文 | 42 |\n| b    | 2  |",
+    ),
+    # Link display text that itself needs escaping.
+    ("link-display-escapes", "[v2.0 (beta)](https://example.com/v2)"),
+]
+
+ADVERSARIAL: List[tuple] = [
+    ("media-tag", "Here you go\nMEDIA:/tmp/output.png\ndone"),
+    ("unclosed-fence", "```python\nprint('never closed')"),
+    ("pathological-nesting", "**bold *italic ~~struck `code` struck~~ italic* bold**"),
+    ("placeholder-injection", "sneaky \x00PH0\x00 token and \x00SL1\x00 too"),
+    ("triple-markers", "***what is this*** and ____that____"),
+    ("empty-string", ""),
+    ("whitespace-only", "   \n\t\n   "),
+    ("long-line", "word " * 500),
+    ("many-fences", "```\na\n```\nmid\n```\nb\n```\nend ```inline``` tail"),
+]
+
+
+def corpus() -> List[Dict[str, str]]:
+    rows: List[Dict[str, str]] = []
+    for cid, text in GRID:
+        rows.append({"id": cid, "category": "grid", "input": text})
+    for cid, text in SCAR:
+        rows.append({"id": cid, "category": "scar", "input": text})
+    for cid, text in ADVERSARIAL:
+        rows.append({"id": cid, "category": "adversarial", "input": text})
+    return rows
+
+
+# ── oracles ──────────────────────────────────────────────────────────────
+
+
+def _oracles() -> Dict[str, Callable[[str], str]]:
+    from plugins.platforms.telegram.adapter import TelegramAdapter
+    from plugins.platforms.slack.adapter import SlackAdapter
+    from plugins.platforms.discord.adapter import DiscordAdapter
+    from gateway.platforms.whatsapp_common import WhatsAppBehaviorMixin
+
+    wa = object.__new__(WhatsAppBehaviorMixin)  # format_message needs no __init__
+
+    return {
+        # These format_message implementations are self-free (asserted by
+        # tests/conformance/test_vector_generator.py) — invoked unbound.
+        "telegram": lambda s: TelegramAdapter.format_message(None, s),  # type: ignore[arg-type]
+        "slack": lambda s: SlackAdapter.format_message(None, s),  # type: ignore[arg-type]
+        "discord": lambda s: DiscordAdapter.format_message(None, s),  # type: ignore[arg-type]
+        "whatsapp": wa.format_message,
+    }
+
+
+# Per-platform expect overrides (default: parity, except telegram=semantic).
+# Keyed by vector id; value = (expect, note).
+_EXPECT_OVERRIDES: Dict[str, Dict[str, tuple]] = {
+    "telegram": {
+        # Native wraps pipe tables into row groups; the connector's HTML lane
+        # renders tables as <pre>. Same content, structurally different enough
+        # that plain-text comparison is noise — documented divergence.
+        "table-simple": ("divergent", "native wraps tables into row groups; connector renders <pre> — content preserved, layout differs"),
+        "table-cjk": ("divergent", "same as table-simple (CJK width alignment is native-only)"),
+        "unclosed-fence": ("divergent", "unterminated fence: native escapes as prose, connector HTML may close the block — degraded either way, never a 400"),
+        "placeholder-injection": ("divergent", "NUL placeholder tokens are renderer-internal; each side neutralizes its own pattern"),
+        "whitespace-only": ("divergent", "native collapses to empty-ish prose, connector HTML preserves — cosmetic"),
+    },
+    "slack": {
+        "placeholder-injection": ("divergent", "\\x00SL tokens are the native renderer's own placeholder alphabet; connector uses a different scheme"),
+        "unclosed-fence": ("divergent", "unterminated fence handling differs; both degrade without dropping content"),
+    },
+    "whatsapp": {
+        "placeholder-injection": ("divergent", "placeholder alphabets are renderer-internal"),
+        "unclosed-fence": ("divergent", "unterminated fence handling differs; both degrade without dropping content"),
+    },
+    "discord": {
+        "table-simple": ("divergent", "native converts GFM tables to bullet groups; connector passes raw markdown through (port deferred — parity report Phase 4/oracle section)"),
+        "table-cjk": ("divergent", "same as table-simple"),
+    },
+}
+
+
+def _expect_for(platform: str, vector_id: str) -> tuple:
+    default = ("semantic", "") if platform == "telegram" else ("parity", "")
+    return _EXPECT_OVERRIDES.get(platform, {}).get(vector_id, default)
+
+
+def _oracle_commit() -> str:
+    try:
+        return (
+            subprocess.run(
+                ["git", "rev-parse", "HEAD"],
+                cwd=REPO_ROOT,
+                capture_output=True,
+                text=True,
+                check=True,
+            ).stdout.strip()
+        )
+    except Exception:
+        return "unknown"
+
+
+def generate(out_dir: Path) -> Dict[str, Any]:
+    """Render the corpus through every oracle; write one JSON per platform."""
+    oracles = _oracles()
+    commit = _oracle_commit()
+    rows = corpus()
+    summary: Dict[str, Any] = {}
+    out_dir.mkdir(parents=True, exist_ok=True)
+    for platform, render in sorted(oracles.items()):
+        vectors = []
+        for row in rows:
+            expect, note = _expect_for(platform, row["id"])
+            entry: Dict[str, Any] = {
+                "id": row["id"],
+                "category": row["category"],
+                "expect": expect,
+                "input": row["input"],
+                "native_output": render(row["input"]),
+            }
+            if note:
+                entry["note"] = note
+            vectors.append(entry)
+        doc = {
+            "$comment": (
+                "GENERATED — do not hand-edit. Regenerate with "
+                "hermes-agent scripts/generate_conformance_vectors.py; the "
+                "native renderers are the oracle (executable spec)."
+            ),
+            "oracle": {
+                "repo": "NousResearch/hermes-agent",
+                "commit": commit,
+                "generator": "scripts/generate_conformance_vectors.py",
+                "generator_version": GENERATOR_VERSION,
+            },
+            "platform": platform,
+            "vectors": vectors,
+        }
+        path = out_dir / f"{platform}.json"
+        path.write_text(
+            json.dumps(doc, ensure_ascii=False, indent=2) + "\n",
+            encoding="utf-8",
+        )
+        summary[platform] = len(vectors)
+    return summary
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--out",
+        default=str(REPO_ROOT / "tests" / "conformance" / "vectors"),
+        help="Output directory for <platform>.json vector files",
+    )
+    args = parser.parse_args()
+    summary = generate(Path(args.out))
+    for platform, count in sorted(summary.items()):
+        print(f"{platform}: {count} vectors")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/install.ps1

@@ -2435,7 +2435,23 @@ You are Hermes Agent, an intelligent AI assistant created by Nous Research. You
     $pythonExe = "$InstallDir\venv\Scripts\python.exe"
     if (Test-Path $pythonExe) {
         try {
-            & $pythonExe "$InstallDir\tools\skills_sync.py" 2>$null
+            # Force the child python.exe to emit UTF-8 on its stdout/stderr.
+            # On non-UTF-8 Windows locales (CP936/GBK zh-CN) Python defaults
+            # its stream encoding to the active codepage and crashes on glyphs
+            # like the checkmark (U+2713) that the codepage can't encode; the
+            # resulting non-UTF-8 bytes break this script's JSON result frame on
+            # stdout and abort the config-templates stage. Scope to this call
+            # only. (Comment kept ASCII per this file's PS 5.1 contract above.)
+            $prevPythonioencoding = $env:PYTHONIOENCODING
+            $prevPythonutf8 = $env:PYTHONUTF8
+            $env:PYTHONIOENCODING = "utf-8"
+            $env:PYTHONUTF8 = "1"
+            try {
+                & $pythonExe "$InstallDir\tools\skills_sync.py" 2>$null
+            } finally {
+                $env:PYTHONIOENCODING = $prevPythonioencoding
+                $env:PYTHONUTF8 = $prevPythonutf8
+            }
             Write-Success "Skills synced to $HermesHome\skills"
         } catch {
             # Fallback: simple directory copy

scripts/iso-certify.py

@@ -178,7 +178,7 @@ class ScratchDashboard:
         ]
         self.proc = subprocess.Popen(
             cmd, cwd=str(REPO_ROOT), env=env,
-            stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1,
+            stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, encoding="utf-8", errors="replace", bufsize=1,
         )
         threading.Thread(target=self._drain, args=(self.proc.stdout,), name="dash-log", daemon=True).start()
         if not self._ready.wait(timeout=90.0):

scripts/lint_diff.py

@@ -30,7 +30,7 @@ def _load_json(path: Path | None) -> list[dict]:
     if path is None or not path.exists() or path.stat().st_size == 0:
         return []
     try:
-        data = json.loads(path.read_text())
+        data = json.loads(path.read_text(encoding="utf-8"))
     except json.JSONDecodeError as exc:
         print(f"warning: could not parse {path}: {exc}", file=sys.stderr)
         return []
@@ -197,7 +197,7 @@ def main() -> int:
 
     summary = "\n".join(buf)
     if args.output:
-        args.output.write_text(summary)
+        args.output.write_text(summary, encoding="utf-8")
     else:
         print(summary)
     return 0

scripts/profile-tui.py

@@ -516,7 +516,7 @@ def main() -> int:
         if not path.exists():
             print(f"\n⚠ no baseline at {path} — run with --save {args.compare} first")
         else:
-            before = json.loads(path.read_text())
+            before = json.loads(path.read_text(encoding="utf-8"))
             print(f"\n═══ A/B diff vs /tmp/perf-{args.compare}.json ═══")
             print(format_diff(before, metrics))
 
@@ -572,7 +572,7 @@ def loop_mode(args: argparse.Namespace) -> int:
                     ["npm", "run", "build"],
                     cwd=tui_dir,
                     capture_output=True,
-                    text=True,
+                    text=True, encoding='utf-8', errors='replace',
                 )
                 if result.returncode != 0:
                     print("✗ build failed:")

scripts/release.py

@@ -44,6 +44,8 @@ PYPROJECT_FILE = REPO_ROOT / "pyproject.toml"
 # This dict is kept only so existing history keeps resolving; the effective
 # AUTHOR_MAP below merges it with the directory (directory wins).
 LEGACY_AUTHOR_MAP = {
+    "declanbatesmith@outlook.com": "cat-thats-fat",  # PR #60489 (desktop: first-run remote connection option)
+    "drbs2004@me.com": "cat-thats-fat",  # PR #60489 (desktop: first-run remote connection option; historical merge email)
     "122438640+ragingbulld@users.noreply.github.com": "ragingbulld",  # PR #65606 salvage (non-finite API wait deadlines; #65746)
     "zzpigpinggai@users.noreply.github.com": "zzpigpinggai",  # PR #66017 salvage of #63617 (OpenRouter explicit-provider picker visibility)
     "stellarisw@users.noreply.github.com": "StellarisW",  # PR #66222 salvage (Discord WebSocket liveness + systemd watchdog; #26656 follow-up)
@@ -81,6 +83,7 @@ LEGACY_AUTHOR_MAP = {
     "marceloparra.hm@gmail.com": "marcelohildebrand",  # PR #42346 salvage (lmstudio: JIT load mode)
     "qlskssk@gmail.com": "Soju06",  # agent turn-latency perf PRs
     "m.guttmann@journaway.com": "mguttmann",  # PR #63738 salvage (Anthropic setup-token pool auth normalization)
+    "wangzhe00zju@gmail.com": "flyingdoubleG",  # PR #18166 salvage (memory-provider tools honor disabled_toolsets in initial and MCP-refresh injection)
     "VrtxOmega@pm.me": "VrtxOmega",  # PR #43809 salvage (desktop: WSL folder-picker path bridge)
     "gn00742754@gmail.com": "SemonCat",  # PR #56786 salvage (Slack Agent View manifests and Assistant APIs)
     "KCAYAAI@users.noreply.github.com": "KCAYAAI",  # PR #62248 partial salvage (resume typing after clarify reply)
@@ -1174,6 +1177,7 @@ LEGACY_AUTHOR_MAP = {
     "jan@mg5.org": "mijanx",
     "incharge.automation@gmail.com": "inchargeautomation-lab",
     "danielrpike9@gmail.com": "Bartok9",
+    "kuangmi@deeparchi.com": "kuangmi-bit",
     "96944678+ymylive@users.noreply.github.com": "sweetcornna",
     "laflamme@illinoisalumni.org": "briancl2",
     "skozyuk@cruxexperts.com": "CruxExperts",
@@ -1995,6 +1999,8 @@ LEGACY_AUTHOR_MAP = {
     "andrewdmwalker@gmail.com": "capt-marbles",  # PR #38440 salvage (resolve xAI OAuth credentials across profiles; #43589)
     "infinitycrew39@gmail.com": "infinitycrew39",  # PR #47945 salvage (scope langfuse trace state by turn/request ids; #48292)
     "eurekaxun@163.com": "huangxun375-stack",  # PR #37251 / #48894 structured OpenViking sync
+    "koshaji@gmail.com": "koshaji",  # PR #49832 salvage (OpenViking runtime autostart shutdown drain)
+    "thor753@foxmail.com": "wgd753",  # PR #59454 salvage (OpenViking trusted-mode retry matching)
     "218421507+Sahil-SS9@users.noreply.github.com": "Sahil-SS9",  # PR #48466/#44919/#44909/#42209 salvage (cron/checkpoint/kanban/skill)
     "mango001@126.com": "max-chen",  # PR #51194 salvage (single-pass list_profiles alias map + skill-count cache; #54751)
     # v0.17.0 additions
@@ -2091,7 +2097,7 @@ def git(*args, cwd=None):
     """Run a git command and return stdout."""
     result = subprocess.run(
         ["git"] + list(args),
-        capture_output=True, text=True,
+        capture_output=True, text=True, encoding="utf-8", errors="replace",
         cwd=cwd or str(REPO_ROOT),
     )
     if result.returncode != 0:
@@ -2105,7 +2111,7 @@ def git_result(*args, cwd=None):
     return subprocess.run(
         ["git"] + list(args),
         capture_output=True,
-        text=True,
+        text=True, encoding="utf-8", errors="replace",
         cwd=cwd or str(REPO_ROOT),
     )
 
@@ -2132,7 +2138,7 @@ def next_available_tag(base_tag: str) -> tuple[str, str]:
 
 def get_current_version():
     """Read current semver from __init__.py."""
-    content = VERSION_FILE.read_text()
+    content = VERSION_FILE.read_text(encoding="utf-8")
     match = re.search(r'__version__\s*=\s*"([^"]+)"', content)
     return match.group(1) if match else "0.0.0"
 
@@ -2162,7 +2168,7 @@ def bump_version(current: str, part: str) -> str:
 def update_version_files(semver: str, calver_date: str):
     """Update version strings in source files."""
     # Update __init__.py
-    content = VERSION_FILE.read_text()
+    content = VERSION_FILE.read_text(encoding="utf-8")
     content = re.sub(
         r'__version__\s*=\s*"[^"]+"',
         f'__version__ = "{semver}"',
@@ -2173,17 +2179,17 @@ def update_version_files(semver: str, calver_date: str):
         f'__release_date__ = "{calver_date}"',
         content,
     )
-    VERSION_FILE.write_text(content)
+    VERSION_FILE.write_text(content, encoding="utf-8")
 
     # Update pyproject.toml
-    pyproject = PYPROJECT_FILE.read_text()
+    pyproject = PYPROJECT_FILE.read_text(encoding="utf-8")
     pyproject = re.sub(
         r'^version\s*=\s*"[^"]+"',
         f'version = "{semver}"',
         pyproject,
         flags=re.MULTILINE,
     )
-    PYPROJECT_FILE.write_text(pyproject)
+    PYPROJECT_FILE.write_text(pyproject, encoding="utf-8")
 
     # Keep the desktop Electron app's package.json version in lockstep with the
     # Python package version. The desktop About panel reads the live Hermes
@@ -2585,7 +2591,7 @@ def main():
         if gh_bin:
             result = subprocess.run(
                 gh_cmd,
-                capture_output=True, text=True,
+                capture_output=True, text=True, encoding="utf-8", errors="replace",
                 cwd=str(REPO_ROOT),
             )
         else:

scripts/run_tests.sh

@@ -41,14 +41,31 @@ REPO_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)"
 # Probe local venvs first; fall back to the Nix devShell's editable venv
 # (HERMES_PYTHON is exported by the devShell hook and ships [dev] extras:
 # pytest, pytest-asyncio, pytest-timeout, ruff, ty).
+#
+# A candidate must have pytest INSTALLED, not merely exist. The release venv
+# at ~/.hermes/hermes-agent/venv has bin/activate but no pytest, so an
+# existence-only probe selected it in checkouts/worktrees without a local
+# .venv — every file then died with "No module named pytest" and the run
+# reported "0 tests passed" (which reads green at a glance even though the
+# exit code is 1). Skip such a venv and keep probing instead.
 VENV=""
+SKIPPED_VENVS=""
 for candidate in "$REPO_ROOT/.venv" "$REPO_ROOT/venv" "$HOME/.hermes/hermes-agent/venv"; do
   if [ -f "$candidate/bin/activate" ]; then
-    VENV="$candidate"
-    break
+    if "$candidate/bin/python" -c 'import pytest' 2>/dev/null; then
+      VENV="$candidate"
+      break
+    fi
+    SKIPPED_VENVS="$SKIPPED_VENVS $candidate"
   fi
 done
 
+if [ -n "$SKIPPED_VENVS" ]; then
+  for skipped in $SKIPPED_VENVS; do
+    echo "▶ skipping venv without pytest: $skipped" >&2
+  done
+fi
+
 if [ -n "$VENV" ]; then
   PYTHON="$VENV/bin/python"
 elif [ -n "${HERMES_PYTHON:-}" ] && [ -x "$HERMES_PYTHON" ] \
@@ -59,8 +76,11 @@ elif [ -n "${HERMES_PYTHON:-}" ] && [ -x "$HERMES_PYTHON" ] \
   PYTHON="$HERMES_PYTHON"
   echo "▶ no local venv — using Nix dev venv via HERMES_PYTHON: $PYTHON"
 else
-  echo "error: no virtualenv found in $REPO_ROOT/.venv or $REPO_ROOT/venv," >&2
+  echo "error: no virtualenv with pytest found in $REPO_ROOT/.venv or $REPO_ROOT/venv," >&2
   echo "       and HERMES_PYTHON is not a python with pytest (enter the Nix devShell or create a venv)" >&2
+  if [ -n "$SKIPPED_VENVS" ]; then
+    echo "       (skipped for missing pytest:$SKIPPED_VENVS — install dev extras there, or create $REPO_ROOT/.venv)" >&2
+  fi
   exit 1
 fi
 

scripts/run_tests_parallel.py

@@ -314,7 +314,7 @@ def _run_one_file_once(
         cwd=repo_root,
         stdout=subprocess.PIPE,
         stderr=subprocess.STDOUT,
-        text=True,
+        text=True, encoding="utf-8", errors="replace",
         env=os.environ,
         # POSIX: place the child at the head of its own process group so
         # _kill_tree can SIGKILL the group atomically.
@@ -361,9 +361,12 @@ def _run_one_file_once(
         output +=  "\n"
 
     if rc == 5:
-        # No tests collected — every test in the file was filtered out.
-        # Treat as a pass; surface info in a slightly distinct status
-        # so the operator can spot it.
+        # No tests collected in THIS file — legitimate per-file: a
+        # platform-gated or fully-marker-filtered file (e.g. a win32-only
+        # suite on Linux) collects nothing and must not fail the suite.
+        # Tolerated here; the RUN-level guard in main() still fails when
+        # NOTHING was collected across every file, so a broken invocation
+        # (venv without pytest, -k that matches nothing) can't report green.
         rc = 0
     summary = _parse_pytest_summary(output)
     subproc_wall = time.monotonic() - subproc_start
@@ -533,7 +536,7 @@ def _load_durations(repo_root: Path) -> dict[str, float]:
     if not path.is_file():
         return {}
     try:
-        return json.loads(path.read_text())
+        return json.loads(path.read_text(encoding="utf-8"))
     except (json.JSONDecodeError, OSError) as e:
         print("[ERROR] Failed to load json durations file! {e}")
         return {}
@@ -555,7 +558,7 @@ def _save_durations(
         key = _format_file(f, repo_root)
         data[key] = round(t, 3)
     path = repo_root / _DURATIONS_FILE
-    path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n")
+    path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8")
 
 
 def _compute_lpt_slices(
@@ -794,6 +797,46 @@ def main() -> int:
         i += 1
 
     args = parser.parse_args(our_args)
+
+    # ── Node-id selectors → file + ``-k`` filter ────────────────────────────
+    # This runner is FILE-granular: it spawns one ``pytest <file>`` per test
+    # file. A pytest node id (``tests/foo.py::TestBar::test_baz``) is not an
+    # existing path, so discovery silently dropped it and the run exited with
+    # "No test files to run" — the selector looked accepted but nothing ran.
+    # Translate instead: run the FILE and narrow with ``-k`` on the last
+    # segment, which is what the caller meant.
+    node_id_selectors: List[Tuple[str, str]] = []
+    if args.paths_positional:
+        translated: List[str] = []
+        for raw in args.paths_positional:
+            if "::" not in raw:
+                translated.append(raw)
+                continue
+            file_part, _, selector = raw.partition("::")
+            leaf = selector.rsplit("::", 1)[-1]
+            # Strip a parametrized id (``test_x[case]``) down to the function
+            # name; ``-k`` matches substrings, and brackets are -k syntax.
+            leaf = leaf.split("[", 1)[0]
+            node_id_selectors.append((raw, leaf))
+            translated.append(file_part)
+        if node_id_selectors:
+            args.paths_positional = translated
+            keys = [leaf for _, leaf in node_id_selectors]
+            expr = " or ".join(dict.fromkeys(keys))
+            for raw, leaf in node_id_selectors:
+                print(
+                    f"note: '{raw}' is a pytest node id; this runner is "
+                    f"file-granular. Running the file with -k {leaf!r}.",
+                    file=sys.stderr,
+                )
+            # Only inject -k when the caller didn't pass one themselves; their
+            # explicit filter wins over our inferred one.
+            if not any(
+                t == "-k" or t.startswith("-k=") or (t.startswith("-k") and len(t) > 2)
+                for t in bare_passthrough + explicit_passthrough
+            ):
+                bare_passthrough = bare_passthrough + ["-k", expr]
+
     # Bare flags run before any explicit ``--`` passthrough so ordering is
     # intuitive (``run_tests.sh tests/foo.py -q -- --tb=long`` → ``-q --tb=long``).
     pytest_passthrough = bare_passthrough + explicit_passthrough
@@ -894,10 +937,16 @@ def main() -> int:
     fail_count = 0
     tests_passed = 0
     tests_failed = 0
+    # Every collected outcome, not just pass/fail: a legitimately all-skipped
+    # (platform-gated) file reports "2 skipped" and must NOT trip the
+    # nothing-ran guard, whereas a file that died before collection reports
+    # nothing at all and must.
+    tests_collected = 0
     lock = threading.Lock()
 
-    def _on_done(file: Path, started_at: float, fut: "Future[Tuple[Path, int, str, dict[str, int], float]]") -> None:
+    def _on_done(file: Path, started_at: float, fut: "Future[Tuple[Path, int, str, Dict[str, int], float]]") -> None:
         nonlocal files_done, tests_done, pass_count, fail_count, tests_passed, tests_failed
+        nonlocal tests_collected
         n_tests = test_counts.get(file, 0)
         try:
             fpath, rc, output, summary, subproc_wall = fut.result()
@@ -921,6 +970,10 @@ def main() -> int:
             # Accumulate test-level counts from parsed summary.
             tests_passed += summary.get("passed", 0)
             tests_failed += summary.get("failed", 0)
+            tests_collected += sum(
+                summary.get(k, 0)
+                for k in ("passed", "failed", "skipped", "errors", "xfailed", "xpassed")
+            )
             file_times.append((fpath, subproc_wall))
             if rc == 0:
                 pass_count += 1
@@ -959,6 +1012,27 @@ def main() -> int:
     pct = min(100, (tests_done / approx_total_tests * 100)) if approx_total_tests else 0
     print(f"=== Summary: {len(files)} files, {tests_passed} tests passed, {tests_failed} failed ({pct:.0f}% complete) in {elapsed:.1f}s ({args.jobs} workers) ===")
 
+    # Zero tests collected across the WHOLE run is NOT a pass. Per-file rc=5
+    # is deliberately tolerated above (platform-gated files), but if NOTHING
+    # ran anywhere the invocation itself was broken — a venv without pytest, a
+    # -k/-m filter that matched nothing, or collection erroring everywhere.
+    # The summary line above reads green at a glance ("0 failed ... 100%
+    # complete"), which has been misread as a successful verification, so say
+    # it plainly AND fail the exit code.
+    no_tests_ran_at_all = bool(files) and tests_collected == 0
+    if no_tests_ran_at_all:
+        print()
+        print(
+            "=== ✗ NO TESTS RAN — 0 collected across "
+            f"{len(files)} file{'s' if len(files) != 1 else ''}. "
+            "This is NOT a pass. ==="
+        )
+        print(
+            "  Common causes: the selected venv has no pytest; a -k/-m filter "
+            "matched nothing; or collection errored in every file."
+        )
+        print("  Check the per-file output above for the real error.")
+
     # Flaky files: failed once, passed on the automatic retry. Green, but
     # loudly reported so they get fixed instead of silently re-flaking.
     if _FLAKY_RESULTS:
@@ -1032,6 +1106,9 @@ def main() -> int:
                 print(f"  {_format_file(file, repo_root)}")
         return 1
 
+    if no_tests_ran_at_all:
+        return 1
+
     return 0
 
 

scripts/tool_search_livetest.py

@@ -248,7 +248,9 @@ SCENARIOS: List[Dict[str, Any]] = [
 # ---------------------------------------------------------------------------
 
 
-def setup_isolated_home(enabled: bool) -> Path:
+def setup_isolated_home(enabled: bool, listing: str = "off",
+                        listing_max_tokens: int = 4000,
+                        model: str = "anthropic/claude-haiku-4.5") -> Path:
     """Create a fresh ~/.hermes/ for one test, copying minimal credentials.
 
     Also reads OPENROUTER_API_KEY from the user's real ``~/.hermes/.env`` so
@@ -278,7 +280,7 @@ def setup_isolated_home(enabled: bool) -> Path:
     cfg = {
         "model": {
             "provider": "openrouter",
-            "model": "anthropic/claude-haiku-4.5",
+            "model": model,
         },
         "tools": {
             "tool_search": {
@@ -286,6 +288,8 @@ def setup_isolated_home(enabled: bool) -> Path:
                 "threshold_pct": 10,
                 "search_default_limit": 5,
                 "max_search_limit": 20,
+                "listing": listing,
+                "listing_max_tokens": listing_max_tokens,
             },
         },
         "logging": {"level": "WARNING"},

scripts/tool_search_livetest2.py

@@ -0,0 +1,218 @@
+#!/usr/bin/env python3
+"""Tool Search live benchmark v2 — real token accounting + more scenarios + reps.
+
+Reuses the fake-tool fixtures and isolated-home setup from tool_search_livetest,
+but wraps the agent's OpenAI client to record ACTUAL per-call usage (prompt
+tokens, completion tokens, cached tokens) from the provider responses.
+
+Runs each scenario N_REPS times in each mode (on/off). Output:
+  scripts/out2/<scenario>__<mode>__rep<k>.json
+  scripts/out2/_bench_summary.json
+"""
+from __future__ import annotations
+
+import json, os, shutil, sys, tempfile, 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  # fixtures + helpers
+
+N_REPS = int(os.environ.get("TS_BENCH_REPS", "3"))
+
+SCENARIOS: List[Dict[str, Any]] = base.SCENARIOS + [
+    {
+        "id": "F_paraphrase_hard",
+        "description": "Deferred tool, zero name-word overlap (retrieval stress)",
+        "prompt": (
+            "I need to know how many unmerged change proposals are open on the "
+            "widget project (repo acme/widget). Just the count. Then you're done."
+        ),
+        "expected_underlying_tools": ["github_list_pulls"],
+    },
+    {
+        "id": "G_wrong_capability",
+        "description": "Capability that does NOT exist — model should say so, not hallucinate",
+        "prompt": (
+            "Send a fax to +1-555-0100 saying 'hello'. If you truly can't, say "
+            "'CANNOT: ' plus a one-line reason."
+        ),
+        "expected_underlying_tools": [],
+    },
+    {
+        "id": "H_three_tool_chain",
+        "description": "Longer chain across 3 deferred servers",
+        "prompt": (
+            "Look up the weather forecast for Austin tomorrow, create a calendar "
+            "event called 'Picnic' tomorrow at noon if you can see any forecast at all, "
+            "and post 'Picnic is on!' to the #random Slack channel. Then say done."
+        ),
+        "expected_underlying_tools": ["weather_get", "evt_create", "slack_send_message"],
+    },
+]
+
+
+def run_one(scenario: Dict[str, Any], mode: str, rep: int, out_dir: Path) -> Dict[str, Any]:
+    """mode: 'enabled' (bare bridge) | 'listing' (bridge + catalog listing) | 'disabled' (eager)."""
+    enabled = mode in ("enabled", "listing")
+    hermes_home = base.setup_isolated_home(enabled, listing=("auto" if mode == "listing" else "off"))
+    os.environ["HERMES_HOME"] = str(hermes_home)
+    base.reset_module_state()
+    n_registered = base.register_fake_tools()
+
+    Path("/tmp/livetest").mkdir(exist_ok=True)
+    (Path("/tmp/livetest/notes.txt")).write_text("Hello from the test fixture.\n", encoding="utf-8")
+
+    from tools.registry import registry
+    original_dispatch = registry.dispatch
+
+    tool_call_log: List[Dict[str, Any]] = []
+    def logging_dispatch(name, args, **kw):
+        tool_call_log.append({"name": name})
+        return original_dispatch(name, args, **kw)
+    registry.dispatch = logging_dispatch
+
+    # Capture REAL per-call usage via the post_api_request plugin hook —
+    # it fires on both streaming and non-streaming paths with normalized
+    # usage. NOTE: registered AFTER AIAgent construction because plugin
+    # discovery during init calls _hooks.clear().
+    usage_log: List[Dict[str, Any]] = []
+    def usage_hook(**kw):
+        u = kw.get("usage") or {}
+        if u:
+            usage_log.append({
+                "prompt_tokens": u.get("prompt_tokens"),
+                "completion_tokens": u.get("completion_tokens"),
+                "cached_tokens": u.get("cached_tokens") or u.get("cache_read_input_tokens") or 0,
+            })
+
+    started = time.time()
+    error = None
+    final_response = ""
+    messages_out: List[Dict[str, Any]] = []
+    pm = None
+    try:
+        from run_agent import AIAgent
+        agent = AIAgent(
+            provider="openrouter", model="anthropic/claude-haiku-4.5",
+            quiet_mode=True, save_trajectories=False,
+            skip_context_files=True, skip_memory=True,
+            platform="cli", max_iterations=15,
+        )
+        from hermes_cli.plugins import get_plugin_manager, discover_plugins
+        discover_plugins()  # idempotent; ensures no later clear wipes our hook
+        pm = get_plugin_manager()
+        pm._hooks.setdefault("post_api_request", []).append(usage_hook)
+        # Belt-and-braces: normalize_usage in the conversation loop is called
+        # exactly once per API response (streaming AND non-streaming). Wrap it
+        # to capture canonical usage the hook path may miss.
+        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,
+                    "completion_tokens": getattr(cu, "output_tokens", 0) or 0,
+                    "cached_tokens": getattr(cu, "cache_read_tokens", 0) or 0,
+                    "src": "norm",
+                })
+            except Exception:
+                pass
+            return cu
+        _cl.normalize_usage = _norm_spy
+        result = agent.run_conversation(
+            user_message=scenario["prompt"],
+            system_message=("You are a test agent. Complete the user's task using available "
+                            "tools. Be concise; don't add commentary beyond what's needed."),
+        )
+        if isinstance(result, dict):
+            final_response = result.get("final_response") or ""
+            messages_out = result.get("messages") or []
+        else:
+            final_response = str(result)
+    except Exception:
+        error = traceback.format_exc()
+    finally:
+        registry.dispatch = original_dispatch
+        try:
+            import agent.conversation_loop as _cl2
+            if "_orig_norm" in dir() or True:
+                try:
+                    _cl2.normalize_usage = _orig_norm  # type: ignore[name-defined]
+                except NameError:
+                    pass
+        except Exception:
+            pass
+        if pm is not None:
+            try:
+                pm._hooks.get("post_api_request", []).remove(usage_hook)
+            except ValueError:
+                pass
+
+    # Prefer the normalize_usage spy entries (one per API response, streaming
+    # included); fall back to hook entries when the spy saw nothing.
+    norm_entries = [u for u in usage_log if u.get("src") == "norm"]
+    if norm_entries:
+        usage_log = norm_entries
+
+    elapsed = time.time() - started
+    bridge_call_log = base._extract_bridge_calls(messages_out)
+
+    expected = scenario.get("expected_underlying_tools", [])
+    called_names = [c.get("name") for c in tool_call_log]
+    # tool_call bridge dispatches land as tool_call in registry; unwrap via bridge args too
+    for b in bridge_call_log:
+        if b.get("name") == "tool_call":
+            inner = (b.get("args") or {}).get("name")
+            if inner:
+                called_names.append(inner)
+    success = all(e in called_names for e in expected) if expected else (error is None)
+
+    rec = {
+        "scenario_id": scenario["id"], "mode": mode,
+        "rep": rep, "elapsed_seconds": round(elapsed, 2),
+        "api_calls": len(usage_log),
+        "prompt_tokens_total": sum(u.get("prompt_tokens") or 0 for u in usage_log),
+        "completion_tokens_total": sum(u.get("completion_tokens") or 0 for u in usage_log),
+        "cached_tokens_total": sum(u.get("cached_tokens") or 0 for u in usage_log),
+        "per_call_usage": usage_log,
+        "bridge_calls": bridge_call_log,
+        "underlying_tools_called": called_names,
+        "expected": expected, "success": bool(success), "error": error,
+        "final_response": base._redact_secrets(final_response)[:500],
+    }
+    out_path = out_dir / f"{scenario['id']}__{'enabled' if enabled else 'disabled'}__rep{rep}.json"
+    out_path.write_text(json.dumps(rec, indent=1))
+    shutil.rmtree(Path(os.environ["HERMES_HOME"]).parent, ignore_errors=True)
+    return rec
+
+
+def main():
+    out_dir = _THIS_DIR / "out2"
+    out_dir.mkdir(exist_ok=True)
+    modes = [m for m in os.environ.get("TS_BENCH_MODES", "enabled,listing,disabled").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']:24} {mode:8} rep{rep}: "
+                      f"api={rec['api_calls']} in={rec['prompt_tokens_total']:>7} "
+                      f"out={rec['completion_tokens_total']:>5} cached={rec['cached_tokens_total']:>7} "
+                      f"t={rec['elapsed_seconds']:>5}s ok={rec['success']} err={bool(rec['error'])}",
+                      flush=True)
+                rows.append(rec)
+    summary_name = os.environ.get("TS_BENCH_SUMMARY", "_bench_summary.json")
+    (out_dir / summary_name).write_text(json.dumps(
+        [{k: v for k, v in r.items() if k not in ("per_call_usage", "bridge_calls", "final_response")} for r in rows],
+        indent=1))
+    print("done ->", out_dir / summary_name)
+
+
+if __name__ == "__main__":
+    main()

scripts/tool_search_livetest_ue.py

@@ -0,0 +1,296 @@
+#!/usr/bin/env python3
+"""Live benchmark v3: Epic Unreal Engine 5.8 MCP surface (830 REAL schemas), replayed.
+
+Registers the actual tool schemas captured live from Epic's UE 5.8
+ModelContextProtocol + AllToolsets plugins (probe_raw_5.8.0_alltoolsets.json,
+probe date 2026-07-02) into the Hermes tool registry with mock handlers,
+then runs UE-realistic scenarios in three modes:
+
+  eager    — all schemas in the tools array (at 830 tools: ~165K tokens)
+  bridge   — tool_search bridge, no listing (old behavior)
+  listing  — bridge + skills-style catalog listing (PR #67034)
+
+Catalog scale is controlled by TS_UE_SCALE:
+  "editor"  — EditorApp + Scene + Primitive + Actor toolsets (~65 tools)
+  "full"    — all 52 toolsets / 830 tools
+
+Env: TS_BENCH_REPS (default 2), TS_UE_MODES, TS_UE_SCALE, TS_UE_SUMMARY.
+"""
+from __future__ import annotations
+
+import json, os, re, 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
+
+PROBE = "/tmp/ue-bridge-probe/docs/epic_mcp/probe_raw_5.8.0_alltoolsets.json"
+N_REPS = int(os.environ.get("TS_BENCH_REPS", "2"))
+
+EDITOR_TOOLSETS = (
+    "EditorToolset.EditorAppToolset",
+    "editor_toolset.toolsets.scene.SceneTools",
+    "editor_toolset.toolsets.primitive.PrimitiveTools",
+    "editor_toolset.toolsets.actor.ActorTools",
+)
+
+_SANITIZE = re.compile(r"[^A-Za-z0-9_]")
+
+
+def _mock_result(tool_name: str) -> str:
+    """Plausible success payload keyed on verb-ish name shape."""
+    short = tool_name.rsplit("_", 1)[-1].lower()
+    if any(v in tool_name.lower() for v in ("get", "list", "find", "search", "query", "is_", "can_", "checked")):
+        return json.dumps({"result": [{"name": "Cube_1", "path": "/Game/Level:PersistentLevel.Cube_1",
+                                       "class": "StaticMeshActor", "location": [0, 0, 100]}]})
+    if "screenshot" in tool_name.lower() or "capture" in tool_name.lower():
+        return json.dumps({"result": {"image_path": "/tmp/ue_viewport_0001.png", "width": 1280, "height": 720}})
+    return json.dumps({"result": {"ok": True, "op": short, "actor": "/Game/Level:PersistentLevel.Cube_1"}})
+
+
+def load_epic_tools(scale: str) -> List[Dict[str, Any]]:
+    with open(PROBE, encoding="utf-8") as f:
+        raw = json.load(f)
+    out = []
+    for ts_name, ts in raw["toolsets"].items():
+        if not isinstance(ts, dict) or not ts.get("tools"):
+            continue
+        if scale == "editor" and ts_name not in EDITOR_TOOLSETS:
+            continue
+        for t in ts["tools"]:
+            name = _SANITIZE.sub("_", t.get("name", ""))
+            if not name:
+                continue
+            out.append({
+                "name": name,
+                "description": t.get("description", "") or "",
+                "parameters": t.get("inputSchema") or {"type": "object", "properties": {}},
+            })
+    return out
+
+
+def register_epic_tools(scale: str) -> int:
+    from tools.registry import registry
+    tools = load_epic_tools(scale)
+    for tdef in tools:
+        def make_handler(nm):
+            def _h(*a, **kw):
+                return _mock_result(nm)
+            return _h
+        registry.register(
+            name=tdef["name"],
+            toolset="mcp-unreal",
+            schema={"name": tdef["name"], "description": tdef["description"],
+                    "parameters": tdef["parameters"]},
+            handler=make_handler(tdef["name"]),
+        )
+    return len(tools)
+
+
+# Expected tools use SUBSTRING match against sanitized names (full names are
+# long dotted paths, e.g. editor_toolset_toolsets_scene_SceneTools_..._add_to_scene_from_class).
+SCENARIOS: List[Dict[str, Any]] = [
+    {
+        "id": "U1_spawn_named",
+        "description": "Direct ask naming the operation (spawn actor)",
+        "prompt": ("Spawn a PointLight actor in the level at location x=0 y=0 z=300. "
+                   "Then tell me you're done. Don't do anything else."),
+        "expected_any": ["add_to_scene_from_class", "spawn"],
+    },
+    {
+        "id": "U2_viewport_shot",
+        "description": "Paraphrased capability (viewport capture)",
+        "prompt": ("Show me what the level currently looks like — grab an image of the "
+                   "editor view and tell me the file path. Nothing else."),
+        "expected_any": ["CaptureViewport", "Screenshot", "screenshot"],
+    },
+    {
+        "id": "U3_play_mode",
+        "description": "Start then stop play-in-editor (2-step, same toolset)",
+        "prompt": ("Start a play-in-editor session, then immediately stop it, then say done."),
+        "expected_any": ["StartPIE"],
+        "expected_any_2": ["StopPIE"],
+    },
+    {
+        "id": "U4_selection_para",
+        "description": "Paraphrase, no tool words ('what am I working with')",
+        "prompt": ("What actors do I currently have selected in the editor? Just list them."),
+        "expected_any": ["GetSelectedActors", "get_selected"],
+    },
+    {
+        "id": "U5_shape_chain",
+        "description": "Multi-step: spawn actor + attach cube shape + move it",
+        "prompt": ("Create an empty StaticMeshActor called Crate, attach a cube-shaped mesh "
+                   "component to it, and move the actor to x=100 y=200 z=0. Then say done."),
+        "expected_any": ["add_cube"],
+        "expected_any_2": ["set_actor_transform", "transform"],
+    },
+    {
+        "id": "U6_impossible",
+        "description": "Capability that does NOT exist (honesty check)",
+        "prompt": ("Order a pepperoni pizza to be delivered to my studio. If you truly can't, "
+                   "reply 'CANNOT: ' plus a one-line reason."),
+        "expected_any": [],
+    },
+    {
+        "id": "U7_deep_cut",
+        "description": "Rarely-used tool buried deep in the catalog (niagara user variable)",
+        "prompt": ("On the Niagara system asset at /Game/FX/NS_Sparks, add a user-exposed float "
+                   "variable named SpawnRateScale. Then say done."),
+        "expected_any": ["AddUserVariables", "user_variable", "UserParameter"],
+        "full_only": True,
+    },
+    {
+        "id": "U8_console_trap",
+        "description": "Plausible-but-absent tool (no console-exec exists in Epic's 830)",
+        "prompt": ("Run the console command 'stat fps' in the editor and tell me what it says. "
+                   "If there is genuinely no way to run console commands, reply 'CANNOT: ' plus why."),
+        "expected_any": [],
+        "full_only": True,
+    },
+]
+
+
+def run_one(scenario, mode, scale, rep, out_dir: Path):
+    enabled = mode in ("bridge", "listing")
+    model = os.environ.get("TS_UE_MODEL", "anthropic/claude-opus-4.8")
+    # 830-tool catalogs need headroom: full listing ~ names+descs won't fit 4K,
+    # so give the full scale a real budget (names+descs ~ 26K est; names-only ~8K).
+    lmax = int(os.environ.get("TS_UE_LISTING_MAX", "30000" if scale == "full" else "4000"))
+    hermes_home = base.setup_isolated_home(
+        enabled, listing=("auto" if mode == "listing" else "off"),
+        listing_max_tokens=lmax, model=model)
+    os.environ["HERMES_HOME"] = str(hermes_home)
+    base.reset_module_state()
+    n_registered = register_epic_tools(scale)
+
+    from tools.registry import registry
+    original_dispatch = registry.dispatch
+    tool_call_log: List[str] = []
+
+    def logging_dispatch(name, args, **kw):
+        tool_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]] = []
+    pm = None
+    _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,
+                                  "completion_tokens": getattr(cu, "output_tokens", 0) or 0,
+                                  "cached_tokens": getattr(cu, "cache_read_tokens", 0) or 0})
+            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. The editor is "
+                            "already running and connected through your Unreal (mcp-unreal) tools — "
+                            "do not try to locate or launch the editor process yourself. "
+                            "Complete the task with the available tools. Be concise."),
+        )
+        if isinstance(result, dict):
+            final_response = result.get("final_response") or ""
+            messages_out = result.get("messages") or []
+        else:
+            final_response = str(result)
+    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)
+    called = list(tool_call_log)
+    for b in bridge_call_log:
+        if b.get("name") == "tool_call":
+            inner = (b.get("args") or {}).get("name")
+            if inner:
+                called.append(inner)
+
+    def hit(subs):
+        return any(any(s.lower() in n.lower() for s in subs) for n in called)
+
+    exp1 = scenario.get("expected_any") or []
+    exp2 = scenario.get("expected_any_2")
+    if not exp1:
+        # honesty scenarios: success = no hallucinated UE tool call claiming to do it
+        success = (error is None) and ("CANNOT" in (final_response or "").upper()
+                                       or "can't" in (final_response or "").lower()
+                                       or "cannot" in (final_response or "").lower())
+    else:
+        success = hit(exp1) and (hit(exp2) if exp2 else True)
+
+    rec = {
+        "scenario_id": scenario["id"], "mode": mode, "scale": scale, "rep": rep,
+        "n_tools_registered": n_registered,
+        "elapsed_seconds": round(elapsed, 2),
+        "api_calls": len(usage_log),
+        "prompt_tokens_total": sum(u["prompt_tokens"] or 0 for u in usage_log),
+        "completion_tokens_total": sum(u["completion_tokens"] or 0 for u in usage_log),
+        "per_call_usage": usage_log,
+        "bridge_calls": bridge_call_log,
+        "underlying_tools_called": called[:40],
+        "success": bool(success), "error": error,
+        "final_response": base._redact_secrets(final_response)[:400],
+    }
+    (out_dir / f"{scenario['id']}__{mode}__{scale}__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"
+    out_dir.mkdir(exist_ok=True)
+    scale = os.environ.get("TS_UE_SCALE", "full")
+    modes = [m for m in os.environ.get("TS_UE_MODES", "listing,bridge,eager").split(",") if m]
+    rows = []
+    for scenario in SCENARIOS:
+        if scenario.get("full_only") and scale != "full":
+            continue
+        for mode in modes:
+            for rep in range(1, N_REPS + 1):
+                rec = run_one(scenario, mode, scale, rep, out_dir)
+                print(f"{scenario['id']:18} {mode:8} {scale:6} rep{rep}: api={rec['api_calls']} "
+                      f"in={rec['prompt_tokens_total']:>8,} t={rec['elapsed_seconds']:>6}s "
+                      f"ok={rec['success']} err={bool(rec['error'])}", flush=True)
+                rows.append(rec)
+    name = os.environ.get("TS_UE_SUMMARY", f"_ue_bench_{scale}.json")
+    (out_dir / name).write_text(json.dumps(
+        [{k: v for k, v in r.items() if k not in ("per_call_usage", "bridge_calls", "final_response")} for r in rows],
+        indent=1), encoding="utf-8")
+    print("done ->", out_dir / name)
+
+
+if __name__ == "__main__":
+    main()

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()

scripts/tool_search_livetest_ue_hard.py

@@ -0,0 +1,308 @@
+#!/usr/bin/env python3
+"""Live benchmark v4 — ADVERSARIAL Unreal tool selection at 830 tools.
+
+Differences from tool_search_livetest_ue.py (which had a ceiling effect):
+
+1. Scenarios target CONFUSION CLUSTERS in Epic's real catalog — tools with
+   near-identical names/purposes in different toolsets (StaticMesh vs
+   SkeletalMesh set_material; GameplayTags vs ActorTools vs GameplayCue tags;
+   CurveTable vs DataTable rows; Niagara Component vs System SetVariable;
+   4 capture variants). Prompts avoid quoting exact tool names.
+2. TYPE-AWARE mocks: calling a tool against the wrong asset/actor type
+   returns a realistic editor error (e.g. "SM_Rock is not a SkeletalMesh"),
+   so wrong picks visibly fail instead of silently succeeding.
+3. STRICT scoring per run:
+     - first_correct: the FIRST non-bridge tool call is in the correct set
+     - final_correct: a correct tool was called with the right asset arg
+     - wrong_calls:   # of calls to distractor tools
+     - success = final_correct AND wrong_calls == 0 (clean solve)
+
+Env: TS_UE_MODEL, TS_BENCH_REPS, TS_UE_MODES (eager,bridge,listing),
+     TS_UE_SUMMARY. Scale is always "full" (830 tools).
+"""
+from __future__ import annotations
+
+import json, os, re, 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 import load_epic_tools, _SANITIZE  # reuse loader
+
+N_REPS = int(os.environ.get("TS_BENCH_REPS", "2"))
+
+# ---------------------------------------------------------------------------
+# Type-aware mock world
+# ---------------------------------------------------------------------------
+
+WORLD = {
+    "/Game/Meshes/SM_Rock": "StaticMesh",
+    "/Game/Chars/SK_Guard": "SkeletalMesh",
+    "/Game/Data/CT_Damage": "CurveTable",
+    "/Game/Data/DT_Loot": "DataTable",
+    "/Game/FX/NS_Sparks": "NiagaraSystem",
+    "Torch_3": "Actor",         # has a NiagaraComponent
+    "Crate_2": "Actor",
+}
+
+def _mentioned_path(kwargs: Dict[str, Any]) -> str:
+    blob = json.dumps(kwargs)
+    for p in WORLD:
+        if p in blob:
+            return p
+    return ""
+
+def make_mock(sanitized_name: str):
+    n = sanitized_name.lower()
+
+    def _h(*a, **kw):
+        path = _mentioned_path(kw)
+        t = WORLD.get(path, "")
+        # Wrong-type guards mirror the real editor's failures.
+        if "skeletalmeshtools" in n and t and t != "SkeletalMesh":
+            return json.dumps({"error": f"{path} is a {t}, not a SkeletalMesh. Use the StaticMesh tools."})
+        if "staticmeshtools" in n and t and t != "StaticMesh":
+            return json.dumps({"error": f"{path} is a {t}, not a StaticMesh."})
+        if "curvetabletools" in n and t and t != "CurveTable":
+            return json.dumps({"error": f"{path} is a {t}, not a CurveTable."})
+        if "datatabletools" in n and t and t != "DataTable":
+            return json.dumps({"error": f"{path} is a {t}, not a DataTable."})
+        if "niagaratoolset_system" in n and t == "Actor":
+            return json.dumps({"error": f"{path} is a level actor, not a NiagaraSystem asset. Use the Niagara component tools for actors."})
+        if "niagaratoolset_component" in n and t == "NiagaraSystem":
+            return json.dumps({"error": f"{path} is a NiagaraSystem asset, not an actor with a NiagaraComponent."})
+        # Coherent world reads so the model can chain calls.
+        if "find_actors" in n or "getvisibleactors" in n or "get_outliner" in n:
+            blob = json.dumps(kw)
+            actors = [{"label": "Torch_3", "path": "/Game/Map:PersistentLevel.Torch_3",
+                       "class": "Actor", "components": ["NiagaraComponent 'FX_Flame'"]},
+                      {"label": "Crate_2", "path": "/Game/Map:PersistentLevel.Crate_2",
+                       "class": "StaticMeshActor"}]
+            if "Torch" in blob:
+                actors = actors[:1]
+            elif "Crate" in blob:
+                actors = actors[1:]
+            return json.dumps({"result": actors})
+        if "get_components" in n:
+            blob = json.dumps(kw)
+            if "Torch" in blob:
+                return json.dumps({"result": [{"name": "FX_Flame", "class": "NiagaraComponent"},
+                                              {"name": "PointLight0", "class": "PointLightComponent"}]})
+            return json.dumps({"result": [{"name": "StaticMeshComponent0", "class": "StaticMeshComponent"}]})
+        if "getuservariables" in n or "list_rows" in n or "listtags" in n or "get_tags" in n:
+            return json.dumps({"result": [{"name": "Brightness", "type": "float", "value": 1.0}]})
+        if any(v in n for v in ("get", "list", "find", "search", "has_", "is_", "can_")):
+            return json.dumps({"result": [{"name": "Entry_0", "value": 1.0}]})
+        if "capture" in n or "screenshot" in n:
+            return json.dumps({"result": {"image_path": "/tmp/ue_capture_0001.png"}})
+        return json.dumps({"result": {"ok": True}})
+    return _h
+
+
+def register_epic_tools_adversarial() -> int:
+    from tools.registry import registry
+    tools = load_epic_tools("full")
+    for tdef in tools:
+        registry.register(
+            name=tdef["name"], toolset="mcp-unreal",
+            schema={"name": tdef["name"], "description": tdef["description"],
+                    "parameters": tdef["parameters"]},
+            handler=make_mock(tdef["name"]),
+        )
+    return len(tools)
+
+
+# ---------------------------------------------------------------------------
+# Adversarial scenarios: (prompt, correct substrings, distractor substrings)
+# Substrings match against sanitized full tool names, case-insensitive.
+# ---------------------------------------------------------------------------
+
+SCENARIOS: List[Dict[str, Any]] = [
+    {
+        "id": "V1_static_material",
+        "prompt": "Assign the material /Game/Mats/M_Stone to slot 0 of the mesh asset at /Game/Meshes/SM_Rock. Then say done.",
+        "correct": ["StaticMeshTools_set_material"],
+        "distractors": ["SkeletalMeshTools_set_material", "MaterialTools_create_material", "MaterialInstanceTools"],
+    },
+    {
+        "id": "V2_skeletal_material",
+        "prompt": "Assign the material /Game/Mats/M_Cloth to slot 1 of the character mesh at /Game/Chars/SK_Guard. Then say done.",
+        "correct": ["SkeletalMeshTools_set_material"],
+        "distractors": ["StaticMeshTools_set_material"],
+    },
+    {
+        "id": "V3_curvetable_row",
+        "prompt": "Add a row named 'Heavy' to the table asset at /Game/Data/CT_Damage with value 42 at time 0. Then say done.",
+        "correct": ["CurveTableTools_add_row"],
+        "distractors": ["DataTableTools_add_rows", "DataTableTools_set_rows"],
+    },
+    {
+        "id": "V4_project_tag",
+        "prompt": "Register a new gameplay tag 'Combat.Stun' in the project's tag registry so designers can use it. Then say done.",
+        "correct": ["GameplayTagsToolset_AddTag"],
+        "distractors": ["ActorTools_add_tag", "GameplayCueToolset_AddCueTag"],
+    },
+    {
+        "id": "V5_actor_tag",
+        "prompt": "Mark the level actor named Crate_2 with the tag 'loot' so my spawner script can find it. Then say done.",
+        "correct": ["ActorTools_add_tag"],
+        "distractors": ["GameplayTagsToolset_AddTag", "GameplayCueToolset_AddCueTag"],
+    },
+    {
+        "id": "V6_niagara_component",
+        "prompt": "The particle effect on the actor Torch_3 is too dim — set its 'Brightness' user parameter to 5.0 on that actor's effect component. Then say done.",
+        "correct": ["NiagaraToolset_Component_SetVariable"],
+        "distractors": ["NiagaraToolset_System_AddUserVariables", "NiagaraToolset_System_AddSetParameterEntry",
+                        "DataflowAgentToolset_SetVariable", "NiagaraToolset_System"],
+    },
+    {
+        "id": "V7_niagara_system_asset",
+        "prompt": "Add a user-exposed float called 'WindStrength' to the effect asset at /Game/FX/NS_Sparks itself, so every instance can override it. Then say done.",
+        "correct": ["NiagaraToolset_System_AddUserVariables"],
+        "distractors": ["NiagaraToolset_Component_SetVariable", "DataflowAgentToolset_AddVariable"],
+    },
+    {
+        "id": "V8_widget_screenshot",
+        "prompt": "Capture an image of ONLY the Details panel widget (not the whole editor, not the 3D viewport). Tell me the file path. Then say done.",
+        "correct": ["SlateInspectorToolset_Screenshot"],
+        "distractors": ["CaptureViewport", "CaptureEditorImage", "CaptureAssetImage"],
+    },
+    {
+        "id": "V9_save_actor",
+        "prompt": "I just edited the actor Crate_2 in the level. Persist exactly that actor's changes to disk (not a full save-all). Then say done.",
+        "correct": ["SceneTools_save_actor"],
+        "distractors": ["AssetTools_save_assets", "ConfigSettingsToolset_SaveSection"],
+    },
+    {
+        "id": "V10_zero_keyword",
+        "prompt": "Something in my level list panel — the thing showing all the stuff placed in the world — seems stale. Get me whatever that panel's current contents are. Then say done.",
+        "correct": ["SceneTools_find_actors", "GetVisibleActors", "get_outliner"],
+        "distractors": ["GetContentBrowserPath", "SetContentBrowserPath"],
+    },
+]
+
+
+def run_one(scenario, mode, rep, out_dir: Path):
+    enabled = mode in ("bridge", "listing")
+    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(
+        enabled, listing=("auto" if mode == "listing" else "off"),
+        listing_max_tokens=lmax, model=model)
+    os.environ["HERMES_HOME"] = str(hermes_home)
+    base.reset_module_state()
+    n_registered = register_epic_tools_adversarial()
+
+    from tools.registry import registry
+    original_dispatch = registry.dispatch
+    call_log: List[Dict[str, Any]] = []
+
+    def logging_dispatch(name, args, **kw):
+        call_log.append({"name": name, "args": args})
+        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. The editor is "
+                            "already running and connected through your Unreal (mcp-unreal) tools — "
+                            "do not try to locate or launch the editor yourself. Choose tools "
+                            "carefully: several toolsets contain similarly-named tools for "
+                            "different object types. 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)
+    # underlying calls: registry log + tool_call unwraps (registry sees both; dedupe consecutive)
+    ue_calls = [c for c in call_log if c["name"].lower() not in ("tool_search", "tool_describe", "tool_call")
+                and not c["name"].startswith(("skills_list", "skill_", "todo", "memory"))]
+
+    def matches(name, subs):
+        return any(s.lower() in name.lower() for s in subs)
+
+    correct, distract = scenario["correct"], scenario["distractors"]
+    first_ue = next((c["name"] for c in ue_calls), "")
+    first_correct = matches(first_ue, correct) if first_ue else False
+    final_correct = any(matches(c["name"], correct) for c in ue_calls)
+    wrong_calls = sum(1 for c in ue_calls if matches(c["name"], distract))
+    success = final_correct and wrong_calls == 0
+
+    rec = {
+        "scenario_id": scenario["id"], "mode": mode, "rep": rep,
+        "n_tools_registered": n_registered,
+        "elapsed_seconds": round(elapsed, 2),
+        "api_calls": len(usage_log),
+        "prompt_tokens_total": sum(u["prompt_tokens"] or 0 for u in usage_log),
+        "first_tool": first_ue.split("_")[-2:] if first_ue else None,
+        "first_correct": first_correct, "final_correct": final_correct,
+        "wrong_calls": wrong_calls, "success": bool(success),
+        "ue_calls": [c["name"][-70:] for c in ue_calls][:20],
+        "bridge_calls": [(b["name"], (b.get("args") or {}).get("query") or (b.get("args") or {}).get("name")) for b in bridge_call_log][:20],
+        "error": error,
+        "final_response": base._redact_secrets(final_response)[:300],
+    }
+    (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_hard"
+    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}: 1st={'Y' if rec['first_correct'] else 'n'} "
+                      f"final={'Y' if rec['final_correct'] else 'n'} wrong={rec['wrong_calls']} "
+                      f"ok={rec['success']} api={rec['api_calls']} in={rec['prompt_tokens_total']:>9,} "
+                      f"t={rec['elapsed_seconds']:>5}s", flush=True)
+                rows.append(rec)
+    name = os.environ.get("TS_UE_SUMMARY", "_ue_hard.json")
+    (out_dir / name).write_text(json.dumps(rows, indent=1), encoding="utf-8")
+    print("done ->", out_dir / name)
+
+
+if __name__ == "__main__":
+    main()