ue5_interface.py

import os
import logging
import time
import json
from typing import Optional, Any
from functools import wraps
from middleware.special_agent import CLIENT as AGENT

logger = logging.getLogger("BloomPath.UE5Interface")

# Configuration
# We keep the Actor Path as a fallback, but we will primarily search by Tag
UE5_ACTOR_PATH = os.getenv("UE5_ACTOR_PATH", "/Game/Garden.Garden:PersistentLevel.BP_GrowerActor_C_UAID_E89C257B5B95AEB802_2047448423")
UE5_ACTOR_TAG = "GrowerActor" # Ensure the BP has this tag, or we find by class

PRIORITY_COLORS = {
    "Highest": {"R": 1.0, "G": 0.2, "B": 0.2},
    "High": {"R": 1.0, "G": 0.6, "B": 0.1},
    "Medium": {"R": 0.3, "G": 0.8, "B": 0.3},
    "Low": {"R": 0.4, "G": 0.6, "B": 0.4},
    "Lowest": {"R": 0.5, "G": 0.5, "B": 0.5},
    "Low": {"R": 0.4, "G": 0.6, "B": 0.4},
    "Lowest": {"R": 0.5, "G": 0.5, "B": 0.5},
}


class CommandBatcher:
    """Buffers UE5 commands to execute them in a single call."""

    def __init__(self):
        self._buffer: list[str] = []

    def add(self, command_script: str):
        """Add a python script snippet to the batch."""
        self._buffer.append(command_script)

    def flush(self) -> dict[str, Any]:
        """Execute all buffered commands as one script."""
        if not self._buffer:
            return {"status": "empty"}

        # Combine all scripts, ensuring imports are handled at the top if needed
        # For simplicity, we just concatenate. UE5 Python environment persists globals
        # but re-importing 'unreal' is cheap/safe.
        combined_script = "\n".join(self._buffer)
        
        try:
            # We wrap in a try-except block in the script itself to prevent one failure stopping others
            # or just let it fail fast. Here we let it run sequentially.
            logger.info(f"🚀 Executing batch of {len(self._buffer)} commands")
            output = AGENT.execute_python(combined_script)
            self._buffer.clear()
            return {"output": output}
        except Exception as e:
            logger.error(f"Batch execution failed: {e}")
            return {"error": str(e)}

    def clear(self):
        self._buffer.clear()


# Global batcher instance
BATCHER = CommandBatcher()

def retry_on_failure(max_retries: int = 3, delay: float = 1.0):
    """Decorator to retry a function on failure."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception: Optional[Exception] = None
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    logger.warning(f"Attempt {attempt + 1}/{max_retries} failed: {e}")
                    if attempt < max_retries - 1:
                        time.sleep(delay * (attempt + 1))
            logger.error(f"All {max_retries} attempts failed")
            raise last_exception
        return wrapper
    return decorator

@retry_on_failure()
def trigger_ue5_growth(
    branch_id: str,
    growth_type: str = "leaf",
    growth_modifier: float = 1.0,
    project_id: Optional[str] = None,
    is_bloom: bool = False
) -> dict[str, Any]:
    """
    Trigger growth visualization in UE5 using the python remote API.
    
    Args:
        branch_id: The issue identifier (e.g., KAN-123)
        growth_type: 'leaf', 'flower', 'branch', 'trunk', 'fruit'
        growth_modifier: Multiplier for growth amount based on priority
        project_id: The project zone to grow in
        is_bloom: If True, specifically bloom existing plant
    """
    logger.info(f"Triggering UE5 {'bloom' if is_bloom else 'growth'}: {branch_id} ({growth_type}) in {project_id}")
    
    c_r, c_g, c_b = TYPE_COLOR_MAP.get(growth_type, TYPE_COLOR_MAP["leaf"])
    
    script = f"""
import unreal
world = unreal.EditorLevelLibrary.get_editor_world()
actors = unreal.GameplayStatics.get_all_actors_with_tag(world, "{UE5_ACTOR_TAG}")
actor = actors[0] if actors else unreal.find_object(None, "{UE5_ACTOR_PATH}")

if actor:
    # Run script with kwargs (Validated names: target_branch_id, growth_type, growth_factor, color_r...)
    # Note: We construct the dictionary string manually for the embedded script
    # c_r, c_g, c_b are Python variables here, we insert their VALUES into the script string
    # Verified Positional Signature (6 args): 
    # (branch_id: str, r: float, g: float, b: float, project_id: str, growth_modifier: float)
    # Note: growth_type is implicit in the function name "Grow_Leaves"
    actor.call_method("Grow_Leaves", (
        "{branch_id}", 
        {c_r}, 
        {c_g}, 
        {c_b}, 
        "{project_id or ''}", 
        {growth_modifier}
    ))
    print("Success")
else:
    print("Error: Actor not found")
"""
    output = AGENT.execute_python(script)
    return {"output": output}

@retry_on_failure()
def trigger_ue5_shrink(branch_id: str) -> dict[str, Any]:
    logger.info(f"Triggering UE5 shrink: {branch_id}")
    script = f"""
import unreal
world = unreal.EditorLevelLibrary.get_editor_world()
actors = unreal.GameplayStatics.get_all_actors_with_tag(world, "{UE5_ACTOR_TAG}")
actor = actors[0] if actors else unreal.find_object(None, "{UE5_ACTOR_PATH}")
if actor:
    actor.call_method("Shrink_Leaves", ("{branch_id}",))
    print("Success")
"""
    return {"output": AGENT.execute_python(script)}

@retry_on_failure()
def trigger_ue5_thorns(branch_id: str, epic_key: Optional[str] = None) -> dict[str, Any]:
    logger.info(f"Triggering UE5 thorns: {branch_id}")
    script = f"""
import unreal
world = unreal.EditorLevelLibrary.get_editor_world()
actors = unreal.GameplayStatics.get_all_actors_with_tag(world, "{UE5_ACTOR_TAG}")
actor = actors[0] if actors else unreal.find_object(None, "{UE5_ACTOR_PATH}")
if actor:
    actor.call_method("Add_Thorns", ("{branch_id}", "{epic_key or ''}"))
    print("Success")
"""
    return {"output": AGENT.execute_python(script)}

@retry_on_failure()
def trigger_ue5_remove_thorns(branch_id: str) -> dict[str, Any]:
    logger.info(f"Triggering UE5 remove thorns: {branch_id}")
    script = f"""
import unreal
world = unreal.EditorLevelLibrary.get_editor_world()
actors = unreal.GameplayStatics.get_all_actors_with_tag(world, "{UE5_ACTOR_TAG}")
actor = actors[0] if actors else unreal.find_object(None, "{UE5_ACTOR_PATH}")
if actor:
    actor.call_method("Remove_Thorns", ("{branch_id}",))
    print("Success")
"""
    return {"output": AGENT.execute_python(script)}

@retry_on_failure()
def trigger_ue5_weather(weather: str) -> dict[str, Any]:
    logger.info(f"Setting UE5 weather: {weather}")
    script = f"""
import unreal
world = unreal.EditorLevelLibrary.get_editor_world()
actors = unreal.GameplayStatics.get_all_actors_with_tag(world, "{UE5_ACTOR_TAG}")
actor = actors[0] if actors else unreal.find_object(None, "{UE5_ACTOR_PATH}")
if actor:
    actor.SetWeather("{weather}")
"""
    return {"output": AGENT.execute_python(script)}

@retry_on_failure()
def trigger_ue5_time(progress: float) -> dict[str, Any]:
    logger.info(f"Setting UE5 time: {progress:.2%}")
    script = f"""
import unreal
world = unreal.EditorLevelLibrary.get_editor_world()
actors = unreal.GameplayStatics.get_all_actors_with_tag(world, "{UE5_ACTOR_TAG}")
actor = actors[0] if actors else unreal.find_object(None, "{UE5_ACTOR_PATH}")
if actor:
    actor.Set_Time_Of_Day({progress})
"""
    return {"output": AGENT.execute_python(script)}

@retry_on_failure()
def trigger_ue5_set_tag(actor_name: str, tag: str) -> dict[str, Any]:
    script = f"""
import unreal
world = unreal.EditorLevelLibrary.get_editor_world()
actors = unreal.GameplayStatics.get_all_actors_with_tag(world, "{UE5_ACTOR_TAG}")
actor = actors[0] if actors else unreal.find_object(None, "{UE5_ACTOR_PATH}")
if actor:
    actor.Set_Actor_Tag("{actor_name}", "{tag}")
"""
    return {"output": AGENT.execute_python(script)}

@retry_on_failure()
def trigger_ue5_spawn_interactive(item_class_name: str, semantic_target: str) -> dict[str, Any]:
    logger.info(f"✨ Spawning interactive {item_class_name} on {semantic_target}")
    script = f"""
import unreal
world = unreal.EditorLevelLibrary.get_editor_world()
actors = unreal.GameplayStatics.get_all_actors_with_tag(world, "{UE5_ACTOR_TAG}")
actor = actors[0] if actors else unreal.find_object(None, "{UE5_ACTOR_PATH}")
if actor:
    # UE5 Blueprint should implement Spawn_Interactive_Item(ItemClass, SemanticTarget)
    actor.call_method("Spawn_Interactive_Item", ("{item_class_name}", "{semantic_target}"))
    print("Success")
"""
    return {"output": AGENT.execute_python(script)}

# ── Avatar Animations (Social Layer) ────────────────────────────────

AVATAR_ANIMATIONS = {
    "celebrate": {"montage": "AM_Celebrate", "duration": 3.0},
    "frustrated": {"montage": "AM_Frustrated", "duration": 2.5},
    "confused":   {"montage": "AM_Confused",   "duration": 2.0},
    "working":    {"montage": "AM_Working",    "duration": 0.0},  # looping
    "relieved":   {"montage": "AM_Relieved",   "duration": 2.0},
    "idle":       {"montage": "AM_Idle",       "duration": 0.0},  # looping
}

@retry_on_failure()
def trigger_ue5_avatar_animation(
    user_id: str,
    animation_name: str,
    intensity: float = 1.0
) -> dict[str, Any]:
    """Play an animation on a gardener NPC avatar."""
    anim_data = AVATAR_ANIMATIONS.get(animation_name, AVATAR_ANIMATIONS["idle"])
    montage = anim_data["montage"]
    logger.info(f"🎭 Avatar animation: {animation_name} ({montage}) for {user_id}")
    script = f"""
import unreal
world = unreal.EditorLevelLibrary.get_editor_world()
actors = unreal.GameplayStatics.get_all_actors_with_tag(world, "{UE5_ACTOR_TAG}")
actor = actors[0] if actors else unreal.find_object(None, "{UE5_ACTOR_PATH}")
if actor:
    actor.Play_Avatar_Animation("{user_id}", "{montage}", {intensity})
"""
    return {"output": AGENT.execute_python(script)}


@retry_on_failure()
def trigger_ue5_spawn_avatar(
    account_id: str,
    target_issue_id: str,
    display_name: str,
    avatar_url: str = ""
) -> dict[str, Any]:
    logger.info(f"👤 Spawning avatar for {display_name}")
    script = f"""
import unreal
world = unreal.EditorLevelLibrary.get_editor_world()
actors = unreal.GameplayStatics.get_all_actors_with_tag(world, "{UE5_ACTOR_TAG}")
actor = actors[0] if actors else unreal.find_object(None, "{UE5_ACTOR_PATH}")
if actor:
    # UE5 Blueprint should be updated to use Spawn_Avatar_At_Issue or similar method that takes the issue ID
    actor.Spawn_Avatar_At_Issue("{account_id}", "{display_name}", "{target_issue_id}", "{avatar_url}")
"""
    return {"output": AGENT.execute_python(script)}

@retry_on_failure()
def trigger_ue5_play_sound_2d(
    sound_name: str,
    volume_multiplier: float = 1.0,
    pitch_multiplier: float = 1.0
) -> dict[str, Any]:
    logger.info(f"🔊 Playing sound: {sound_name}")
    script = f"""
import unreal
world = unreal.EditorLevelLibrary.get_editor_world()
actors = unreal.GameplayStatics.get_all_actors_with_tag(world, "{UE5_ACTOR_TAG}")
actor = actors[0] if actors else unreal.find_object(None, "{UE5_ACTOR_PATH}")
if actor:
    # Play_Sound_2D(Name, Volume, Pitch)
    actor.Play_Sound_2D("{sound_name}", {volume_multiplier}, {pitch_multiplier})
"""
    return {"output": AGENT.execute_python(script)}

@retry_on_failure()
def trigger_phantom_warning(
    location_name: str,
    risk_level: float = 0.5
) -> dict[str, Any]:
    logger.info(f"👻 Spawning phantom at {location_name}")
    script = f"""
import unreal
world = unreal.EditorLevelLibrary.get_editor_world()
actors = unreal.GameplayStatics.get_all_actors_with_tag(world, "{UE5_ACTOR_TAG}")
actor = actors[0] if actors else unreal.find_object(None, "{UE5_ACTOR_PATH}")
if actor:
    actor.Spawn_Phantom_Hazard("{location_name}", {max(0.1, min(1.0, risk_level))})
"""
    return {"output": AGENT.execute_python(script)}

@retry_on_failure()
def trigger_ue5_load_level(file_path: str) -> dict[str, Any]:
    logger.info(f"🏗️ Loading generated level: {file_path}")
    # Convert windows slashes to forward slashes for UE blueprint compatibility
    ue_file_path = file_path.replace("\\", "/")
    script = f"""
import unreal
world = unreal.EditorLevelLibrary.get_editor_world()
actors = unreal.GameplayStatics.get_all_actors_with_tag(world, "{UE5_ACTOR_TAG}")
actor = actors[0] if actors else unreal.find_object(None, "{UE5_ACTOR_PATH}")
if actor:
    actor.Load_Generated_Level(r"{ue_file_path}")
"""
    return {"output": AGENT.execute_python(script)}

# Map semantic type to actor (Helper, pure logic)
def map_semantic_type_to_actor(semantic_type: str) -> Optional[str]:
    st = semantic_type.lower()
    if "path" in st or "ground" in st:
        return "Floor"
    elif "wall" in st:
        return "Wall_North"
    elif "water" in st:
        return "Pond_Surface"
    return None

VINE_STYLES = {
    "blocked_by": {"color": {"R": 0.8, "G": 0.1, "B": 0.1}, "thickness": 0.15, "has_thorns": True, "animation": "pulse_warning"},
    "blocks": {"color": {"R": 0.8, "G": 0.4, "B": 0.1}, "thickness": 0.12, "has_thorns": True, "animation": "pulse_slow"},
    "relates_to": {"color": {"R": 0.3, "G": 0.7, "B": 0.3}, "thickness": 0.08, "has_thorns": False, "animation": "none"},
    "parent": {"color": {"R": 0.4, "G": 0.3, "B": 0.2}, "thickness": 0.2, "has_thorns": False, "animation": "none"},
    "child": {"color": {"R": 0.5, "G": 0.8, "B": 0.5}, "thickness": 0.06, "has_thorns": False, "animation": "none"}
}

@retry_on_failure()
def trigger_ue5_dependency_vine(from_id: str, to_id: str, relation_type: str = "relates_to") -> dict:
    style = VINE_STYLES.get(relation_type, VINE_STYLES["relates_to"])
    c = style["color"]
    vine_id = f"vine_{from_id}_{to_id}_{relation_type}"
    
    logger.info(f"🔗 Spawning vine {from_id}->{to_id}")
    script = f"""
import unreal
world = unreal.EditorLevelLibrary.get_editor_world()
actors = unreal.GameplayStatics.get_all_actors_with_tag(world, "{UE5_ACTOR_TAG}")
actor = actors[0] if actors else unreal.find_object(None, "{UE5_ACTOR_PATH}")
if actor:
    # Spawn_Dependency_Vine(Vine_ID, From, To, Type, Color_R, G, B, Thick, Thorns, Anim)
    actor.Spawn_Dependency_Vine("{vine_id}", "{from_id}", "{to_id}", "{relation_type}", 
        {c['R']}, {c['G']}, {c['B']}, {style['thickness']}, {style['has_thorns']}, "{style['animation']}")
"""
    return {"output": AGENT.execute_python(script)}

@retry_on_failure()
def trigger_ue5_remove_vine(from_id: str, to_id: str, relation_type: str = "relates_to") -> dict:
    vine_id = f"vine_{from_id}_{to_id}_{relation_type}"
    logger.info(f"✂️ Removing vine {vine_id}")
    script = f"""
import unreal
world = unreal.EditorLevelLibrary.get_editor_world()
actors = unreal.GameplayStatics.get_all_actors_with_tag(world, "{UE5_ACTOR_TAG}")
actor = actors[0] if actors else unreal.find_object(None, "{UE5_ACTOR_PATH}")
if actor:
    actor.Remove_Dependency_Vine("{vine_id}")
"""
    return {"output": AGENT.execute_python(script)}

@retry_on_failure()
def trigger_ue5_sync_all_vines(dependencies: list[dict]) -> dict[str, Any]:
    """
    Sync all vines in one go using the new Batcher.
    
    This replaces the old loop-based approach.
    """
    logger.info(f"Syncing {len(dependencies)} dependency vines...")
    
    BATCHER.clear()
    
    # Common imports for the batch script
    BATCHER.add("import unreal")
    BATCHER.add("world = unreal.EditorLevelLibrary.get_editor_world()")
    BATCHER.add(f"actors = unreal.GameplayStatics.get_all_actors_with_tag(world, '{UE5_ACTOR_TAG}')")
    BATCHER.add(f"actor = actors[0] if actors else unreal.find_object(None, '{UE5_ACTOR_PATH}')")
    
    for dep in dependencies:
        from_id = dep.get('from')
        to_id = dep.get('to')
        rtype = dep.get('type', 'relates_to')
        
        style = VINE_STYLES.get(rtype, VINE_STYLES["relates_to"])
        color = style["color"]
        c = color # shorter alias
        vine_id = f"vine_{from_id}_{to_id}_{rtype}"
        
        # We append the specific call to the batch script
        script = f"""
if actor:
    # Spawn_Dependency_Vine(Vine_ID, From, To, Type, Color_R, G, B, Thick, Thorns, Anim)
    actor.Spawn_Dependency_Vine("{vine_id}", "{from_id}", "{to_id}", "{rtype}", 
        {c['R']}, {c['G']}, {c['B']}, {style['thickness']}, {style['has_thorns']}, "{style['animation']}")
"""
        BATCHER.add(script)
        
    return BATCHER.flush()


@retry_on_failure()
def trigger_ue5_move_avatar(user_id: str, target_issue_id: str) -> dict[str, Any]:
    logger.info(f"👤 Moving avatar {user_id} to {target_issue_id}")
    script = f"""
import unreal
world = unreal.EditorLevelLibrary.get_editor_world()
actors = unreal.GameplayStatics.get_all_actors_with_tag(world, "{UE5_ACTOR_TAG}")
actor = actors[0] if actors else unreal.find_object(None, "{UE5_ACTOR_PATH}")
if actor:
    actor.Move_Avatar("{user_id}", "{target_issue_id}")
    print("Success")
"""
    return {"output": AGENT.execute_python(script)}


@retry_on_failure()
def trigger_ue5_remove_avatar(user_id: str) -> dict[str, Any]:
    logger.info(f"👤 Removing avatar {user_id}")
    script = f"""
import unreal
world = unreal.EditorLevelLibrary.get_editor_world()
actors = unreal.GameplayStatics.get_all_actors_with_tag(world, "{UE5_ACTOR_TAG}")
actor = actors[0] if actors else unreal.find_object(None, "{UE5_ACTOR_PATH}")
if actor:
    actor.Remove_Avatar("{user_id}")
    print("Success")
"""
    return {"output": AGENT.execute_python(script)}


@retry_on_failure()
def trigger_ue5_ambience(intensity: float) -> dict[str, Any]:
    logger.info(f"🔊 Setting ambience intensity: {intensity:.2f}")
    # Clamp between 0.0 and 1.0
    val = max(0.0, min(1.0, intensity))
    script = f"""
import unreal
world = unreal.EditorLevelLibrary.get_editor_world()
actors = unreal.GameplayStatics.get_all_actors_with_tag(world, "{UE5_ACTOR_TAG}")
actor = actors[0] if actors else unreal.find_object(None, "{UE5_ACTOR_PATH}")
if actor:
    actor.Set_Ambience_Intensity({val})
"""
    return {"output": AGENT.execute_python(script)}


@retry_on_failure()
def trigger_ue5_storm_cloud(zone_id: str, intensity: float) -> dict[str, Any]:
    """Spawn or update a localized storm cloud over a specific dependency zone."""
    val = max(0.0, min(1.0, intensity))
    logger.info(f"⛈️ Spawning localized storm cloud over zone {zone_id} (intensity: {val:.2f})")
    script = f"""
import unreal
world = unreal.EditorLevelLibrary.get_editor_world()
actors = unreal.GameplayStatics.get_all_actors_with_tag(world, "{UE5_ACTOR_TAG}")
actor = actors[0] if actors else unreal.find_object(None, "{UE5_ACTOR_PATH}")
if actor:
    actor.call_method("Spawn_Storm_Cloud", ("{zone_id}", {val}))
"""
    return {"output": AGENT.execute_python(script)}


@retry_on_failure()
def trigger_ue5_reset_garden() -> dict[str, Any]:
    logger.info("Executed: Reset_Garden (Clear All)")
    script = f"""
import unreal
world = unreal.EditorLevelLibrary.get_editor_world()
actors = unreal.GameplayStatics.get_all_actors_with_tag(world, "{UE5_ACTOR_TAG}")
actor = actors[0] if actors else unreal.find_object(None, "{UE5_ACTOR_PATH}")
if actor:
    # Assuming Blueprint has a 'Reset_Garden' function that clears arrays/actors
    actor.call_method("Reset_Garden", ())
    print("Success")
"""
    return {"output": AGENT.execute_python(script)}


@retry_on_failure()
def trigger_ue5_scrub_timeline(reconstructed_tickets: list) -> dict[str, Any]:
    """
    Clears the UE5 garden and rebuilds it instantly using the BATCHER based on a historical list of tickets.
    """
    logger.info(f"⏪ Scrubbing timeline: rebuilding {len(reconstructed_tickets)} historical elements...")
    
    # First, let's reset the garden synchronously so it's clean before we batch spawn
    trigger_ue5_reset_garden()
    
    BATCHER.clear()
    BATCHER.add("import unreal")
    BATCHER.add("world = unreal.EditorLevelLibrary.get_editor_world()")
    BATCHER.add(f"actors = unreal.GameplayStatics.get_all_actors_with_tag(world, '{UE5_ACTOR_TAG}')")
    BATCHER.add(f"actor = actors[0] if actors else unreal.find_object(None, '{UE5_ACTOR_PATH}')")
    BATCHER.add("if actor:")
    
    for ticket in reconstructed_tickets:
        from middleware.models.ticket import IssueStatus, IssueType
        
        # We need to map the raw ticket data to the appropriate UE5 Spawn commands
        # Re-using the logic from core.py
        growth_type_map = {
            IssueType.EPIC: "trunk",
            IssueType.FEATURE: "branch",
            IssueType.BUG: "flower",
            IssueType.TASK: "leaf",
            IssueType.CHORE: "bud",
        }
        growth_type = growth_type_map.get(ticket.issue_type, "leaf")
        
        priority_map = {5: 2.0, 4: 1.5, 3: 1.0, 2: 0.75, 1: 0.5}
        growth_mod = priority_map.get(ticket.priority, 1.0)
        
        pid = ticket.project_id if ticket.project_id else ""
        
        # Did it bloom historically?
        is_bloom = ticket.status == IssueStatus.DONE
        
        # Did it have thorns historically?
        # A ticket has thorns if the last event was "blocked".
        # If it was completed, or just created, it has no thorns.
        # This is a bit of an assumption based on the timeline row.
        last_event = getattr(ticket, '_last_historical_event', 'updated')
        
        # Sub-commands added to batch buffer with correct indentation
        BATCHER.add(f"    actor.Trigger_Growth('{ticket.id}', '{growth_type}', {growth_mod}, '{pid}', {str(is_bloom).lower()})")
        
        if last_event == 'blocked':
            BATCHER.add(f"    actor.Add_Thorns('{ticket.id}', '{ticket.parent_id or ''}')")
            
    return BATCHER.flush()


# ── Ghost Garden (Dreaming Engine) ──────────────────────────────────


@retry_on_failure()
def trigger_ue5_ghost_overlay(scenario_id: str, intensity: float = 0.5) -> dict[str, Any]:
    """Apply a semi-transparent ghost overlay for a dreaming scenario."""
    intensity = max(0.0, min(1.0, intensity))
    logger.info(f"Executed: Ghost_Overlay (scenario={scenario_id}, intensity={intensity})")
    script = f"""
import unreal
world = unreal.EditorLevelLibrary.get_editor_world()
actors = unreal.GameplayStatics.get_all_actors_with_tag(world, "{UE5_ACTOR_TAG}")
actor = actors[0] if actors else unreal.find_object(None, "{UE5_ACTOR_PATH}")
if actor:
    actor.call_method("Set_Ghost_Overlay", ("{scenario_id}", {intensity}))
    print("Success")
"""
    return {"output": AGENT.execute_python(script)}


@retry_on_failure()
def trigger_ue5_ghost_growth(
    branch_id: str,
    growth_type: str = "leaf",
    opacity: float = 0.5
) -> dict[str, Any]:
    """Spawn a ghosted (semi-transparent) plant for a simulation."""
    opacity = max(0.0, min(1.0, opacity))
    logger.info(f"Executed: Ghost_Grow ({branch_id}, type={growth_type}, opacity={opacity})")
    script = f"""
import unreal
world = unreal.EditorLevelLibrary.get_editor_world()
actors = unreal.GameplayStatics.get_all_actors_with_tag(world, "{UE5_ACTOR_TAG}")
actor = actors[0] if actors else unreal.find_object(None, "{UE5_ACTOR_PATH}")
if actor:
    actor.call_method("Ghost_Grow", ("{branch_id}", "{growth_type}", {opacity}))
    print("Success")
"""
    return {"output": AGENT.execute_python(script)}


@retry_on_failure()
def trigger_ue5_clear_ghosts() -> dict[str, Any]:
    """Remove all ghost overlays from the garden."""
    logger.info("Executed: Clear_Ghosts (Remove All)")
    script = f"""
import unreal
world = unreal.EditorLevelLibrary.get_editor_world()
actors = unreal.GameplayStatics.get_all_actors_with_tag(world, "{UE5_ACTOR_TAG}")
actor = actors[0] if actors else unreal.find_object(None, "{UE5_ACTOR_PATH}")
if actor:
    actor.call_method("Clear_Ghosts", ())
    print("Success")
"""
    return {"output": AGENT.execute_python(script)}