#!/usr/bin/env python3 """Local headless test harness for MCP for Unity. A single, stdlib-only, cross-platform Python orchestrator that, from one command, boots a headless Hub-licensed Unity Editor, reuses the resident route-B stdio bridge, runs up to three test legs (bridge smoke + Unity EditMode + Unity PlayMode over the Unity Test Framework wire protocol), aggregates a JUnit report, and tears down. The same entry point is callable from CI so ``.github/workflows/e2e-bridge.yml`` can collapse its boot/wait/discover/run-smoke shell into one invocation. Design: ``docs/superpowers/specs/2026-06-07-local-headless-test-harness-design.md``. The pure (filesystem/Unity-free) helpers sit at the top of this module and are importable + unit-testable without Unity. The reused server modules (``transport.legacy.unity_connection`` / ``transport.legacy.port_discovery``) are imported INSIDE the live functions (after prepending ``Server/src`` to ``sys.path``, mirroring ``Server/tests/e2e/bridge_smoke.py``), so importing this module never requires ``Server/src`` on ``sys.path``. Exit-code contract: 0 All blocking legs passed. PlayMode non-blocking failures still 0 (surfaced in the report). 1 A blocking leg failed = real regression (smoke returncode 1; OR EditMode status=="failed"; OR PlayMode failure under --strict-playmode). 2 Bridge unreachable / setup failure (smoke returncode 2; OR editor PID/container died during wait with no license/compile signal; OR the overall watchdog timed out). 3 Project does not compile (read_console compile probe OR compile_fatal log-grep). 4 No Unity license / Hub seat (license_fatal log-grep after warm-up grace). 5 Editor binary/version not found (discovery layer, before any boot; lists every searched path). Run locally (against TestProjects/UnityMCPTests, default isolated tmp status dir):: python tools/local_harness.py --legs smoke,editmode,playmode \ --project-path TestProjects/UnityMCPTests # Attach to an already-resident bridge instead of booting one: python tools/local_harness.py --reuse --legs smoke \ --project-path TestProjects/UnityMCPTests # Point at an arbitrary consumer project with an explicit editor binary: python tools/local_harness.py --editor /path/to/Unity \ --project-path ~/TestbedMCP --legs smoke,editmode # CI parity (DockerLauncher; license threaded as an opaque editor arg): python tools/local_harness.py --ci --no-warmup \ --legs smoke,editmode,playmode \ --project-path TestProjects/UnityMCPTests --status-dir .unity-mcp \ --editor-arg -manualLicenseFile --editor-arg /root/.../Unity_lic.ulf """ from __future__ import annotations import argparse import glob import json import os import re import shutil import signal import socket import subprocess import sys import tempfile import threading import time import xml.etree.ElementTree as ET from dataclasses import dataclass, field from pathlib import Path from typing import Any, Callable # --------------------------------------------------------------------------- # Repo geometry. This file lives at /tools/local_harness.py, so the repo # root is parents[1] and Server/src is /Server/src. We compute these once # but only mutate sys.path lazily inside the live functions (see _ensure_src_on_path). # --------------------------------------------------------------------------- _THIS = Path(__file__).resolve() REPO_ROOT = _THIS.parents[1] SERVER_SRC = REPO_ROOT / "Server" / "src" DEFAULT_PROJECT_PATH = "TestProjects/UnityMCPTests" BOOT_METHOD = "MCPForUnity.Editor.McpCiBoot.StartStdioForCi" # Socket-release delay before any resident re-launch, matching StdioBridgeHost's # own toWait.Wait(2000) for the #688/#692/#1173 Windows TcpListener leak fix. SOCKET_RELEASE_MS = 2000 # License-fatal grace: tolerate transient Licensing chatter for this long after # boot (matches claude-nl-suite.yml fatal_after = boot_start + 120 s). LICENSE_GRACE_S = 120 # PlayMode init-timeout (ms). C# default is 15000; 120000 is only a docstring # recommendation, so the harness passes it explicitly. DEFAULT_PLAYMODE_INIT_TIMEOUT_MS = 120000 # =========================================================================== # Dataclasses / exceptions (Unity-free) # =========================================================================== @dataclass class EditorNotFound(Exception): """Raised when no Unity editor binary can be resolved. Carries the full list of absolute paths probed so main() can print a remediation hint and exit 5. """ searched: list[str] = field(default_factory=list) def __str__(self) -> str: # pragma: no cover - trivial return "No Unity editor binary found. Searched:\n " + "\n ".join(self.searched) @dataclass class EditorSpec: """A resolved editor binary plus the Unity version it satisfies.""" binary: str version: str @dataclass class ReadyInfo: """Result of bridge discovery: the port, the instance id, and the status file.""" port: int instance_id: str status_file: str @dataclass class JUnitCase: name: str time_s: float = 0.0 failure: str | None = None skipped: bool = False @dataclass class JUnitSuite: name: str cases: list[JUnitCase] = field(default_factory=list) @dataclass class LegOutcome: """The result of running one leg. status is one of: "pass", "fail", "skip", "error". blocking marks whether a failure here can raise the top-level exit code. exit_code carries a specific severity (1/2/3/4) a failing leg maps to. """ name: str status: str blocking: bool detail: str = "" exit_code: int = 0 junit_suite: JUnitSuite | None = None # =========================================================================== # Pure helpers: version resolution # =========================================================================== def read_project_version(project_path: str | Path) -> str | None: """Parse

/ProjectSettings/ProjectVersion.txt for the m_EditorVersion line. Returns the trimmed token after the colon, or None if the file/line is absent or malformed (BOM, garbled). Tolerant: never raises. """ try: pv = Path(project_path) / "ProjectSettings" / "ProjectVersion.txt" text = pv.read_text(encoding="utf-8-sig", errors="replace") except OSError: return None for raw in text.splitlines(): line = raw.lstrip("").strip() if line.startswith("m_EditorVersion:"): token = line.split(":", 1)[1].strip() return token or None return None def read_default_version(versions_json: str | Path | None = None) -> str | None: """Read defaultVersion from tools/unity-versions.json. Tolerant: returns None rather than raising on a missing/garbled file.""" path = Path(versions_json) if versions_json else (REPO_ROOT / "tools" / "unity-versions.json") try: data = json.loads(Path(path).read_text(encoding="utf-8")) except (OSError, ValueError): return None val = data.get("defaultVersion") return val if isinstance(val, str) and val else None def resolve_version(project_path: str | Path, versions_json: str | Path | None = None) -> str | None: """Version precedence: ProjectVersion.txt, else unity-versions.json defaultVersion.""" return read_project_version(project_path) or read_default_version(versions_json) def parse_version(version: str) -> tuple[int, int, int, str]: """Parse '6000.0.75f1' -> (6000, 0, 75, 'f1'). Tolerant of partial versions; missing numeric parts default to 0 and a missing suffix to "". Used for comparison and nearest-patch selection. """ m = re.match(r"^\s*(\d+)(?:\.(\d+))?(?:\.(\d+))?\s*([A-Za-z]\w*)?", version or "") if not m: return (0, 0, 0, "") major = int(m.group(1)) minor = int(m.group(2)) if m.group(2) else 0 patch = int(m.group(3)) if m.group(3) else 0 suffix = m.group(4) or "" return (major, minor, patch, suffix) # =========================================================================== # Pure helpers: editor binary discovery # =========================================================================== def editor_relpath(platform: str | None = None) -> str: """Per-OS path of the Unity binary relative to a Hub/ directory.""" plat = platform or sys.platform if plat == "darwin": return "Unity.app/Contents/MacOS/Unity" if plat.startswith("win"): return "Editor/Unity.exe" return "Editor/Unity" # linux + everything else def hub_roots(platform: str | None = None, environ: dict[str, str] | None = None) -> list[str]: """Per-OS Hub Editor install roots (the directory that holds / dirs). Mirrors check-unity-versions.{sh,ps1}. On Windows, both ProgramFiles roots are iterated. """ plat = platform or sys.platform env = environ if environ is not None else os.environ home = env.get("HOME") or env.get("USERPROFILE") or str(Path.home()) if plat == "darwin": return ["/Applications/Unity/Hub/Editor"] if plat.startswith("win"): roots: list[str] = [] for var in ("ProgramFiles", "ProgramFiles(x86)"): base = env.get(var) if base: roots.append(str(Path(base) / "Unity" / "Hub" / "Editor")) if not roots: roots.append(r"C:\Program Files\Unity\Hub\Editor") return roots # linux + everything else return [str(Path(home) / "Unity" / "Hub" / "Editor")] def read_secondary_install_path(platform: str | None = None, environ: dict[str, str] | None = None, read_text: Callable[[str], str] | None = None) -> str | None: """Read Hub's secondaryInstallPath JSON-encoded string. macOS: ~/Library/Application Support/UnityHub/secondaryInstallPath.json Windows: %APPDATA%/UnityHub/secondaryInstallPath.json Linux: ~/.config/UnityHub/secondaryInstallPath.json Tolerates a missing file, "" or a non-string payload (returns None). """ plat = platform or sys.platform env = environ if environ is not None else os.environ home = env.get("HOME") or env.get("USERPROFILE") or str(Path.home()) if plat == "darwin": cfg = Path(home) / "Library" / "Application Support" / "UnityHub" / "secondaryInstallPath.json" elif plat.startswith("win"): appdata = env.get("APPDATA") or str(Path(home) / "AppData" / "Roaming") cfg = Path(appdata) / "UnityHub" / "secondaryInstallPath.json" else: xdg = env.get("XDG_CONFIG_HOME") or str(Path(home) / ".config") cfg = Path(xdg) / "UnityHub" / "secondaryInstallPath.json" reader = read_text or (lambda p: Path(p).read_text(encoding="utf-8")) try: raw = reader(str(cfg)) except OSError: return None try: val = json.loads(raw) except ValueError: # The file may store a bare path string without JSON quoting. val = raw.strip() if isinstance(val, str) and val.strip(): return val.strip() return None def candidate_editor_paths(version: str, explicit_editor: str | None = None, platform: str | None = None, environ: dict[str, str] | None = None, read_text: Callable[[str], str] | None = None) -> list[str]: """Ordered candidate editor binary paths for a resolved version. Precedence: 1. --editor (explicit_editor) 2. $UNITY_EDITOR 3. Per-OS Hub layout for 4. Hub secondaryInstallPath root, searched for Nearest-patch fallback (precedence 5) is NOT enumerated here — it is handled by discover_editor(), which needs to inspect the filesystem. """ plat = platform or sys.platform env = environ if environ is not None else os.environ relpath = editor_relpath(plat) out: list[str] = [] if explicit_editor: out.append(explicit_editor) env_editor = env.get("UNITY_EDITOR") if env_editor: out.append(env_editor) for root in hub_roots(plat, env): out.append(str(Path(root) / version / relpath)) sec = read_secondary_install_path(plat, env, read_text) if sec: out.append(str(Path(sec) / version / relpath)) return out def _default_exists(path: str) -> bool: return os.path.exists(path) def _default_is_exec(path: str) -> bool: # On Windows the +x bit is meaningless; existence is enough. if sys.platform.startswith("win"): return os.path.isfile(path) return os.path.isfile(path) and os.access(path, os.X_OK) def discover_editor(version: str, explicit_editor: str | None = None, platform: str | None = None, environ: dict[str, str] | None = None, exists: Callable[[str], bool] | None = None, is_exec: Callable[[str], bool] | None = None, list_dir: Callable[[str], list[str]] | None = None, read_text: Callable[[str], str] | None = None) -> EditorSpec: """Resolve a Unity editor binary for the requested version. First existing + executable candidate wins (precedence 1-4 from candidate_editor_paths). If none match, fall back to the nearest patch *restricted to the same major.minor*: enumerate installed dirs under every Hub/secondary root, keep only those whose (major, minor) equals the target's, pick the max by (patch, suffix). Never cross major.minor. Filesystem access is injectable (exists / is_exec / list_dir / read_text) so the whole function is hermetically unit-testable. Raises EditorNotFound(searched=[...]) carrying every absolute path probed. """ plat = platform or sys.platform env = environ if environ is not None else os.environ _exists = exists or _default_exists _is_exec = is_exec or _default_is_exec _listdir = list_dir or (lambda d: os.listdir(d) if os.path.isdir(d) else []) relpath = editor_relpath(plat) searched: list[str] = [] # Precedence 1-4: direct candidates. for cand in candidate_editor_paths(version, explicit_editor, plat, env, read_text): searched.append(cand) if _exists(cand) and _is_exec(cand): return EditorSpec(binary=cand, version=version) # Precedence 5: nearest-patch fallback, same major.minor only. target = parse_version(version) roots = list(hub_roots(plat, env)) sec = read_secondary_install_path(plat, env, read_text) if sec: roots.append(sec) best: tuple[tuple[int, str], str, str] | None = None # ((patch, suffix), binary, dir_version) for root in roots: try: entries = _listdir(root) except OSError: continue for name in entries: pv = parse_version(name) if (pv[0], pv[1]) != (target[0], target[1]): continue binary = str(Path(root) / name / relpath) searched.append(binary) if not (_exists(binary) and _is_exec(binary)): continue key = (pv[2], pv[3]) if best is None or key > best[0]: best = (key, binary, name) if best is not None: return EditorSpec(binary=best[1], version=best[2]) raise EditorNotFound(searched=searched) # =========================================================================== # Pure helpers: status-file discovery # =========================================================================== def newest_status_file(status_dir: str | Path, glob_fn: Callable[[str], list[str]] | None = None, mtime_fn: Callable[[str], float] | None = None) -> str | None: """Return the path to the newest unity-mcp-status-*.json under status_dir, or None.""" pattern = str(Path(status_dir) / "unity-mcp-status-*.json") g = glob_fn or glob.glob files = list(g(pattern)) if not files: return None m = mtime_fn or (lambda p: os.path.getmtime(p)) try: return max(files, key=m) except OSError: # If stat races a deletion, fall back to lexical newest. return sorted(files)[-1] def instance_id_from_status(status_file: str | Path, data: dict[str, Any] | None = None) -> str: """Derive the @ instance id from a status file path + payload. hash = the filename segment between 'unity-mcp-status-' and '.json' (= sha1(Application.dataPath)[:8]). Glob the newest file rather than recomputing the hash so the harness is robust to /Users vs /private/var symlink canonicalization. name = project root folder name, derived from the status payload's project_path (strip trailing /Assets) when available, else project_name, else "Unknown" — mirroring the C# project_name derivation. """ fname = os.path.basename(str(status_file)) h = fname if h.startswith("unity-mcp-status-"): h = h[len("unity-mcp-status-"):] if h.endswith(".json"): h = h[: -len(".json")] name = "Unknown" if data: project_path = data.get("project_path") if isinstance(project_path, str) and project_path: p = project_path.rstrip("/\\") if p.lower().endswith("assets"): p = p[:-6].rstrip("/\\") base = os.path.basename(p) if base: name = base if name == "Unknown": pn = data.get("project_name") if isinstance(pn, str) and pn: name = pn return f"{name}@{h}" def port_from_status(data: dict[str, Any] | None) -> int | None: """Extract unity_port from a status payload, or None.""" if not isinstance(data, dict): return None port = data.get("unity_port") return port if isinstance(port, int) else None # =========================================================================== # Pure helpers: log classification + redaction # =========================================================================== _LICENSE_RE = re.compile( r"No valid Unity" r"|License is not active" r"|cannot load ULF" r"|Signature element not found" r"|Token not found" r"|0 entitlement" r"|Entitlement.*(failed|denied)" r"|License (activation|return|renewal).*(failed|expired|denied)", re.IGNORECASE, ) _COMPILE_RE = re.compile( r"error CS\d|Scripts have compiler errors|Compilation failed", re.IGNORECASE, ) _READY_RE = re.compile( r"(Bridge|MCP(For)?Unity|AutoConnect).*(listening|ready|started|port|bound)", re.IGNORECASE, ) _REDACT_RE = re.compile(r"(?i)((email|serial|license|password|token)\S*)") _CS_ERROR_RE = re.compile(r"error CS\d") def redact(text: str) -> str: """Redact secret-ish tokens from log echoes. Mirrors the CI sed idiom `sed -E 's/((email|serial|license|password|token)[^[:space:]]*)/[REDACTED]/Ig'`. """ return _REDACT_RE.sub("[REDACTED]", text or "") def classify_log(text: str, license_grace_elapsed: bool = True) -> str: """Classify an editor log tail. Returns one of: "license_fatal", "compile_fatal", "ready_ok", "none". Order: license > compile > ready > none. The license gate is suppressed until license_grace_elapsed is True (the caller passes whether boot_start + LICENSE_GRACE_S has passed) to tolerate transient Licensing::... chatter during warm-up. """ if not text: return "none" if license_grace_elapsed and _LICENSE_RE.search(text): return "license_fatal" if _COMPILE_RE.search(text): return "compile_fatal" if _READY_RE.search(text): return "ready_ok" return "none" # =========================================================================== # Pure helpers: bridge_smoke-shaped response handling (mirrors bridge_smoke.py) # =========================================================================== def _ok(resp: Any) -> bool: """True when Unity reports success, tolerant of both response shapes.""" if not isinstance(resp, dict): return False return resp.get("success") is True or resp.get("status") == "success" def _result(resp: Any) -> dict: """Extract the payload dict from either response shape.""" if isinstance(resp, dict): inner = resp.get("result") if isinstance(inner, dict): return inner if isinstance(inner, str): return {"message": inner} return resp return {} def _message(resp: Any) -> str: """Best-effort human-readable message from a response.""" if not isinstance(resp, dict): return repr(resp) return ( str(resp.get("message")) if resp.get("message") else str(resp.get("error")) if resp.get("error") else str(_result(resp).get("message")) if _result(resp).get("message") else "" ) def _dig(resp: Any, key: str) -> Any: """Return the first value found for ``key`` anywhere in a nested response. Tolerates the transport wrapping the C# payload under result/data so contract checks depend on the field, not the exact envelope shape. """ stack: list[Any] = [resp] while stack: cur = stack.pop() if isinstance(cur, dict): if key in cur: return cur[key] stack.extend(cur.values()) elif isinstance(cur, list): stack.extend(cur) return None # =========================================================================== # Pure helpers: JUnit merge + exit aggregation # =========================================================================== def merge_junit(suites: list[JUnitSuite]) -> ET.ElementTree: """Merge JUnitSuites into a single ElementTree.""" root = ET.Element("testsuites") total_tests = total_failures = total_skipped = 0 total_time = 0.0 for suite in suites: if suite is None: continue s_failures = sum(1 for c in suite.cases if c.failure is not None) s_skipped = sum(1 for c in suite.cases if c.skipped) s_time = sum(c.time_s for c in suite.cases) total_tests += len(suite.cases) total_failures += s_failures total_skipped += s_skipped total_time += s_time suite_el = ET.SubElement( root, "testsuite", { "name": suite.name, "tests": str(len(suite.cases)), "failures": str(s_failures), "skipped": str(s_skipped), "time": f"{s_time:.3f}", }, ) for case in suite.cases: case_el = ET.SubElement( suite_el, "testcase", {"name": case.name, "classname": suite.name, "time": f"{case.time_s:.3f}"}, ) if case.skipped: ET.SubElement(case_el, "skipped") if case.failure is not None: fail_el = ET.SubElement(case_el, "failure", {"message": case.failure[:200]}) fail_el.text = case.failure root.set("tests", str(total_tests)) root.set("failures", str(total_failures)) root.set("skipped", str(total_skipped)) root.set("time", f"{total_time:.3f}") return ET.ElementTree(root) def aggregate_exit(outcomes: list[LegOutcome]) -> int: """Top-level exit code = max-severity by precedence 5 > 4 > 3 > 2 > 1 > 0. Setup/infra codes dominate test-result codes so CI can tell environment problems from regressions. A non-blocking leg (PlayMode without --strict-playmode) never raises the code above 0 on its own. """ code = 0 precedence = {5: 5, 4: 4, 3: 3, 2: 2, 1: 1, 0: 0} best_rank = -1 for o in outcomes: if o.status not in ("fail", "error"): continue if not o.blocking: continue c = o.exit_code if o.exit_code else 1 rank = precedence.get(c, 0) if rank > best_rank: best_rank = rank code = c return code # =========================================================================== # Argument parser (pure, importable, Unity-free) # =========================================================================== def build_arg_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( prog="local_harness.py", description="Local headless test harness: boot Unity, run smoke/EditMode/PlayMode legs, aggregate, teardown.", formatter_class=argparse.RawDescriptionHelpFormatter, ) p.add_argument( "--legs", default="smoke,editmode,playmode", help="Comma list of legs to run (subset of smoke,editmode,playmode).", ) p.add_argument( "--project-path", default=DEFAULT_PROJECT_PATH, help="Unity project to boot (repo-relative or absolute). Default: TestProjects/UnityMCPTests.", ) p.add_argument("--editor", default=None, help="Explicit Unity binary (discovery precedence 1).") p.add_argument("--unity-version", default=None, help="Override the resolved Unity version.") p.add_argument("--ci", action="store_true", help="Use DockerLauncher + repo .unity-mcp status dir; implies --no-warmup semantics.") p.add_argument("--status-dir", default=None, help="Status-file directory. Default: fresh tmp dir (local) / /.unity-mcp (--ci).") p.add_argument("--reuse", action="store_true", help="Attach to an already-resident bridge via ~/.unity-mcp; owns_editor=False.") p.add_argument("--keep-alive", action="store_true", help="Leave the editor running after legs (no teardown of an owned editor).") p.add_argument("--no-warmup", action="store_true", help="Skip the warm-up phase.") p.add_argument("--strict-playmode", action="store_true", help="Promote a PlayMode failure to a blocking failure (exit 1).") p.add_argument("--native-editmode", action="store_true", help="Optional CI-parity native -runTests EditMode leg, serialized after resident teardown.") p.add_argument( "--junit", default="reports/junit-e2e-bridge.xml", help="Smoke JUnit path. EditMode/PlayMode write reports/junit-editmode.xml / reports/junit-playmode.xml alongside.", ) p.add_argument("--reports", default=None, help="Reports directory. Default: dirname(--junit) or reports/.") p.add_argument("--boot-timeout", type=int, default=900, help="Warm-up + resident-boot budget (s).") p.add_argument("--bridge-wait", type=int, default=600, help="Bridge-ready budget (s).") p.add_argument( "--playmode-init-timeout", type=int, default=DEFAULT_PLAYMODE_INIT_TIMEOUT_MS, help="initTimeout (ms) passed verbatim to PlayMode run_tests.", ) p.add_argument("--overall-timeout", type=int, default=2400, help="Wall-clock watchdog (s); on expiry kill owned PID, exit 2.") p.add_argument("--max-retries", type=int, default=8, help="Reload retries per wire command.") p.add_argument("--retry-ms", type=int, default=250, help="Reload retry delay (ms).") p.add_argument("--editor-arg", action="append", default=[], dest="editor_args", help="Opaque extra editor arg appended to the resident argv (repeatable).") return p # Task-prose aliases for the canonical pure helpers (design spec §13 uses # discover_editor / resolve_version / classify_log / aggregate_exit; the harness # task wording uses these descriptive names). Both are importable + Unity-free. resolve_editor_binary = discover_editor resolve_unity_version = resolve_version classify_editor_log = classify_log aggregate_exit_code = aggregate_exit ALLOWED_LEGS = ("smoke", "editmode", "playmode") def parse_legs(legs: str) -> list[str]: """Normalize the --legs CSV into an ordered, de-duplicated list. Lenient by design -- unknown values are dropped (see test_drops_unknown_legs). The CLI entry point (main) validates raw input against ALLOWED_LEGS and rejects unknown values before calling this; keep this a pure normalizer. """ seen: set[str] = set() out: list[str] = [] for part in (legs or "").split(","): leg = part.strip().lower() if leg and leg in ALLOWED_LEGS and leg not in seen: seen.add(leg) out.append(leg) return out # =========================================================================== # Launcher seam (local vs docker). Inlined here so the harness is a single, # self-contained, runnable file. Arg-builders are pure string assemblers. # =========================================================================== @dataclass class Handle: """A live editor handle. Exactly one of (proc, container) is meaningful.""" proc: Any = None # subprocess.Popen for LocalLauncher container: str | None = None # container name for DockerLauncher log_path: str | None = None pid: int | None = None class LocalLauncher: """Boots a native Hub editor via detached Popen; PID-based liveness.""" def __init__(self, args: argparse.Namespace): self.args = args def resolve_editor(self, project_path: Path) -> EditorSpec: version = self.args.unity_version or resolve_version(project_path) if not version: raise EditorNotFound(searched=[]) return discover_editor(version, explicit_editor=self.args.editor) @staticmethod def warmup_argv(editor: str, project_path: Path, log_path: Path) -> list[str]: return [ editor, "-batchmode", "-nographics", "-quit", "-projectPath", str(project_path), "-logFile", str(log_path), ] @staticmethod def resident_argv(editor: str, project_path: Path, log_path: Path, extra_editor_args: list[str]) -> list[str]: return [ editor, "-batchmode", "-nographics", "-projectPath", str(project_path), "-logFile", str(log_path), *list(extra_editor_args or []), "-executeMethod", BOOT_METHOD, ] @staticmethod def resident_env(base_env: dict[str, str], status_dir: Path) -> dict[str, str]: env = dict(base_env) env["UNITY_MCP_ALLOW_BATCH"] = "1" env["UNITY_MCP_STATUS_DIR"] = str(status_dir) return env def warmup(self, editor: str, project_path: Path, log_path: Path, timeout_s: int) -> int: argv = self.warmup_argv(editor, project_path, log_path) proc = subprocess.Popen(argv) try: return proc.wait(timeout=timeout_s) except subprocess.TimeoutExpired: proc.kill() proc.wait() return -1 def launch(self, editor: str, project_path: Path, status_dir: Path, log_path: Path, extra_editor_args: list[str]) -> Handle: argv = self.resident_argv(editor, project_path, log_path, extra_editor_args) env = self.resident_env(os.environ, status_dir) kwargs: dict[str, Any] = {"env": env} if hasattr(os, "setsid"): kwargs["start_new_session"] = True elif sys.platform.startswith("win"): kwargs["creationflags"] = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) proc = subprocess.Popen(argv, **kwargs) return Handle(proc=proc, log_path=str(log_path), pid=proc.pid) def is_alive(self, handle: Handle) -> bool: if handle.proc is not None: return handle.proc.poll() is None if handle.pid is None: return False try: os.kill(handle.pid, 0) return True except OSError: return False def tail_log(self, handle: Handle, n: int) -> str: if not handle.log_path: return "" try: with open(handle.log_path, "r", encoding="utf-8", errors="replace") as f: return "".join(f.readlines()[-n:]) except OSError: return "" def fixup_permissions(self, status_dir: Path) -> None: return # local: no-op def teardown(self, handle: Handle, grace_s: float = 10.0) -> None: proc = handle.proc pid = handle.pid try: if proc is not None: proc.terminate() try: proc.wait(timeout=grace_s) return except subprocess.TimeoutExpired: proc.kill() proc.wait() return if pid is not None: os.kill(pid, signal.SIGTERM) deadline = time.time() + grace_s while time.time() < deadline: try: os.kill(pid, 0) except OSError: return time.sleep(0.2) try: os.kill(pid, signal.SIGKILL) except OSError: pass except OSError: pass class DockerLauncher: """Reproduces e2e-bridge.yml's docker run exactly; docker-based liveness.""" CONTAINER = "unity-mcp" def __init__(self, args: argparse.Namespace): self.args = args def resolve_editor(self, project_path: Path) -> EditorSpec: # CI short-circuits to the fixed image entrypoint; never touches Hub. return EditorSpec(binary="/opt/unity/Editor/Unity", version=self.args.unity_version or "docker") @staticmethod def docker_run_argv(image: str, workspace: Path, project_path: Path, status_dir: Path, log_path: str, extra_editor_args: list[str], container: str = "unity-mcp", runner_temp: str | None = None) -> list[str]: # When RUNNER_TEMP is present (GitHub Actions), mount the same license / # config / cache volumes the warm-up + activation steps populated so the # resident bridge container sees the staged ULF/EBL seat. Locally these # mounts are simply omitted. license_mounts: list[str] = [] if runner_temp: rt = str(runner_temp) license_mounts = [ "-v", f"{rt}/unity-config:/root/.config/unity3d", "-v", f"{rt}/unity-local:/root/.local/share/unity3d", "-v", f"{rt}/unity-cache:/root/.cache/unity3d", ] return [ "docker", "run", "-d", "--name", container, "--network", "host", "-e", "HOME=/root", "-e", "UNITY_MCP_ALLOW_BATCH=1", "-e", f"UNITY_MCP_STATUS_DIR={status_dir}", "-e", "UNITY_MCP_BIND_HOST=127.0.0.1", "-v", f"{workspace}:{workspace}", "-w", str(workspace), *license_mounts, image, "/opt/unity/Editor/Unity", "-batchmode", "-nographics", "-logFile", log_path, "-projectPath", str(project_path), *list(extra_editor_args or []), "-executeMethod", BOOT_METHOD, ] def warmup(self, editor: str, project_path: Path, log_path: Path, timeout_s: int) -> int: return 0 # no-op in CI (YAML already warmed up) def launch(self, editor: str, project_path: Path, status_dir: Path, log_path: Path, extra_editor_args: list[str]) -> Handle: image = os.environ.get("UNITY_IMAGE", "") workspace = Path(os.environ.get("GITHUB_WORKSPACE", str(REPO_ROOT))) runner_temp = os.environ.get("RUNNER_TEMP") container_log = "/root/.config/unity3d/Editor.log" subprocess.run(["docker", "rm", "-f", self.CONTAINER], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False) argv = self.docker_run_argv(image, workspace, project_path, status_dir, container_log, extra_editor_args, self.CONTAINER, runner_temp) subprocess.run(argv, check=True) return Handle(container=self.CONTAINER, log_path=container_log) def is_alive(self, handle: Handle) -> bool: try: out = subprocess.run( ["docker", "inspect", "-f", "{{.State.Status}}", self.CONTAINER], capture_output=True, text=True, check=False, ) return out.stdout.strip() == "running" except OSError: return False def tail_log(self, handle: Handle, n: int) -> str: try: out = subprocess.run( ["docker", "logs", "--tail", str(n), self.CONTAINER], capture_output=True, text=True, check=False, ) return (out.stdout or "") + (out.stderr or "") except OSError: return "" def fixup_permissions(self, status_dir: Path) -> None: subprocess.run(["docker", "exec", self.CONTAINER, "chmod", "-R", "a+rwx", str(status_dir)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False) def teardown(self, handle: Handle, grace_s: float = 10.0) -> None: subprocess.run(["docker", "rm", "-f", self.CONTAINER], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False) def make_launcher(args: argparse.Namespace): return DockerLauncher(args) if args.ci else LocalLauncher(args) # =========================================================================== # Live functions (Unity-touching; guarded under main()). # =========================================================================== def _ensure_src_on_path() -> None: """Prepend /Server/src to sys.path (mirrors bridge_smoke.py).""" src = str(SERVER_SRC) if SERVER_SRC.is_dir() and src not in sys.path: sys.path.insert(0, src) def _read_status(status_file: str) -> dict[str, Any] | None: try: with open(status_file, "r", encoding="utf-8") as f: return json.load(f) except (OSError, ValueError): return None def _tcp_probe(port: int, timeout: float = 0.5) -> bool: try: with socket.create_connection(("127.0.0.1", int(port)), timeout): return True except OSError: return False def wait_for_ready(launcher, handle: Handle, status_dir: Path, bridge_wait_s: int, boot_start: float, deadline: float) -> ReadyInfo: """Poll status-file discovery + TCP probe up to bridge_wait_s. Liveness (launcher.is_alive) is the authoritative death signal: a slow-but- healthy cold boot may show zero discoverable instances for a while, which is tolerated. On editor death, classify the log tail -> raise SystemExit with the mapped exit code (4 license / 3 compile / else 2). On overall watchdog expiry -> SystemExit(2). """ wait_deadline = time.time() + bridge_wait_s while True: now = time.time() if now > deadline: _redacted_tail(launcher, handle) raise SystemExit(2) if now > wait_deadline: print("::error:: bridge not ready before --bridge-wait deadline") _redacted_tail(launcher, handle) raise SystemExit(2) if not launcher.is_alive(handle): grace_elapsed = (now - boot_start) >= LICENSE_GRACE_S tail = launcher.tail_log(handle, 200) kind = classify_log(tail, license_grace_elapsed=grace_elapsed) print("::error:: editor process died during bridge wait") print(redact(tail)) if kind == "license_fatal": raise SystemExit(4) if kind == "compile_fatal": raise SystemExit(3) raise SystemExit(2) status_file = newest_status_file(status_dir) if status_file: data = _read_status(status_file) port = port_from_status(data) if port and _tcp_probe(port): instance_id = instance_id_from_status(status_file, data) launcher.fixup_permissions(status_dir) return ReadyInfo(port=port, instance_id=instance_id, status_file=status_file) time.sleep(2) def _redacted_tail(launcher, handle: Handle, n: int = 200) -> None: tail = launcher.tail_log(handle, n) if tail: print(redact(tail)) def _console_entries(resp: Any) -> list[Any]: """Pull read_console log entries out of either envelope shape. Non-paging read_console returns the entries as a BARE LIST under "data" (ReadConsole.HandleCommand -> SuccessResponse(message, entries)); paging returns them under data.items. Either may sit under a top-level "result". """ body = _dig(resp, "data") if isinstance(body, list): return body if isinstance(body, dict) and isinstance(body.get("items"), list): return body["items"] dug = _dig(resp, "items") if isinstance(dug, list): return dug raw = resp.get("result") if isinstance(resp, dict) else None if isinstance(raw, list): return raw return [] def compile_probe(instance_id: str, max_retries: int, retry_ms: int, send=None) -> bool: """Run a read-only read_console probe before any UTF leg. Returns True if the project compiles (no `error CS\\d`), False if a compile error is detected. Driving raw run_tests bypasses the Python preflight() that would otherwise hang on a non-compiling project, so this probe guards the UTF legs explicitly. ``send`` is an injection seam for tests; in production it defaults to the real bridge wire. """ if send is None: _ensure_src_on_path() from transport.legacy.unity_connection import send_command_with_retry as send try: resp = send( "read_console", {"action": "get", "types": ["error"], "count": "200", "include_stacktrace": False}, instance_id=instance_id, max_retries=max_retries, retry_ms=retry_ms, retry_on_reload=True, ) except Exception: # A timed-out / errored probe is inconclusive; do not block the UTF legs. return True if not _ok(resp): # An errored probe is inconclusive; do not block on it. return True entries = _console_entries(resp) for e in entries: if isinstance(e, dict): msg = str(e.get("message") or "") elif isinstance(e, str): msg = e else: msg = "" if _CS_ERROR_RE.search(msg): return False return True def run_smoke_leg(instance_id: str, junit_path: Path, max_retries: int, retry_ms: int, deadline: float | None = None, python_exe: str | None = None) -> LegOutcome: """Run bridge_smoke.py as a subprocess; honor its 0/1/2 exit contract. Bounded by the overall deadline so a wedged smoke run cannot outlive the --overall-timeout budget (the watchdog is the hard backstop; passing a timeout here reaps the child cleanly first). retry_ms is threaded through so all three legs share the configured reload-retry delay. """ smoke = REPO_ROOT / "Server" / "tests" / "e2e" / "bridge_smoke.py" py = python_exe or sys.executable argv = [py, str(smoke), "--instance", instance_id, "--junit", str(junit_path), "--max-retries", str(max_retries), "--retry-ms", str(retry_ms)] timeout = max(1.0, deadline - time.time()) if deadline is not None else None try: rc = subprocess.run(argv, check=False, timeout=timeout).returncode except subprocess.TimeoutExpired: return LegOutcome("smoke", "error", blocking=True, detail="bridge smoke timed out (overall budget)", exit_code=2) if rc == 0: return LegOutcome("smoke", "pass", blocking=True, detail="bridge smoke passed", exit_code=0) if rc == 1: return LegOutcome("smoke", "fail", blocking=True, detail="bridge smoke assertion regression", exit_code=1) # rc == 2 (or anything else) -> bridge unreachable / setup failure. return LegOutcome("smoke", "error", blocking=True, detail="no bridge reachable", exit_code=2) def _start_utf(send, mode: str, instance_id: str, init_timeout_ms: int | None, max_retries: int, retry_ms: int) -> tuple[str | None, dict[str, Any] | Any]: """Issue run_tests; return (job_id, raw_start_response). Gates on result.success.""" params: dict[str, Any] = {"mode": mode, "includeFailedTests": True} if init_timeout_ms is not None: params["initTimeout"] = init_timeout_ms # tests_running back-off: a "tests already running" reply is an ErrorResponse # (success:false), so it must be detected BEFORE the _ok() gate, not after. for _ in range(5): try: start = send("run_tests", params, instance_id=instance_id, max_retries=max_retries, retry_ms=retry_ms, retry_on_reload=True) except Exception: # Transport hiccup starting the job (editor briefly busy / reloading); # back off and retry rather than crashing. time.sleep(min(float(retry_ms) / 1000.0 * 2, 2.0)) continue err = None if isinstance(start, dict): err = start.get("error") or start.get("code") or _dig(start, "error") if err == "tests_running": back = _dig(start, "retry_after_ms") or 1000 time.sleep(min(float(back) / 1000.0, 5.0)) continue if not _ok(start): return None, start job_id = _dig(start, "job_id") if isinstance(job_id, (str, int)): return str(job_id), start return None, start return None, {"error": "tests_running (exhausted)"} def _poll_utf(send, job_id: str, instance_id: str, deadline: float, max_retries: int, retry_ms: int) -> dict[str, Any] | Any: """Poll get_test_job until terminal {succeeded, failed} or deadline. A returned MCPResponse / reason=="reloading" / hint=="retry" is treated as non-terminal (keep polling). Returns the last response (terminal or a wedge marker dict when the deadline expires). """ while time.time() < deadline: try: poll = send("get_test_job", {"job_id": job_id, "includeFailedTests": True}, instance_id=instance_id, max_retries=max_retries, retry_ms=retry_ms, retry_on_reload=True) except Exception: # Unity blocks its main thread while running tests, so get_test_job can # time out mid-run. That is NOT terminal -- keep polling until the editor # frees up or the deadline expires (rather than crashing the harness). time.sleep(2) continue if not isinstance(poll, dict): time.sleep(2) continue reason = _dig(poll, "reason") hint = _dig(poll, "hint") if reason == "reloading" or hint == "retry": time.sleep(2) continue status = _dig(poll, "status") if status in ("succeeded", "failed"): return poll time.sleep(2) return {"_wedge": True} def _outcome_from_terminal(name: str, mode: str, terminal: dict[str, Any] | Any, blocking: bool) -> LegOutcome: """Map a terminal get_test_job response into a LegOutcome + JUnit suite.""" status = _dig(terminal, "status") suite = JUnitSuite(name=name) if isinstance(terminal, dict) and terminal.get("_wedge"): suite.cases.append(JUnitCase(name=f"{mode}.wedge", failure="no terminal status within budget")) return LegOutcome(name, "fail", blocking=blocking, detail="wedge (no terminal status)", exit_code=1, junit_suite=suite) if status == "succeeded": result = _dig(terminal, "result") or {} summary = (result.get("summary") if isinstance(result, dict) else None) or {} total = int(summary.get("total", 0) or 0) passed = int(summary.get("passed", 0) or 0) failed = int(summary.get("failed", 0) or 0) skipped = int(summary.get("skipped", 0) or 0) duration = float(summary.get("durationSeconds", 0.0) or 0.0) rows = result.get("results") if isinstance(result, dict) else None if isinstance(rows, list) and rows: for r in rows: if not isinstance(r, dict): continue rname = str(r.get("fullName") or r.get("name") or f"{mode}.test") rtime = float(r.get("durationSeconds", 0.0) or 0.0) state = str(r.get("state") or "") if state.lower() in ("failed", "error"): fmsg = str(r.get("message") or "") + "\n" + str(r.get("stackTrace") or "") suite.cases.append(JUnitCase(name=rname, time_s=rtime, failure=fmsg.strip())) elif state.lower() in ("skipped", "ignored", "inconclusive"): suite.cases.append(JUnitCase(name=rname, time_s=rtime, skipped=True)) else: suite.cases.append(JUnitCase(name=rname, time_s=rtime)) else: # No per-test rows: synthesize from the summary. for i in range(passed): suite.cases.append(JUnitCase(name=f"{mode}.passed.{i}")) for i in range(failed): suite.cases.append(JUnitCase(name=f"{mode}.failed.{i}", failure="failed (no detail)")) for i in range(skipped): suite.cases.append(JUnitCase(name=f"{mode}.skipped.{i}", skipped=True)) if not suite.cases and total == 0: suite.cases.append(JUnitCase(name=f"{mode}.empty", time_s=duration)) if failed > 0: return LegOutcome(name, "fail", blocking=blocking, detail=f"{failed}/{total} {mode} tests failed", exit_code=1, junit_suite=suite) return LegOutcome(name, "pass", blocking=blocking, detail=f"{passed}/{total} {mode} tests passed", exit_code=0, junit_suite=suite) # status == "failed": data.result is null; surface error + capped failures. error = _dig(terminal, "error") or "test job failed" failures = _dig(terminal, "failures_so_far") or [] detail = str(error) fail_text = detail if isinstance(failures, list) and failures: for fr in failures: if isinstance(fr, dict): fail_text += "\n - " + str(fr.get("full_name") or "") + ": " + str(fr.get("message") or "") suite.cases.append(JUnitCase(name=f"{mode}.job", failure=fail_text)) return LegOutcome(name, "fail", blocking=blocking, detail=detail, exit_code=1, junit_suite=suite) def run_utf_leg(mode: str, instance_id: str, blocking: bool, deadline: float, max_retries: int, retry_ms: int, init_timeout_ms: int | None = None) -> LegOutcome: """Drive one EditMode/PlayMode leg over the raw run_tests/get_test_job wire.""" _ensure_src_on_path() from transport.legacy.unity_connection import send_command_with_retry as send name = "editmode" if mode == "EditMode" else "playmode" job_id, start = _start_utf(send, mode, instance_id, init_timeout_ms, max_retries, retry_ms) if job_id is None: suite = JUnitSuite(name=name, cases=[JUnitCase(name=f"{name}.start", failure=_message(start) or "run_tests start failed")]) return LegOutcome(name, "fail", blocking=blocking, detail="run_tests start failed", exit_code=1, junit_suite=suite) terminal = _poll_utf(send, job_id, instance_id, deadline, max_retries, retry_ms) return _outcome_from_terminal(name, mode, terminal, blocking) def _ensure_clean_editmode(send, instance_id: str, max_retries: int, retry_ms: int, deadline: float) -> None: """Best-effort: wait until no PlayMode/test job is running (S0).""" end = min(time.time() + 30, deadline) while time.time() < end: try: resp = send("get_test_job", {"job_id": ""}, instance_id=instance_id, max_retries=max_retries, retry_ms=retry_ms, retry_on_reload=True) except Exception: # Editor busy/unresponsive; best-effort wait, treat as still settling. time.sleep(2) continue status = _dig(resp, "status") if status not in ("running",): return time.sleep(2) def run_playmode_with_retry(instance_id: str, deadline: float, max_retries: int, retry_ms: int, init_timeout_ms: int, strict: bool, relaunch: Callable[[], str] | None = None) -> LegOutcome: """PlayMode state machine: start, poll, classify-can-rerun, retry ONCE. Non-blocking by default; --strict-playmode promotes failure to blocking. On a "can rerun" failure (error contains 'failed to initialize') or a wedge, re-establish clean EditMode and repeat once. A wedge may relaunch the editor (honoring the socket-release delay) before the single retry. """ _ensure_src_on_path() from transport.legacy.unity_connection import send_command_with_retry as send blocking = bool(strict) def attempt(inst: str) -> LegOutcome: _ensure_clean_editmode(send, inst, max_retries, retry_ms, deadline) job_id, start = _start_utf(send, "PlayMode", inst, init_timeout_ms, max_retries, retry_ms) if job_id is None: suite = JUnitSuite(name="playmode", cases=[JUnitCase(name="playmode.start", failure=_message(start) or "run_tests start failed")]) return LegOutcome("playmode", "fail", blocking=blocking, detail="run_tests start failed", exit_code=1, junit_suite=suite) terminal = _poll_utf(send, job_id, inst, deadline, max_retries, retry_ms) return _outcome_from_terminal("playmode", "PlayMode", terminal, blocking) first = attempt(instance_id) if first.status == "pass": return first error_text = (first.detail or "").lower() can_rerun = ("failed to initialize" in error_text) or ("wedge" in error_text) if not can_rerun: return first # A wedge may need the editor relaunched (respecting the socket-release delay). inst = instance_id if "wedge" in error_text and relaunch is not None: time.sleep(SOCKET_RELEASE_MS / 1000.0) inst = relaunch() second = attempt(inst) second.detail = f"retry-once: {second.detail}" return second def write_reports(junit_path: Path, reports_dir: Path, outcomes: list[LegOutcome]) -> None: """Write merged + per-leg JUnit files: smoke -> --junit, others alongside.""" reports_dir.mkdir(parents=True, exist_ok=True) per_leg_file = { "smoke": junit_path, "editmode": reports_dir / "junit-editmode.xml", "playmode": reports_dir / "junit-playmode.xml", } for o in outcomes: if o.junit_suite is None: continue target = per_leg_file.get(o.name, reports_dir / f"junit-{o.name}.xml") merge_junit([o.junit_suite]).write(str(target), encoding="utf-8", xml_declaration=True) # Also write a combined report. suites = [o.junit_suite for o in outcomes if o.junit_suite is not None] if suites: merge_junit(suites).write(str(reports_dir / "junit-all.xml"), encoding="utf-8", xml_declaration=True) def _print_summary(outcomes: list[LegOutcome], exit_code: int) -> None: print("== local harness summary ==") for o in outcomes: tag = {"pass": "PASS", "fail": "FAIL", "skip": "SKIP", "error": "ERROR"}.get(o.status, o.status.upper()) block = "blocking" if o.blocking else "non-blocking" print(f" [{tag}] {o.name} ({block}) -- {o.detail}") print(f"== exit {exit_code} ==") def main(argv: list[str] | None = None) -> int: args = build_arg_parser().parse_args(argv) requested_legs = [p.strip().lower() for p in (args.legs or "").split(",") if p.strip()] invalid_legs = [leg for leg in requested_legs if leg not in ALLOWED_LEGS] if invalid_legs: print(f"::error:: --legs included invalid value(s): {', '.join(invalid_legs)} " f"(allowed: {', '.join(ALLOWED_LEGS)})") return 2 legs = parse_legs(args.legs) if not legs: print(f"::error:: --legs did not include any valid values (allowed: {', '.join(ALLOWED_LEGS)})") return 2 if args.ci: args.no_warmup = True if not os.environ.get("UNITY_IMAGE"): print("::error:: --ci requires the UNITY_IMAGE environment variable " "(the unityci/editor image to run the headless Editor in)") return 2 # Resolve project path (repo-relative or absolute). project_path = Path(args.project_path) if not project_path.is_absolute(): project_path = (REPO_ROOT / project_path).resolve() # Reports / JUnit geometry. junit_path = Path(args.junit) if not junit_path.is_absolute(): junit_path = (REPO_ROOT / junit_path).resolve() reports_dir = Path(args.reports).resolve() if args.reports else junit_path.parent # Status-dir + isolation. Default local: fresh tmp dir; --ci / --status-dir override. owns_status_dir = False if args.status_dir: sd = Path(args.status_dir) status_dir = sd if sd.is_absolute() else (REPO_ROOT / sd).resolve() status_dir.mkdir(parents=True, exist_ok=True) elif args.ci: status_dir = (REPO_ROOT / ".unity-mcp").resolve() status_dir.mkdir(parents=True, exist_ok=True) elif args.reuse: status_dir = Path.home() / ".unity-mcp" else: status_dir = Path(tempfile.mkdtemp(prefix="unity-mcp-harness-")) owns_status_dir = True launcher = make_launcher(args) boot_start = time.time() deadline = boot_start + args.overall_timeout handle: Handle | None = None owns_editor = not (args.reuse or args.keep_alive) outcomes: list[LegOutcome] = [] def do_teardown() -> None: # Only kill the editor we started; clean only our own status files. if handle is not None and owns_editor and not args.keep_alive: try: launcher.teardown(handle) except Exception: pass if owns_status_dir: try: for p in glob.glob(str(status_dir / "unity-mcp-status-*.json")): try: os.remove(p) except OSError: pass shutil.rmtree(status_dir, ignore_errors=True) except OSError: pass def _signal_handler(signum, frame): # pragma: no cover - signal path do_teardown() raise SystemExit(2) for sig in (signal.SIGINT, signal.SIGTERM): try: signal.signal(sig, _signal_handler) except (ValueError, OSError): pass # Hard wall-clock watchdog: the polled deadlines inside the wait/poll loops do # not cover blocking subprocess/socket calls (smoke subprocess, compile probe), # so a background daemon enforces --overall-timeout across ALL phases. _watchdog_stop = threading.Event() def _watchdog() -> None: # pragma: no cover - timing/daemon path remaining = deadline - time.time() if remaining > 0: _watchdog_stop.wait(remaining) if _watchdog_stop.is_set(): return print(f"::error:: overall watchdog timed out after {args.overall_timeout}s -- killing editor", flush=True) try: do_teardown() finally: os._exit(2) threading.Thread(target=_watchdog, name="harness-watchdog", daemon=True).start() try: # --- Reuse path: attach to a resident bridge via ~/.unity-mcp --- if args.reuse: # Read the user's default registry without our own STATUS_DIR override. prev = os.environ.pop("UNITY_MCP_STATUS_DIR", None) try: status_file = _find_reuse_status(status_dir, project_path) finally: if prev is not None: os.environ["UNITY_MCP_STATUS_DIR"] = prev if not status_file: print(f"::error:: --reuse: no resident bridge found for {project_path} under {status_dir}") return 2 data = _read_status(status_file) port = port_from_status(data) if not port or not _tcp_probe(port): print(f"::error:: --reuse: resident bridge for {project_path} is not reachable") return 2 os.environ["UNITY_MCP_STATUS_DIR"] = str(status_dir) instance_id = instance_id_from_status(status_file, data) ready = ReadyInfo(port=port, instance_id=instance_id, status_file=status_file) else: # --- Boot path --- try: spec = launcher.resolve_editor(project_path) except EditorNotFound as e: searched = e.searched if e.searched else [""] print("::error:: no matching Unity editor found") for s in searched: print(f" searched: {s}") want = args.unity_version or resolve_version(project_path) or "" maj_min = ".".join(str(x) for x in parse_version(want)[:2]) print(f" install a matching {maj_min}.x editor or pass --editor (wanted {want})") return 5 os.environ["UNITY_MCP_STATUS_DIR"] = str(status_dir) # Phase 1 -- warm-up (skip with --no-warmup / --ci). if not args.no_warmup: warmup_log = status_dir / "warmup.log" rc = launcher.warmup(spec.binary, project_path, warmup_log, args.boot_timeout) if rc not in (0,): tail = launcher.tail_log(Handle(log_path=str(warmup_log)), 200) kind = classify_log(tail, license_grace_elapsed=(time.time() - boot_start) >= LICENSE_GRACE_S) if kind == "license_fatal": print(redact(tail)) return 4 if kind == "compile_fatal": print(redact(tail)) return 3 # Non-zero warm-up without a clear signal: continue to resident boot. # Phase 2 -- resident (NO -quit). editor_log = status_dir / "editor.log" handle = launcher.launch(spec.binary, project_path, status_dir, editor_log, args.editor_args) ready = wait_for_ready(launcher, handle, status_dir, args.bridge_wait, boot_start, deadline) instance_id = ready.instance_id # Pin the instance so smoke + UTF target our own editor. os.environ["UNITY_MCP_DEFAULT_INSTANCE"] = instance_id print(f"== bridge ready: instance={instance_id} port={ready.port} ==") # --- Compile probe before any UTF leg (exit 3 on compile failure) --- wants_utf = ("editmode" in legs) or ("playmode" in legs) compile_ok = True if wants_utf: compile_ok = compile_probe(instance_id, args.max_retries, args.retry_ms) if not compile_ok: print("::error:: project does not compile -- skipping UTF legs") # --- Smoke leg --- if "smoke" in legs: outcomes.append(run_smoke_leg(instance_id, junit_path, args.max_retries, args.retry_ms, deadline=deadline)) # --- EditMode leg --- if "editmode" in legs: if not compile_ok: outcomes.append(LegOutcome("editmode", "fail", blocking=True, detail="project does not compile", exit_code=3)) else: outcomes.append(run_utf_leg("EditMode", instance_id, blocking=True, deadline=deadline, max_retries=args.max_retries, retry_ms=args.retry_ms)) # --- PlayMode leg (default-ON, NON-BLOCKING unless --strict-playmode) --- if "playmode" in legs: if not compile_ok: outcomes.append(LegOutcome("playmode", "fail", blocking=bool(args.strict_playmode), detail="project does not compile", exit_code=3)) else: def _relaunch() -> str: nonlocal handle, ready, instance_id if handle is not None and owns_editor: launcher.teardown(handle) time.sleep(SOCKET_RELEASE_MS / 1000.0) editor_log = status_dir / "editor.log" handle = launcher.launch(spec.binary, project_path, status_dir, editor_log, args.editor_args) ready = wait_for_ready(launcher, handle, status_dir, args.bridge_wait, time.time(), deadline) instance_id = ready.instance_id os.environ["UNITY_MCP_DEFAULT_INSTANCE"] = instance_id return instance_id relaunch = _relaunch if (owns_editor and not args.reuse) else None outcomes.append(run_playmode_with_retry( instance_id, deadline, args.max_retries, args.retry_ms, args.playmode_init_timeout, bool(args.strict_playmode), relaunch=relaunch)) # Aggregate + write reports. write_reports(junit_path, reports_dir, outcomes) exit_code = aggregate_exit(outcomes) # If a compile failure surfaced, it dominates (precedence keeps 3). _print_summary(outcomes, exit_code) return exit_code except SystemExit as e: return int(e.code) if isinstance(e.code, int) else 2 finally: _watchdog_stop.set() do_teardown() def _norm_project_root(p: str) -> str: """Normalize a project path for comparison: strip a trailing Assets, case, sep.""" p = p.rstrip("/\\") if p.lower().endswith("assets"): p = p[:-6].rstrip("/\\") return os.path.normcase(os.path.normpath(p)) def _find_reuse_status(status_dir: Path, project_path: Path) -> str | None: """Find a status file under status_dir whose project root matches project_path. Matches on the full normalized project root (not just the leaf folder name) so two projects sharing a folder name never attach to the wrong resident editor. Refuses (returns None) rather than attach to a non-matching instance. """ want = _norm_project_root(str(project_path)) files = sorted( glob.glob(str(status_dir / "unity-mcp-status-*.json")), key=lambda p: os.path.getmtime(p) if os.path.exists(p) else 0, reverse=True, ) for f in files: data = _read_status(f) if not data: continue pp = data.get("project_path") or "" if isinstance(pp, str) and pp and _norm_project_root(pp) == want: return f return None # refuse rather than attach to a non-matching instance if __name__ == "__main__": raise SystemExit(main())