unity-mcp/UnityMcpBridge/UnityMcpServer~/src/tools/manage_script_edits.py

from mcp.server.fastmcp import FastMCP, Context
from typing import Dict, Any, List, Tuple
import base64
import re
import os
from unity_connection import send_command_with_retry


def _apply_edits_locally(original_text: str, edits: List[Dict[str, Any]]) -> str:
    text = original_text
    for edit in edits or []:
        op = (
            (edit.get("op")
             or edit.get("operation")
             or edit.get("type")
             or edit.get("mode")
             or "")
            .strip()
            .lower()
        )

        if not op:
            allowed = "anchor_insert, prepend, append, replace_range, regex_replace"
            raise RuntimeError(
                f"op is required; allowed: {allowed}. Use 'op' (aliases accepted: type/mode/operation)."
            )

        if op == "prepend":
            prepend_text = edit.get("text", "")
            text = (prepend_text if prepend_text.endswith("\n") else prepend_text + "\n") + text
        elif op == "append":
            append_text = edit.get("text", "")
            if not text.endswith("\n"):
                text += "\n"
            text += append_text
            if not text.endswith("\n"):
                text += "\n"
        elif op == "anchor_insert":
            anchor = edit.get("anchor", "")
            position = (edit.get("position") or "before").lower()
            insert_text = edit.get("text", "")
            flags = re.MULTILINE | (re.IGNORECASE if edit.get("ignore_case") else 0)
            m = re.search(anchor, text, flags)
            if not m:
                if edit.get("allow_noop", True):
                    continue
                raise RuntimeError(f"anchor not found: {anchor}")
            idx = m.start() if position == "before" else m.end()
            text = text[:idx] + insert_text + text[idx:]
        elif op == "replace_range":
            start_line = int(edit.get("startLine", 1))
            start_col  = int(edit.get("startCol", 1))
            end_line   = int(edit.get("endLine", start_line))
            end_col    = int(edit.get("endCol", 1))
            replacement = edit.get("text", "")
            lines = text.splitlines(keepends=True)
            max_line = len(lines) + 1  # 1-based, exclusive end
            if (start_line < 1 or end_line < start_line or end_line > max_line
                    or start_col < 1 or end_col < 1):
                raise RuntimeError("replace_range out of bounds")
            def index_of(line: int, col: int) -> int:
                if line <= len(lines):
                    return sum(len(l) for l in lines[: line - 1]) + (col - 1)
                return sum(len(l) for l in lines)
            a = index_of(start_line, start_col)
            b = index_of(end_line, end_col)
            text = text[:a] + replacement + text[b:]
        elif op == "regex_replace":
            pattern = edit.get("pattern", "")
            repl = edit.get("replacement", "")
            # Translate $n backrefs (our input) to Python \g<n>
            repl_py = re.sub(r"\$(\d+)", r"\\g<\1>", repl)
            count = int(edit.get("count", 0))  # 0 = replace all
            flags = re.MULTILINE
            if edit.get("ignore_case"):
                flags |= re.IGNORECASE
            text = re.sub(pattern, repl_py, text, count=count, flags=flags)
        else:
            allowed = "anchor_insert, prepend, append, replace_range, regex_replace"
            raise RuntimeError(f"unknown edit op: {op}; allowed: {allowed}. Use 'op' (aliases accepted: type/mode/operation).")
    return text


def _trigger_sentinel_async() -> None:
    """Fire the Unity menu flip on a short-lived background thread.

    This avoids blocking the current request or getting stuck during domain reloads
    (socket reconnects) when the Editor recompiles.
    """
    try:
        import threading, time

        def _flip():
            try:
                import json, glob, os
                # Small delay so write flushes; prefer early flip to avoid editor-focus second reload
                time.sleep(0.1)
                try:
                    files = sorted(glob.glob(os.path.expanduser("~/.unity-mcp/unity-mcp-status-*.json")), key=os.path.getmtime, reverse=True)
                    if files:
                        with open(files[0], "r") as f:
                            st = json.loads(f.read())
                            if st.get("reloading"):
                                return
                except Exception:
                    pass

            except Exception:
                pass

        threading.Thread(target=_flip, daemon=True).start()
    except Exception:
        pass

def _infer_class_name(script_name: str) -> str:
    # Default to script name as class name (common Unity pattern)
    return (script_name or "").strip()


def _extract_code_after(keyword: str, request: str) -> str:
    # Deprecated with NL removal; retained as no-op for compatibility
    idx = request.lower().find(keyword)
    if idx >= 0:
        return request[idx + len(keyword):].strip()
    return ""
def _is_structurally_balanced(text: str) -> bool:
    """Lightweight delimiter balance check for braces/paren/brackets.
    Not a full parser; used to preflight destructive regex deletes.
    """
    brace = paren = bracket = 0
    in_str = in_chr = False
    esc = False
    i = 0
    n = len(text)
    while i < n:
        c = text[i]
        nxt = text[i+1] if i+1 < n else ''
        if in_str:
            if not esc and c == '"':
                in_str = False
            esc = (not esc and c == '\\')
            i += 1
            continue
        if in_chr:
            if not esc and c == "'":
                in_chr = False
            esc = (not esc and c == '\\')
            i += 1
            continue
        # comments
        if c == '/' and nxt == '/':
            # skip to EOL
            i = text.find('\n', i)
            if i == -1:
                break
            i += 1
            continue
        if c == '/' and nxt == '*':
            j = text.find('*/', i+2)
            i = (j + 2) if j != -1 else n
            continue
        if c == '"':
            in_str = True; esc = False; i += 1; continue
        if c == "'":
            in_chr = True; esc = False; i += 1; continue
        if c == '{': brace += 1
        elif c == '}': brace -= 1
        elif c == '(': paren += 1
        elif c == ')': paren -= 1
        elif c == '[': bracket += 1
        elif c == ']': bracket -= 1
        if brace < 0 or paren < 0 or bracket < 0:
            return False
        i += 1
    return brace == 0 and paren == 0 and bracket == 0



def _normalize_script_locator(name: str, path: str) -> Tuple[str, str]:
    """Best-effort normalization of script "name" and "path".

    Accepts any of:
    - name = "SmartReach", path = "Assets/Scripts/Interaction"
    - name = "SmartReach.cs", path = "Assets/Scripts/Interaction"
    - name = "Assets/Scripts/Interaction/SmartReach.cs", path = ""
    - path = "Assets/Scripts/Interaction/SmartReach.cs" (name empty)
    - name or path using uri prefixes: unity://path/..., file://...
    - accidental duplicates like "Assets/.../SmartReach.cs/SmartReach.cs"

    Returns (name_without_extension, directory_path_under_Assets).
    """
    n = (name or "").strip()
    p = (path or "").strip()

    def strip_prefix(s: str) -> str:
        if s.startswith("unity://path/"):
            return s[len("unity://path/"):]
        if s.startswith("file://"):
            return s[len("file://"):]
        return s

    def collapse_duplicate_tail(s: str) -> str:
        # Collapse trailing "/X.cs/X.cs" to "/X.cs"
        parts = s.split("/")
        if len(parts) >= 2 and parts[-1] == parts[-2]:
            parts = parts[:-1]
        return "/".join(parts)

    # Prefer a full path if provided in either field
    candidate = ""
    for v in (n, p):
        v2 = strip_prefix(v)
        if v2.endswith(".cs") or v2.startswith("Assets/"):
            candidate = v2
            break

    if candidate:
        candidate = collapse_duplicate_tail(candidate)
        # If a directory was passed in path and file in name, join them
        if not candidate.endswith(".cs") and n.endswith(".cs"):
            v2 = strip_prefix(n)
            candidate = (candidate.rstrip("/") + "/" + v2.split("/")[-1])
        if candidate.endswith(".cs"):
            parts = candidate.split("/")
            file_name = parts[-1]
            dir_path = "/".join(parts[:-1]) if len(parts) > 1 else "Assets"
            base = file_name[:-3] if file_name.lower().endswith(".cs") else file_name
            return base, dir_path

    # Fall back: remove extension from name if present and return given path
    base_name = n[:-3] if n.lower().endswith(".cs") else n
    return base_name, (p or "Assets")


def _with_norm(resp: Dict[str, Any] | Any, edits: List[Dict[str, Any]], routing: str | None = None) -> Dict[str, Any] | Any:
    if not isinstance(resp, dict):
        return resp
    data = resp.setdefault("data", {})
    data.setdefault("normalizedEdits", edits)
    if routing:
        data["routing"] = routing
    return resp


def _err(code: str, message: str, *, expected: Dict[str, Any] | None = None, rewrite: Dict[str, Any] | None = None,
         normalized: List[Dict[str, Any]] | None = None, routing: str | None = None, extra: Dict[str, Any] | None = None) -> Dict[str, Any]:
    payload: Dict[str, Any] = {"success": False, "code": code, "message": message}
    data: Dict[str, Any] = {}
    if expected:
        data["expected"] = expected
    if rewrite:
        data["rewrite_suggestion"] = rewrite
    if normalized is not None:
        data["normalizedEdits"] = normalized
    if routing:
        data["routing"] = routing
    if extra:
        data.update(extra)
    if data:
        payload["data"] = data
    return payload

# Natural-language parsing removed; clients should send structured edits.


def register_manage_script_edits_tools(mcp: FastMCP):
    @mcp.tool(description=(
        "Structured C# edits (methods/classes) with safer boundaries — prefer this over raw text.\n\n"
        "Best practices:\n"
        "- Prefer anchor_* ops for pattern-based insert/replace near stable markers\n"
        "- Use replace_method/delete_method for whole-method changes (keeps signatures balanced)\n"
        "- Avoid whole-file regex deletes; validators will guard unbalanced braces\n"
        "- For tail insertions, prefer anchor/regex_replace on final brace (class closing)\n"
        "- Pass options.validate='standard' for structural checks; 'relaxed' for interior-only edits\n\n"
        "Canonical fields (use these exact keys):\n"
        "- op: replace_method | insert_method | delete_method | anchor_insert | anchor_delete | anchor_replace\n"
        "- className: string (defaults to 'name' if omitted on method/class ops)\n"
        "- methodName: string (required for replace_method, delete_method)\n"
        "- replacement: string (required for replace_method, insert_method)\n"
        "- position: start | end | after | before (insert_method only)\n"
        "- afterMethodName / beforeMethodName: string (required when position='after'/'before')\n"
        "- anchor: regex string (for anchor_* ops)\n"
        "- text: string (for anchor_insert/anchor_replace)\n\n"
        "Do NOT use: new_method, anchor_method, content, newText (aliases accepted but normalized).\n\n"
        "Examples:\n"
        "1) Replace a method:\n"
        "{ 'name':'SmartReach','path':'Assets/Scripts/Interaction','edits':[\n"
        "  { 'op':'replace_method','className':'SmartReach','methodName':'HasTarget',\n"
        "    'replacement':'public bool HasTarget(){ return currentTarget!=null; }' }\n"
        "], 'options':{'validate':'standard','refresh':'immediate'} }\n\n"
        "2) Insert a method after another:\n"
        "{ 'name':'SmartReach','path':'Assets/Scripts/Interaction','edits':[\n"
        "  { 'op':'insert_method','className':'SmartReach','replacement':'public void PrintSeries(){ Debug.Log(seriesName); }',\n"
        "    'position':'after','afterMethodName':'GetCurrentTarget' }\n"
        "] }\n"
    ))
    def script_apply_edits(
        ctx: Context,
        name: str,
        path: str,
        edits: List[Dict[str, Any]],
        options: Dict[str, Any] | None = None,
        script_type: str = "MonoBehaviour",
        namespace: str = "",
    ) -> Dict[str, Any]:
        # Normalize locator first so downstream calls target the correct script file.
        name, path = _normalize_script_locator(name, path)

        # No NL path: clients must provide structured edits in 'edits'.

        # Normalize unsupported or aliased ops to known structured/text paths
        def _unwrap_and_alias(edit: Dict[str, Any]) -> Dict[str, Any]:
            # Unwrap single-key wrappers like {"replace_method": {...}}
            for wrapper_key in (
                "replace_method","insert_method","delete_method",
                "replace_class","delete_class",
                "anchor_insert","anchor_replace","anchor_delete",
            ):
                if wrapper_key in edit and isinstance(edit[wrapper_key], dict):
                    inner = dict(edit[wrapper_key])
                    inner["op"] = wrapper_key
                    edit = inner
                    break

            e = dict(edit)
            op = (e.get("op") or e.get("operation") or e.get("type") or e.get("mode") or "").strip().lower()
            if op:
                e["op"] = op

            # Common field aliases
            if "class_name" in e and "className" not in e:
                e["className"] = e.pop("class_name")
            if "class" in e and "className" not in e:
                e["className"] = e.pop("class")
            if "method_name" in e and "methodName" not in e:
                e["methodName"] = e.pop("method_name")
            # Some clients use a generic 'target' for method name
            if "target" in e and "methodName" not in e:
                e["methodName"] = e.pop("target")
            if "method" in e and "methodName" not in e:
                e["methodName"] = e.pop("method")
            if "new_content" in e and "replacement" not in e:
                e["replacement"] = e.pop("new_content")
            if "newMethod" in e and "replacement" not in e:
                e["replacement"] = e.pop("newMethod")
            if "new_method" in e and "replacement" not in e:
                e["replacement"] = e.pop("new_method")
            if "content" in e and "replacement" not in e:
                e["replacement"] = e.pop("content")
            if "after" in e and "afterMethodName" not in e:
                e["afterMethodName"] = e.pop("after")
            if "after_method" in e and "afterMethodName" not in e:
                e["afterMethodName"] = e.pop("after_method")
            if "before" in e and "beforeMethodName" not in e:
                e["beforeMethodName"] = e.pop("before")
            if "before_method" in e and "beforeMethodName" not in e:
                e["beforeMethodName"] = e.pop("before_method")
            # anchor_method → before/after based on position (default after)
            if "anchor_method" in e:
                anchor = e.pop("anchor_method")
                pos = (e.get("position") or "after").strip().lower()
                if pos == "before" and "beforeMethodName" not in e:
                    e["beforeMethodName"] = anchor
                elif "afterMethodName" not in e:
                    e["afterMethodName"] = anchor
            if "anchorText" in e and "anchor" not in e:
                e["anchor"] = e.pop("anchorText")
            if "pattern" in e and "anchor" not in e and e.get("op") and e["op"].startswith("anchor_"):
                e["anchor"] = e.pop("pattern")
            if "newText" in e and "text" not in e:
                e["text"] = e.pop("newText")

            # CI compatibility (T‑A/T‑E):
            # Accept method-anchored anchor_insert and upgrade to insert_method
            # Example incoming shape:
            #   {"op":"anchor_insert","afterMethodName":"GetCurrentTarget","text":"..."}
            if (
                e.get("op") == "anchor_insert"
                and not e.get("anchor")
                and (e.get("afterMethodName") or e.get("beforeMethodName"))
            ):
                e["op"] = "insert_method"
                if "replacement" not in e:
                    e["replacement"] = e.get("text", "")

            # LSP-like range edit -> replace_range
            if "range" in e and isinstance(e["range"], dict):
                rng = e.pop("range")
                start = rng.get("start", {})
                end = rng.get("end", {})
                # Convert 0-based to 1-based line/col
                e["op"] = "replace_range"
                e["startLine"] = int(start.get("line", 0)) + 1
                e["startCol"] = int(start.get("character", 0)) + 1
                e["endLine"] = int(end.get("line", 0)) + 1
                e["endCol"] = int(end.get("character", 0)) + 1
                if "newText" in edit and "text" not in e:
                    e["text"] = edit.get("newText", "")
            return e

        normalized_edits: List[Dict[str, Any]] = []
        for raw in edits or []:
            e = _unwrap_and_alias(raw)
            op = (e.get("op") or e.get("operation") or e.get("type") or e.get("mode") or "").strip().lower()

            # Default className to script name if missing on structured method/class ops
            if op in ("replace_class","delete_class","replace_method","delete_method","insert_method") and not e.get("className"):
                e["className"] = name

            # Map common aliases for text ops
            if op in ("text_replace",):
                e["op"] = "replace_range"
                normalized_edits.append(e)
                continue
            if op in ("regex_delete",):
                e["op"] = "regex_replace"
                e.setdefault("text", "")
                normalized_edits.append(e)
                continue
            if op == "regex_replace" and ("replacement" not in e):
                if "text" in e:
                    e["replacement"] = e.get("text", "")
                elif "insert" in e or "content" in e:
                    e["replacement"] = e.get("insert") or e.get("content") or ""
            if op == "anchor_insert" and not (e.get("text") or e.get("insert") or e.get("content") or e.get("replacement")):
                e["op"] = "anchor_delete"
                normalized_edits.append(e)
                continue
            normalized_edits.append(e)

        edits = normalized_edits
        normalized_for_echo = edits

        # Validate required fields and produce machine-parsable hints
        def error_with_hint(message: str, expected: Dict[str, Any], suggestion: Dict[str, Any]) -> Dict[str, Any]:
            return _err("missing_field", message, expected=expected, rewrite=suggestion, normalized=normalized_for_echo)

        for e in edits or []:
            op = e.get("op", "")
            if op == "replace_method":
                if not e.get("methodName"):
                    return error_with_hint(
                        "replace_method requires 'methodName'.",
                        {"op": "replace_method", "required": ["className", "methodName", "replacement"]},
                        {"edits[0].methodName": "HasTarget"}
                    )
                if not (e.get("replacement") or e.get("text")):
                    return error_with_hint(
                        "replace_method requires 'replacement' (inline or base64).",
                        {"op": "replace_method", "required": ["className", "methodName", "replacement"]},
                        {"edits[0].replacement": "public bool X(){ return true; }"}
                    )
            elif op == "insert_method":
                if not (e.get("replacement") or e.get("text")):
                    return error_with_hint(
                        "insert_method requires a non-empty 'replacement'.",
                        {"op": "insert_method", "required": ["className", "replacement"], "position": {"after_requires": "afterMethodName", "before_requires": "beforeMethodName"}},
                        {"edits[0].replacement": "public void PrintSeries(){ Debug.Log(\"1,2,3\"); }"}
                    )
                pos = (e.get("position") or "").lower()
                if pos == "after" and not e.get("afterMethodName"):
                    return error_with_hint(
                        "insert_method with position='after' requires 'afterMethodName'.",
                        {"op": "insert_method", "position": {"after_requires": "afterMethodName"}},
                        {"edits[0].afterMethodName": "GetCurrentTarget"}
                    )
                if pos == "before" and not e.get("beforeMethodName"):
                    return error_with_hint(
                        "insert_method with position='before' requires 'beforeMethodName'.",
                        {"op": "insert_method", "position": {"before_requires": "beforeMethodName"}},
                        {"edits[0].beforeMethodName": "GetCurrentTarget"}
                    )
            elif op == "delete_method":
                if not e.get("methodName"):
                    return error_with_hint(
                        "delete_method requires 'methodName'.",
                        {"op": "delete_method", "required": ["className", "methodName"]},
                        {"edits[0].methodName": "PrintSeries"}
                    )
            elif op in ("anchor_insert", "anchor_replace", "anchor_delete"):
                if not e.get("anchor"):
                    return error_with_hint(
                        f"{op} requires 'anchor' (regex).",
                        {"op": op, "required": ["anchor"]},
                        {"edits[0].anchor": "(?m)^\\s*public\\s+bool\\s+HasTarget\\s*\\("}
                    )
                if op in ("anchor_insert", "anchor_replace") and not (e.get("text") or e.get("replacement")):
                    return error_with_hint(
                        f"{op} requires 'text'.",
                        {"op": op, "required": ["anchor", "text"]},
                        {"edits[0].text": "/* comment */\n"}
                    )

        # Decide routing: structured vs text vs mixed
        STRUCT = {"replace_class","delete_class","replace_method","delete_method","insert_method","anchor_delete","anchor_replace","anchor_insert"}
        TEXT = {"prepend","append","replace_range","regex_replace"}
        ops_set = { (e.get("op") or "").lower() for e in edits or [] }
        all_struct = ops_set.issubset(STRUCT)
        all_text = ops_set.issubset(TEXT)
        mixed = not (all_struct or all_text)

        # If everything is structured (method/class/anchor ops), forward directly to Unity's structured editor.
        if all_struct:
            opts2 = dict(options or {})
            # For structured edits, prefer immediate refresh to avoid missed reloads when Editor is unfocused
            opts2.setdefault("refresh", "immediate")
            params_struct: Dict[str, Any] = {
                "action": "edit",
                "name": name,
                "path": path,
                "namespace": namespace,
                "scriptType": script_type,
                "edits": edits,
                "options": opts2,
            }
            resp_struct = send_command_with_retry("manage_script", params_struct)
            if isinstance(resp_struct, dict) and resp_struct.get("success"):
                # Optional: flip sentinel only if explicitly requested
                if (options or {}).get("force_sentinel_reload"):
                    try:
                        _trigger_sentinel_async()
                    except Exception:
                        pass
            return _with_norm(resp_struct if isinstance(resp_struct, dict) else {"success": False, "message": str(resp_struct)}, normalized_for_echo, routing="structured")

        # 1) read from Unity
        read_resp = send_command_with_retry("manage_script", {
            "action": "read",
            "name": name,
            "path": path,
            "namespace": namespace,
            "scriptType": script_type,
        })
        if not isinstance(read_resp, dict) or not read_resp.get("success"):
            return read_resp if isinstance(read_resp, dict) else {"success": False, "message": str(read_resp)}

        data = read_resp.get("data") or read_resp.get("result", {}).get("data") or {}
        contents = data.get("contents")
        if contents is None and data.get("contentsEncoded") and data.get("encodedContents"):
            contents = base64.b64decode(data["encodedContents"]).decode("utf-8")
        if contents is None:
            return {"success": False, "message": "No contents returned from Unity read."}

        # Optional preview/dry-run: apply locally and return diff without writing
        preview = bool((options or {}).get("preview"))

        # If we have a mixed batch (TEXT + STRUCT), apply text first with precondition, then structured
        if mixed:
            text_edits = [e for e in edits or [] if (e.get("op") or "").lower() in TEXT]
            struct_edits = [e for e in edits or [] if (e.get("op") or "").lower() in STRUCT]
            try:
                base_text = contents
                def line_col_from_index(idx: int) -> Tuple[int, int]:
                    line = base_text.count("\n", 0, idx) + 1
                    last_nl = base_text.rfind("\n", 0, idx)
                    col = (idx - (last_nl + 1)) + 1 if last_nl >= 0 else idx + 1
                    return line, col

                at_edits: List[Dict[str, Any]] = []
                import re as _re
                for e in text_edits:
                    opx = (e.get("op") or e.get("operation") or e.get("type") or e.get("mode") or "").strip().lower()
                    text_field = e.get("text") or e.get("insert") or e.get("content") or e.get("replacement") or ""
                    if opx == "anchor_insert":
                        anchor = e.get("anchor") or ""
                        position = (e.get("position") or "after").lower()
                        flags = _re.MULTILINE | (_re.IGNORECASE if e.get("ignore_case") else 0)
                        try:
                            regex_obj = _re.compile(anchor, flags)
                        except Exception as ex:
                            return _with_norm(_err("bad_regex", f"Invalid anchor regex: {ex}", normalized=normalized_for_echo, routing="mixed/text-first", extra={"hint": "Escape parentheses/braces or use a simpler anchor."}), normalized_for_echo, routing="mixed/text-first")
                        m = regex_obj.search(base_text)
                        if not m:
                            return _with_norm({"success": False, "code": "anchor_not_found", "message": f"anchor not found: {anchor}"}, normalized_for_echo, routing="mixed/text-first")
                        idx = m.start() if position == "before" else m.end()
                        # Normalize insertion to avoid jammed methods
                        text_field_norm = text_field
                        if not text_field_norm.startswith("\n"):
                            text_field_norm = "\n" + text_field_norm
                        if not text_field_norm.endswith("\n"):
                            text_field_norm = text_field_norm + "\n"
                        sl, sc = line_col_from_index(idx)
                        at_edits.append({"startLine": sl, "startCol": sc, "endLine": sl, "endCol": sc, "newText": text_field_norm})
                        # do not mutate base_text when building atomic spans
                    elif opx == "replace_range":
                        if all(k in e for k in ("startLine","startCol","endLine","endCol")):
                            at_edits.append({
                                "startLine": int(e.get("startLine", 1)),
                                "startCol": int(e.get("startCol", 1)),
                                "endLine": int(e.get("endLine", 1)),
                                "endCol": int(e.get("endCol", 1)),
                                "newText": text_field
                            })
                        else:
                            return _with_norm(_err("missing_field", "replace_range requires startLine/startCol/endLine/endCol", normalized=normalized_for_echo, routing="mixed/text-first"), normalized_for_echo, routing="mixed/text-first")
                    elif opx == "regex_replace":
                        pattern = e.get("pattern") or ""
                        try:
                            regex_obj = _re.compile(pattern, _re.MULTILINE | (_re.IGNORECASE if e.get("ignore_case") else 0))
                        except Exception as ex:
                            return _with_norm(_err("bad_regex", f"Invalid regex pattern: {ex}", normalized=normalized_for_echo, routing="mixed/text-first", extra={"hint": "Escape special chars or prefer structured delete for methods."}), normalized_for_echo, routing="mixed/text-first")
                        m = regex_obj.search(base_text)
                        if not m:
                            continue
                        # Expand $1, $2... in replacement using this match
                        def _expand_dollars(rep: str) -> str:
                            return _re.sub(r"\$(\d+)", lambda g: m.group(int(g.group(1))) or "", rep)
                        repl = _expand_dollars(text_field)
                        sl, sc = line_col_from_index(m.start())
                        el, ec = line_col_from_index(m.end())
                        at_edits.append({"startLine": sl, "startCol": sc, "endLine": el, "endCol": ec, "newText": repl})
                        # do not mutate base_text when building atomic spans
                    elif opx in ("prepend","append"):
                        if opx == "prepend":
                            sl, sc = 1, 1
                            at_edits.append({"startLine": sl, "startCol": sc, "endLine": sl, "endCol": sc, "newText": text_field})
                            # prepend can be applied atomically without local mutation
                        else:
                            # Insert at true EOF position (handles both \n and \r\n correctly)
                            eof_idx = len(base_text)
                            sl, sc = line_col_from_index(eof_idx)
                            new_text = ("\n" if not base_text.endswith("\n") else "") + text_field
                            at_edits.append({"startLine": sl, "startCol": sc, "endLine": sl, "endCol": sc, "newText": new_text})
                            # do not mutate base_text when building atomic spans
                    else:
                        return _with_norm(_err("unknown_op", f"Unsupported text edit op: {opx}", normalized=normalized_for_echo, routing="mixed/text-first"), normalized_for_echo, routing="mixed/text-first")

                import hashlib
                sha = hashlib.sha256(base_text.encode("utf-8")).hexdigest()
                if at_edits:
                    params_text: Dict[str, Any] = {
                        "action": "apply_text_edits",
                        "name": name,
                        "path": path,
                        "namespace": namespace,
                        "scriptType": script_type,
                        "edits": at_edits,
                        "precondition_sha256": sha,
                        "options": {"refresh": (options or {}).get("refresh", "debounced"), "validate": (options or {}).get("validate", "standard"), "applyMode": ("atomic" if len(at_edits) > 1 else (options or {}).get("applyMode", "sequential"))}
                    }
                    resp_text = send_command_with_retry("manage_script", params_text)
                    if not (isinstance(resp_text, dict) and resp_text.get("success")):
                        return _with_norm(resp_text if isinstance(resp_text, dict) else {"success": False, "message": str(resp_text)}, normalized_for_echo, routing="mixed/text-first")
                    # Successful text write; flip sentinel only if explicitly requested
                    if (options or {}).get("force_sentinel_reload"):
                        try:
                            _trigger_sentinel_async()
                        except Exception:
                            pass
            except Exception as e:
                return _with_norm({"success": False, "message": f"Text edit conversion failed: {e}"}, normalized_for_echo, routing="mixed/text-first")

            if struct_edits:
                opts2 = dict(options or {})
                # Prefer debounced background refresh unless explicitly overridden
                opts2.setdefault("refresh", "debounced")
                params_struct: Dict[str, Any] = {
                    "action": "edit",
                    "name": name,
                    "path": path,
                    "namespace": namespace,
                    "scriptType": script_type,
                    "edits": struct_edits,
                    "options": opts2
                }
                resp_struct = send_command_with_retry("manage_script", params_struct)
                if isinstance(resp_struct, dict) and resp_struct.get("success"):
                    if (options or {}).get("force_sentinel_reload"):
                        try:
                            _trigger_sentinel_async()
                        except Exception:
                            pass
                return _with_norm(resp_struct if isinstance(resp_struct, dict) else {"success": False, "message": str(resp_struct)}, normalized_for_echo, routing="mixed/text-first")

            return _with_norm({"success": True, "message": "Applied text edits (no structured ops)"}, normalized_for_echo, routing="mixed/text-first")

        # If the edits are text-ops, prefer sending them to Unity's apply_text_edits with precondition
        # so header guards and validation run on the C# side.
        # Supported conversions: anchor_insert, replace_range, regex_replace (first match only).
        text_ops = { (e.get("op") or e.get("operation") or e.get("type") or e.get("mode") or "").strip().lower() for e in (edits or []) }
        structured_kinds = {"replace_class","delete_class","replace_method","delete_method","insert_method","anchor_insert"}
        if not text_ops.issubset(structured_kinds):
            # Convert to apply_text_edits payload
            try:
                base_text = contents
                def line_col_from_index(idx: int) -> Tuple[int, int]:
                    # 1-based line/col against base buffer
                    line = base_text.count("\n", 0, idx) + 1
                    last_nl = base_text.rfind("\n", 0, idx)
                    col = (idx - (last_nl + 1)) + 1 if last_nl >= 0 else idx + 1
                    return line, col

                at_edits: List[Dict[str, Any]] = []
                import re as _re
                for e in edits or []:
                    op = (e.get("op") or e.get("operation") or e.get("type") or e.get("mode") or "").strip().lower()
                    # aliasing for text field
                    text_field = e.get("text") or e.get("insert") or e.get("content") or ""
                    if op == "anchor_insert":
                        anchor = e.get("anchor") or ""
                        position = (e.get("position") or "after").lower()
                        # Early regex compile with helpful errors, honoring ignore_case
                        try:
                            flags = _re.MULTILINE | (_re.IGNORECASE if e.get("ignore_case") else 0)
                            regex_obj = _re.compile(anchor, flags)
                        except Exception as ex:
                            return _with_norm(_err("bad_regex", f"Invalid anchor regex: {ex}", normalized=normalized_for_echo, routing="text", extra={"hint": "Escape parentheses/braces or use a simpler anchor."}), normalized_for_echo, routing="text")
                        m = regex_obj.search(base_text)
                        if not m:
                            return _with_norm({"success": False, "code": "anchor_not_found", "message": f"anchor not found: {anchor}"}, normalized_for_echo, routing="text")
                        idx = m.start() if position == "before" else m.end()
                        # Normalize insertion newlines
                        if text_field and not text_field.startswith("\n"):
                            text_field = "\n" + text_field
                        if text_field and not text_field.endswith("\n"):
                            text_field = text_field + "\n"
                        sl, sc = line_col_from_index(idx)
                        at_edits.append({
                            "startLine": sl,
                            "startCol": sc,
                            "endLine": sl,
                            "endCol": sc,
                            "newText": text_field or ""
                        })
                        # Do not mutate base buffer when building an atomic batch
                    elif op == "replace_range":
                        # Directly forward if already in line/col form
                        if "startLine" in e:
                            at_edits.append({
                                "startLine": int(e.get("startLine", 1)),
                                "startCol": int(e.get("startCol", 1)),
                                "endLine": int(e.get("endLine", 1)),
                                "endCol": int(e.get("endCol", 1)),
                                "newText": text_field
                            })
                        else:
                            # If only indices provided, skip (we don't support index-based here)
                            return _with_norm({"success": False, "code": "missing_field", "message": "replace_range requires startLine/startCol/endLine/endCol"}, normalized_for_echo, routing="text")
                    elif op == "regex_replace":
                        pattern = e.get("pattern") or ""
                        repl = text_field
                        flags = _re.MULTILINE | (_re.IGNORECASE if e.get("ignore_case") else 0)
                        # Early compile for clearer error messages
                        try:
                            regex_obj = _re.compile(pattern, flags)
                        except Exception as ex:
                            return _with_norm(_err("bad_regex", f"Invalid regex pattern: {ex}", normalized=normalized_for_echo, routing="text", extra={"hint": "Escape special chars or prefer structured delete for methods."}), normalized_for_echo, routing="text")
                        m = regex_obj.search(base_text)
                        if not m:
                            continue
                        # Expand $1, $2... backrefs in replacement using the first match (consistent with mixed-path behavior)
                        def _expand_dollars(rep: str) -> str:
                            return _re.sub(r"\$(\d+)", lambda g: m.group(int(g.group(1))) or "", rep)
                        repl_expanded = _expand_dollars(repl)
                        # Preview structural balance after replacement; refuse destructive deletes
                        preview = base_text[:m.start()] + repl_expanded + base_text[m.end():]
                        if not _is_structurally_balanced(preview):
                            return _with_norm(_err("validation_failed", "regex_replace would unbalance braces/parentheses; prefer delete_method",
                                                   normalized=normalized_for_echo, routing="text",
                                                   extra={"status": "validation_failed", "hint": "Use script_apply_edits delete_method for method removal"}), normalized_for_echo, routing="text")
                        sl, sc = line_col_from_index(m.start())
                        el, ec = line_col_from_index(m.end())
                        at_edits.append({
                            "startLine": sl,
                            "startCol": sc,
                            "endLine": el,
                            "endCol": ec,
                            "newText": repl_expanded
                        })
                        # Do not mutate base buffer when building an atomic batch
                    else:
                        return _with_norm({"success": False, "code": "unsupported_op", "message": f"Unsupported text edit op for server-side apply_text_edits: {op}"}, normalized_for_echo, routing="text")

                if not at_edits:
                    return _with_norm({"success": False, "code": "no_spans", "message": "No applicable text edit spans computed (anchor not found or zero-length)."}, normalized_for_echo, routing="text")

                # Send to Unity with precondition SHA to enforce guards and immediate refresh
                import hashlib
                sha = hashlib.sha256(base_text.encode("utf-8")).hexdigest()
                params: Dict[str, Any] = {
                    "action": "apply_text_edits",
                    "name": name,
                    "path": path,
                    "namespace": namespace,
                    "scriptType": script_type,
                    "edits": at_edits,
                    "precondition_sha256": sha,
                    "options": {
                        "refresh": (options or {}).get("refresh", "debounced"),
                        "validate": (options or {}).get("validate", "standard"),
                        "applyMode": ("atomic" if len(at_edits) > 1 else (options or {}).get("applyMode", "sequential"))
                    }
                }
                resp = send_command_with_retry("manage_script", params)
                if isinstance(resp, dict) and resp.get("success"):
                    if (options or {}).get("force_sentinel_reload"):
                        try:
                            _trigger_sentinel_async()
                        except Exception:
                            pass
                return _with_norm(
                    resp if isinstance(resp, dict) else {"success": False, "message": str(resp)},
                    normalized_for_echo,
                    routing="text"
                )
            except Exception as e:
                return _with_norm({"success": False, "code": "conversion_failed", "message": f"Edit conversion failed: {e}"}, normalized_for_echo, routing="text")

        # For regex_replace, honor preview consistently: if preview=true, always return diff without writing.
        # If confirm=false (default) and preview not requested, return diff and instruct confirm=true to apply.
        if "regex_replace" in text_ops and (preview or not (options or {}).get("confirm")):
            try:
                preview_text = _apply_edits_locally(contents, edits)
                import difflib
                diff = list(difflib.unified_diff(contents.splitlines(), preview_text.splitlines(), fromfile="before", tofile="after", n=2))
                if len(diff) > 800:
                    diff = diff[:800] + ["... (diff truncated) ..."]
                if preview:
                    return {"success": True, "message": "Preview only (no write)", "data": {"diff": "\n".join(diff), "normalizedEdits": normalized_for_echo}}
                return _with_norm({"success": False, "message": "Preview diff; set options.confirm=true to apply.", "data": {"diff": "\n".join(diff)}}, normalized_for_echo, routing="text")
            except Exception as e:
                return _with_norm({"success": False, "code": "preview_failed", "message": f"Preview failed: {e}"}, normalized_for_echo, routing="text")
        # 2) apply edits locally (only if not text-ops)
        try:
            new_contents = _apply_edits_locally(contents, edits)
        except Exception as e:
            return {"success": False, "message": f"Edit application failed: {e}"}

        # Short-circuit no-op edits to avoid false "applied" reports downstream
        if new_contents == contents:
            return _with_norm({
                "success": True,
                "message": "No-op: contents unchanged",
                "data": {"no_op": True, "evidence": {"reason": "identical_content"}}
            }, normalized_for_echo, routing="text")

        if preview:
            # Produce a compact unified diff limited to small context
            import difflib
            a = contents.splitlines()
            b = new_contents.splitlines()
            diff = list(difflib.unified_diff(a, b, fromfile="before", tofile="after", n=3))
            # Limit diff size to keep responses small
            if len(diff) > 2000:
                diff = diff[:2000] + ["... (diff truncated) ..."]
            return {"success": True, "message": "Preview only (no write)", "data": {"diff": "\n".join(diff), "normalizedEdits": normalized_for_echo}}

        # 3) update to Unity
        # Default refresh/validate for natural usage on text path as well
        options = dict(options or {})
        options.setdefault("validate", "standard")
        options.setdefault("refresh", "debounced")

        import hashlib
        # Compute the SHA of the current file contents for the precondition
        old_lines = contents.splitlines(keepends=True)
        end_line = len(old_lines) + 1  # 1-based exclusive end
        sha = hashlib.sha256(contents.encode("utf-8")).hexdigest()

        # Apply a whole-file text edit rather than the deprecated 'update' action
        params = {
            "action": "apply_text_edits",
            "name": name,
            "path": path,
            "namespace": namespace,
            "scriptType": script_type,
            "edits": [
                {
                    "startLine": 1,
                    "startCol": 1,
                    "endLine": end_line,
                    "endCol": 1,
                    "newText": new_contents,
                }
            ],
            "precondition_sha256": sha,
            "options": options or {"validate": "standard", "refresh": "debounced"},
        }

        write_resp = send_command_with_retry("manage_script", params)
        if isinstance(write_resp, dict) and write_resp.get("success"):
            if (options or {}).get("force_sentinel_reload"):
                try:
                    _trigger_sentinel_async()
                except Exception:
                    pass
        return _with_norm(
            write_resp if isinstance(write_resp, dict) 
                      else {"success": False, "message": str(write_resp)},
            normalized_for_echo,
            routing="text",
        )




    # safe_script_edit removed to simplify API; clients should call script_apply_edits directly