import os import importlib import sys import logging import ast _deferred_reloads = [] def _verify_syntax(filepath: str, code: str) -> bool: ext = os.path.splitext(filepath)[1].lower() if ext == '.py': try: ast.parse(code) return True except SyntaxError as e: logging.error(f"Auto-Upgrade: AST Syntax Error in {filepath}: {e}") return False elif ext in ['.js', '.jsx', '.ts', '.tsx']: # Basic proxy check for unbalanced brackets brackets = {'(': ')', '{': '}', '[': ']'} stack = [] for char in code: if char in brackets: stack.append(char) elif char in brackets.values(): if not stack: logging.error(f"Auto-Upgrade: JS/TS Syntax Error in {filepath}: Unmatched '{char}'") return False if brackets[stack.pop()] != char: logging.error(f"Auto-Upgrade: JS/TS Syntax Error in {filepath}: Mismatched bracket near '{char}'") return False if stack: logging.error(f"Auto-Upgrade: JS/TS Syntax Error in {filepath}: Unclosed '{stack[-1]}'") return False return True return True async def write_and_hot_reload(filepath: str, code: str) -> bool: """ Writes the code block to the specified filepath and triggers a hot-reload. Returns True if successful, False if syntax failed. """ if not _verify_syntax(filepath, code): logging.error(f"Auto-Upgrade: Aborting write to {filepath} due to syntax error.") return False logging.info(f"Auto-Upgrade: Syntax verified. Writing modified code to {filepath}") # Ensure directory exists os.makedirs(os.path.dirname(os.path.abspath(filepath)), exist_ok=True) # Write code with open(filepath, 'w', encoding='utf-8') as f: f.write(code) from backend.ws.agent_ws import is_agent_busy if is_agent_busy(): logging.warning(f"Auto-Upgrade: Agent session is active. Deferring hot-reload for {filepath}") _deferred_reloads.append(filepath) return True await _trigger_hot_reload(filepath) return True async def process_deferred_reloads(): if not _deferred_reloads: return from backend.ws.agent_ws import is_agent_busy if not is_agent_busy(): logging.info(f"Auto-Upgrade: Agent session ended. Processing {len(_deferred_reloads)} deferred reloads.") for filepath in set(_deferred_reloads): await _trigger_hot_reload(filepath) _deferred_reloads.clear() async def _trigger_hot_reload(filepath: str): ext = os.path.splitext(filepath)[1].lower() if ext == '.py': _reload_python_module(filepath) elif ext in ['.tsx', '.ts', '.jsx', '.js']: await _reload_frontend_component(filepath) elif ext == '.kt': _log_android_recompile_required(filepath) else: logging.info(f"Auto-Upgrade: Saved {filepath}. Reload strategy not defined for {ext}.") def _reload_python_module(filepath: str): # Try to resolve filepath to a module name try: # Assumes filepath is relative to the backend root or absolute abs_path = os.path.abspath(filepath) backend_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) if abs_path.startswith(backend_dir): rel_path = os.path.relpath(abs_path, backend_dir) module_name = "backend." + rel_path.replace(os.sep, ".")[:-3] if module_name in sys.modules: logging.info(f"Auto-Upgrade: Hot-reloading Python module {module_name}...") importlib.reload(sys.modules[module_name]) else: logging.info(f"Auto-Upgrade: Module {module_name} not loaded in sys.modules yet.") except Exception as e: logging.error(f"Auto-Upgrade: Failed to hot-reload Python module for {filepath}: {e}") async def _reload_frontend_component(filepath: str): logging.info(f"Auto-Upgrade: Triggering frontend hot-reload for {filepath}") from backend.ws.agent_ws import ws_manager try: # Emits a signal to the frontend to reload (Vite handles real HMR, but we can nudge it) await ws_manager.broadcast({ "event": "frontend:reload_component", "payload": {"file": filepath} }) except Exception as e: logging.error(f"Auto-Upgrade: Failed to emit frontend reload WS event: {e}") def _log_android_recompile_required(filepath: str): logging.info(f"Auto-Upgrade: Android Kotlin code saved to {filepath}. Manual Gradle Rebuild/APK Install required for changes to take effect.")