Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 0 additions & 4 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,3 @@ line-length = 120
select = ["E", "F", "I", "N", "UP", "B"]
ignore = ["E501"]

[tool.ruff.lint.per-file-ignores]
# mcp_server.py inserts the repo root on sys.path before importing first-party
# modules, so those imports legitimately come after executable statements.
"ue_mcp/mcp_server.py" = ["E402"]
15 changes: 13 additions & 2 deletions remote_control/codegen.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,14 @@
"""


def _escape_for_fstring(value: str) -> str:
"""Escape backslashes and double quotes for embedding in generated code.

Escaping lives HERE for find_assets so every caller (MCP tool, sync
client) is covered; other templates still rely on caller-side escaping.
"""
return value.replace("\\", "\\\\").replace('"', '\\"')


class _CodeGen:
"""Generates UE5 Python scripts. No I/O -- pure string construction."""
Expand Down Expand Up @@ -36,10 +44,12 @@ def spawn_actor_code(

@staticmethod
def delete_actor_code(actor_path: str) -> str:
# Level actors are subobjects (…:PersistentLevel.Name) — the Content-Browser
# asset API returns None for them; load_object resolves both forms.
return f"""
import unreal
subsystem = unreal.get_editor_subsystem(unreal.EditorActorSubsystem)
actor = unreal.EditorAssetLibrary.load_asset("{actor_path}")
actor = unreal.load_object(None, "{actor_path}")
if actor:
subsystem.destroy_actor(actor)
print("RESULT:DELETED")
Expand Down Expand Up @@ -77,7 +87,7 @@ def set_actor_transform_code(
scale: tuple[float, float, float] | None,
) -> str:
lines = ["import unreal"]
lines.append(f'actor = unreal.EditorAssetLibrary.load_asset("{actor_path}")')
lines.append(f'actor = unreal.load_object(None, "{actor_path}")')
lines.append("if actor:")
if location:
lines.append(f" actor.set_actor_location(unreal.Vector({location[0]}, {location[1]}, {location[2]}), False, False)")
Expand All @@ -92,6 +102,7 @@ def set_actor_transform_code(

@staticmethod
def find_assets_code(search_pattern: str, class_filter: str | None = None) -> str:
search_pattern = _escape_for_fstring(search_pattern)
return f"""
import unreal, json
registry = unreal.AssetRegistryHelpers.get_asset_registry()
Expand Down
59 changes: 59 additions & 0 deletions tests/exec_sim/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
# exec_sim — exec-simulating codegen harness

The mock suite asserts on generated code *strings*; this harness **runs** the
generated UE editor Python against a strict fake `unreal` module and asserts on
*behavior*. It exists to catch the bug classes string-assertions are blind to:
syntax errors, NameErrors in success branches, args validated-then-discarded,
phantom (nonexistent) `unreal` APIs, and hard-coded success prints.

## Files

| File | Role |
|------|------|
| `unreal_stub.py` | `make_unreal_stub()` builds a fresh fake `unreal` module (strict: unknown top-level symbols raise `AttributeError`). `exec_generated()` installs it in `sys.modules` and runs a script, capturing stdout. `parse_result()` reuses the product RESULT-line parser. |
| `registry.py` | One `ToolEntry` per registered tool: canonical sentinel kwargs, `mode` (`CODEGEN`/`DIRECT`), `sentinel_checkable` kwargs, `expect_error`, notes. |
| `conftest.py` | Registers all tools once per session against a recording fake server and a `CaptureUE` (real `AsyncUnrealRemoteControl` delegation, `execute_python` captures instead of sending). |
| `test_codegen_exec.py` | The four generic gates (below), parametrized over every CODEGEN tool. |
| `test_honesty.py` | Scripted-failure contracts: force a real-world failure, assert the code does not claim success. |

## The gates and what each catches

1. **Registry completeness** — `REGISTRY` keys must equal the registered tool
names exactly. A new tool cannot dodge the harness.
2. **Compile** — generated Python must `compile()`. Catches interpolation /
indentation breakage (e.g. a label kwarg producing an `IndentationError`).
3. **Exec + honest success** — run under the default all-success stub: must
finish, print a `RESULT:` line, and not report failure. Catches phantom
APIs (`AttributeError`), bare JSON literals (`true` → `NameError`),
unserializable results (`unreal.Name` dict keys → `TypeError`), and dead
tools that can *never* succeed.
4. **Sentinel** — each `sentinel_checkable` kwarg's value must appear literally
(raw or `escape_for_fstring`-escaped) in the source. Catches
validated-then-discarded arguments.
5. **Honesty (scripted failure)** — `stub.configure(load_level=False)`-style
overrides force failure paths; the RESULT must not claim success.

## Adding a new tool

1. Register it as usual in `ue_mcp/tools/`.
2. Add a `ToolEntry` in `registry.py` (gate 1 fails until you do):
- `mode=CODEGEN` with distinctive sentinel kwargs (dyadic floats like
`433.25` survive `str()`/`json.dumps` exactly), listing in
`sentinel_checkable` every kwarg embedded *literally* in the script;
- or `mode=DIRECT` with a `notes` reason (HTTP-only, pass-through, pure
server-side...).
3. If the generated code uses a new `unreal` symbol, add it to the stub —
**only after verifying it exists in the real UE Python API**. Phantom
symbols staying absent is the whole point (`editor_undo`,
`transaction_undo`, and actor `is_hidden()` are deliberately missing).
4. If the tool needs a forced-failure contract, add a `stub.configure(...)`
flag and a test in `test_honesty.py`.

## Notes

- The stub's seeded world (actors `SENTINEL_LBL_9Q`, `SENTINEL_LBL_B2`,
`Cube_1`) matches the registry sentinels so success branches actually run.
- `ue_viewport_percept` is DIRECT (HTTP primary path), but its Python fallback
is exec-simmed via `perception._fallback_capture` in `test_honesty`.
- `ue_status` / `ue_health_check` live in `mcp_server.py`, outside
`register_all_tools`, and are out of scope here.
1 change: 1 addition & 0 deletions tests/exec_sim/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
"""Exec-simulating codegen test harness (see README.md in this directory)."""
98 changes: 98 additions & 0 deletions tests/exec_sim/conftest.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,98 @@
"""Fixtures for the exec-sim harness.

Registers all tools ONCE per session against:

- a recording fake server (satisfies the MCPServer Protocol; stores
``{name: (fn, annotations)}``), and
- a capture UE client that subclasses the REAL ``AsyncUnrealRemoteControl``
but overrides only ``execute_python`` -- so client-side codegen delegation
(spawn_actor -> _CodeGen.spawn_actor_code, ...) is exercised verbatim while
every generated script is captured instead of sent over HTTP.

Tests are plain sync functions; async tool coroutines are driven with
``asyncio.run`` (pytest-asyncio strict mode stays untouched).
"""

from __future__ import annotations

import asyncio
import sys
from pathlib import Path

import pytest

PROJECT_ROOT = Path(__file__).resolve().parents[2]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))

from remote_control.async_client import AsyncUnrealRemoteControl # noqa: E402
from tests.exec_sim.registry import REGISTRY # noqa: E402
from ue_mcp.tools import register_all_tools # noqa: E402


class RecordingServer:
"""Fake MCP server: records registrations instead of serving them."""

def __init__(self):
self.tools: dict[str, tuple] = {}

def tool(self, *, name: str, description: str, annotations: dict | None = None):
def decorator(fn):
if name in self.tools:
raise AssertionError(f"duplicate tool registration: {name}")
self.tools[name] = (fn, annotations)
return fn

return decorator


class CaptureUE(AsyncUnrealRemoteControl):
"""Real client delegation, fake transport.

Deliberately does NOT call super().__init__ -- no HTTP client, no circuit
breaker. The convenience methods (spawn_actor, delete_actor, list_actors,
set_actor_transform, find_assets, get_level_info, save_level) run their
real code paths and land in this overridden execute_python.
"""

def __init__(self): # noqa: D107 -- see class docstring
self.captured: list[str] = []
# Shape mirrors remote_control.execution._parse_result output.
self.result: dict = {"result": None, "output": "", "error": None}

async def execute_python(self, code: str) -> dict:
self.captured.append(code)
return dict(self.result)


class Toolbox:
"""Session-wide registration + code capture with per-tool caching."""

def __init__(self):
self.server = RecordingServer()
self.ue = CaptureUE()
register_all_tools(self.server, self.ue)
self._code_cache: dict[str, list[str]] = {}

@property
def registered_names(self) -> set[str]:
return set(self.server.tools)

def invoke(self, tool_name: str, **kwargs) -> list[str]:
"""Invoke a registered tool coroutine; return the captured scripts."""
fn, _annotations = self.server.tools[tool_name]
self.ue.captured = []
asyncio.run(fn(**kwargs))
return list(self.ue.captured)

def codes_for(self, tool_name: str) -> list[str]:
"""Captured scripts for the registry's canonical sentinel kwargs (cached)."""
if tool_name not in self._code_cache:
entry = REGISTRY[tool_name]
self._code_cache[tool_name] = self.invoke(tool_name, **entry.kwargs)
return self._code_cache[tool_name]


@pytest.fixture(scope="session")
def toolbox() -> Toolbox:
return Toolbox()
Loading
Loading