305 lines
11 KiB
Python
305 lines
11 KiB
Python
import os
|
|
import time
|
|
from typing import Any
|
|
|
|
from fastmcp import Context
|
|
from pydantic import BaseModel
|
|
|
|
from models import MCPResponse
|
|
from services.registry import mcp_for_unity_resource
|
|
from services.tools import get_unity_instance_from_context
|
|
from services.state.external_changes_scanner import external_changes_scanner
|
|
import transport.unity_transport as unity_transport
|
|
from transport.legacy.unity_connection import async_send_command_with_retry
|
|
|
|
|
|
class EditorStateUnity(BaseModel):
|
|
instance_id: str | None = None
|
|
unity_version: str | None = None
|
|
project_id: str | None = None
|
|
platform: str | None = None
|
|
is_batch_mode: bool | None = None
|
|
|
|
|
|
class EditorStatePlayMode(BaseModel):
|
|
is_playing: bool | None = None
|
|
is_paused: bool | None = None
|
|
is_changing: bool | None = None
|
|
|
|
|
|
class EditorStateActiveScene(BaseModel):
|
|
path: str | None = None
|
|
guid: str | None = None
|
|
name: str | None = None
|
|
|
|
|
|
class EditorStateEditor(BaseModel):
|
|
is_focused: bool | None = None
|
|
play_mode: EditorStatePlayMode | None = None
|
|
active_scene: EditorStateActiveScene | None = None
|
|
|
|
|
|
class EditorStateActivity(BaseModel):
|
|
phase: str | None = None
|
|
since_unix_ms: int | None = None
|
|
reasons: list[str] | None = None
|
|
|
|
|
|
class EditorStateCompilation(BaseModel):
|
|
is_compiling: bool | None = None
|
|
is_domain_reload_pending: bool | None = None
|
|
last_compile_started_unix_ms: int | None = None
|
|
last_compile_finished_unix_ms: int | None = None
|
|
last_domain_reload_before_unix_ms: int | None = None
|
|
last_domain_reload_after_unix_ms: int | None = None
|
|
|
|
|
|
class EditorStateRefresh(BaseModel):
|
|
is_refresh_in_progress: bool | None = None
|
|
last_refresh_requested_unix_ms: int | None = None
|
|
last_refresh_finished_unix_ms: int | None = None
|
|
|
|
|
|
class EditorStateAssets(BaseModel):
|
|
is_updating: bool | None = None
|
|
external_changes_dirty: bool | None = None
|
|
external_changes_last_seen_unix_ms: int | None = None
|
|
external_changes_dirty_since_unix_ms: int | None = None
|
|
external_changes_last_cleared_unix_ms: int | None = None
|
|
refresh: EditorStateRefresh | None = None
|
|
|
|
|
|
class EditorStateLastRun(BaseModel):
|
|
finished_unix_ms: int | None = None
|
|
result: str | None = None
|
|
counts: Any | None = None
|
|
|
|
|
|
class EditorStateTests(BaseModel):
|
|
is_running: bool | None = None
|
|
mode: str | None = None
|
|
current_job_id: str | None = None
|
|
started_unix_ms: int | None = None
|
|
started_by: str | None = None
|
|
last_run: EditorStateLastRun | None = None
|
|
|
|
|
|
class EditorStateTransport(BaseModel):
|
|
unity_bridge_connected: bool | None = None
|
|
last_message_unix_ms: int | None = None
|
|
|
|
|
|
class EditorStateAdvice(BaseModel):
|
|
ready_for_tools: bool | None = None
|
|
blocking_reasons: list[str] | None = None
|
|
recommended_retry_after_ms: int | None = None
|
|
recommended_next_action: str | None = None
|
|
|
|
|
|
class EditorStateStaleness(BaseModel):
|
|
age_ms: int | None = None
|
|
is_stale: bool | None = None
|
|
|
|
|
|
class EditorStateData(BaseModel):
|
|
schema_version: str
|
|
observed_at_unix_ms: int
|
|
sequence: int
|
|
unity: EditorStateUnity | None = None
|
|
editor: EditorStateEditor | None = None
|
|
activity: EditorStateActivity | None = None
|
|
compilation: EditorStateCompilation | None = None
|
|
assets: EditorStateAssets | None = None
|
|
tests: EditorStateTests | None = None
|
|
transport: EditorStateTransport | None = None
|
|
advice: EditorStateAdvice | None = None
|
|
staleness: EditorStateStaleness | None = None
|
|
|
|
|
|
def _now_unix_ms() -> int:
|
|
return int(time.time() * 1000)
|
|
|
|
|
|
def _in_pytest() -> bool:
|
|
# Avoid instance-discovery side effects during the Python integration test suite.
|
|
return bool(os.environ.get("PYTEST_CURRENT_TEST"))
|
|
|
|
|
|
async def infer_single_instance_id(ctx: Context) -> str | None:
|
|
"""
|
|
Best-effort: if exactly one Unity instance is connected, return its Name@hash id.
|
|
This makes editor_state outputs self-describing even when no explicit active instance is set.
|
|
"""
|
|
await ctx.info("If exactly one Unity instance is connected, return its Name@hash id.")
|
|
|
|
try:
|
|
transport = unity_transport._current_transport()
|
|
except Exception:
|
|
transport = None
|
|
|
|
if transport == "http":
|
|
# HTTP/WebSocket transport: derive from PluginHub sessions.
|
|
try:
|
|
from transport.plugin_hub import PluginHub
|
|
|
|
sessions_data = await PluginHub.get_sessions()
|
|
sessions = sessions_data.sessions if hasattr(
|
|
sessions_data, "sessions") else {}
|
|
if isinstance(sessions, dict) and len(sessions) == 1:
|
|
session = next(iter(sessions.values()))
|
|
project = getattr(session, "project", None)
|
|
project_hash = getattr(session, "hash", None)
|
|
if project and project_hash:
|
|
return f"{project}@{project_hash}"
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
# Stdio/TCP transport: derive from connection pool discovery.
|
|
try:
|
|
from transport.legacy.unity_connection import get_unity_connection_pool
|
|
|
|
pool = get_unity_connection_pool()
|
|
instances = pool.discover_all_instances(force_refresh=False)
|
|
if isinstance(instances, list) and len(instances) == 1:
|
|
inst = instances[0]
|
|
inst_id = getattr(inst, "id", None)
|
|
return str(inst_id) if inst_id else None
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
|
|
def _enrich_advice_and_staleness(state_v2: dict[str, Any]) -> dict[str, Any]:
|
|
now_ms = _now_unix_ms()
|
|
observed = state_v2.get("observed_at_unix_ms")
|
|
try:
|
|
observed_ms = int(observed)
|
|
except Exception:
|
|
observed_ms = now_ms
|
|
|
|
age_ms = max(0, now_ms - observed_ms)
|
|
# Conservative default: treat >2s as stale (covers common unfocused-editor throttling).
|
|
is_stale = age_ms > 2000
|
|
|
|
compilation = state_v2.get("compilation") or {}
|
|
tests = state_v2.get("tests") or {}
|
|
assets = state_v2.get("assets") or {}
|
|
refresh = (assets.get("refresh") or {}) if isinstance(assets, dict) else {}
|
|
|
|
blocking: list[str] = []
|
|
if compilation.get("is_compiling") is True:
|
|
blocking.append("compiling")
|
|
if compilation.get("is_domain_reload_pending") is True:
|
|
blocking.append("domain_reload")
|
|
if tests.get("is_running") is True:
|
|
blocking.append("running_tests")
|
|
if refresh.get("is_refresh_in_progress") is True:
|
|
blocking.append("asset_refresh")
|
|
if is_stale:
|
|
blocking.append("stale_status")
|
|
|
|
ready_for_tools = len(blocking) == 0
|
|
|
|
state_v2["advice"] = {
|
|
"ready_for_tools": ready_for_tools,
|
|
"blocking_reasons": blocking,
|
|
"recommended_retry_after_ms": 0 if ready_for_tools else 500,
|
|
"recommended_next_action": "none" if ready_for_tools else "retry_later",
|
|
}
|
|
state_v2["staleness"] = {"age_ms": age_ms, "is_stale": is_stale}
|
|
return state_v2
|
|
|
|
|
|
@mcp_for_unity_resource(
|
|
uri="mcpforunity://editor/state",
|
|
name="editor_state",
|
|
description="Canonical editor readiness snapshot. Includes advice and server-computed staleness.",
|
|
)
|
|
async def get_editor_state(ctx: Context) -> MCPResponse:
|
|
unity_instance = get_unity_instance_from_context(ctx)
|
|
|
|
response = await unity_transport.send_with_unity_instance(
|
|
async_send_command_with_retry,
|
|
unity_instance,
|
|
"get_editor_state",
|
|
{},
|
|
)
|
|
|
|
# If Unity returns a structured retry hint or error, surface it directly.
|
|
if isinstance(response, dict) and not response.get("success", True):
|
|
return MCPResponse(**response)
|
|
|
|
state_v2 = response.get("data") if isinstance(
|
|
response, dict) and isinstance(response.get("data"), dict) else {}
|
|
state_v2.setdefault("schema_version", "unity-mcp/editor_state@2")
|
|
state_v2.setdefault("observed_at_unix_ms", _now_unix_ms())
|
|
state_v2.setdefault("sequence", 0)
|
|
|
|
# Ensure the returned snapshot is clearly associated with the targeted instance.
|
|
unity_section = state_v2.get("unity")
|
|
if not isinstance(unity_section, dict):
|
|
unity_section = {}
|
|
state_v2["unity"] = unity_section
|
|
current_instance_id = unity_section.get("instance_id")
|
|
if current_instance_id in (None, ""):
|
|
if unity_instance:
|
|
unity_section["instance_id"] = unity_instance
|
|
else:
|
|
inferred = await infer_single_instance_id(ctx)
|
|
if inferred:
|
|
unity_section["instance_id"] = inferred
|
|
|
|
# External change detection (server-side): compute per instance based on project root path.
|
|
try:
|
|
instance_id = unity_section.get("instance_id")
|
|
if isinstance(instance_id, str) and instance_id.strip():
|
|
from services.resources.project_info import get_project_info
|
|
|
|
proj_resp = await get_project_info(ctx)
|
|
proj = proj_resp.model_dump() if hasattr(
|
|
proj_resp, "model_dump") else proj_resp
|
|
proj_data = proj.get("data") if isinstance(proj, dict) else None
|
|
project_root = proj_data.get("projectRoot") if isinstance(
|
|
proj_data, dict) else None
|
|
if isinstance(project_root, str) and project_root.strip():
|
|
external_changes_scanner.set_project_root(
|
|
instance_id, project_root)
|
|
|
|
ext = external_changes_scanner.update_and_get(instance_id)
|
|
|
|
assets = state_v2.get("assets")
|
|
if not isinstance(assets, dict):
|
|
assets = {}
|
|
state_v2["assets"] = assets
|
|
assets["external_changes_dirty"] = bool(
|
|
ext.get("external_changes_dirty", False))
|
|
assets["external_changes_last_seen_unix_ms"] = ext.get(
|
|
"external_changes_last_seen_unix_ms")
|
|
assets["external_changes_dirty_since_unix_ms"] = ext.get(
|
|
"dirty_since_unix_ms")
|
|
assets["external_changes_last_cleared_unix_ms"] = ext.get(
|
|
"last_cleared_unix_ms")
|
|
except Exception:
|
|
pass
|
|
|
|
state_v2 = _enrich_advice_and_staleness(state_v2)
|
|
|
|
try:
|
|
if hasattr(EditorStateData, "model_validate"):
|
|
validated = EditorStateData.model_validate(state_v2)
|
|
else:
|
|
validated = EditorStateData.parse_obj(
|
|
state_v2) # type: ignore[attr-defined]
|
|
data = validated.model_dump() if hasattr(
|
|
validated, "model_dump") else validated.dict()
|
|
except Exception as e:
|
|
return MCPResponse(
|
|
success=False,
|
|
error="invalid_editor_state",
|
|
message=f"Editor state payload failed validation: {e}",
|
|
data={"raw": state_v2},
|
|
)
|
|
|
|
return MCPResponse(success=True, message="Retrieved editor state.", data=data)
|