# -*- coding: utf-8 -*- import os import py_compile import shutil target_file = r"C:\OptimusProjects\Optimus Developer\src\core\effector_runtime.py" with open(target_file, "r", encoding="utf-8") as f: content = f.read() start_marker = "# ==============================================================================\n# 4. MOUSE & KEYBOARD EFFECTOR\n# ==============================================================================" end_marker = "# ==============================================================================\n# 5. BROWSER EFFECTOR\n# ==============================================================================" if start_marker not in content or end_marker not in content: print("ERROR: markers not found") exit(1) pre_part = content[:content.index(start_marker)] post_part = content[content.index(end_marker):] new_class = """# ============================================================================== # 4. MOUSE & KEYBOARD EFFECTOR # ============================================================================== class MouseKeyboardEffector(BaseEffector): \"\"\" Physical mouse and keyboard effector. Executes native cursor motion, mouse clicks, scrolls, drags, text typing and hotkeys. Returns physical execution evidence. Avoids hardcoding application semantics. Enforces strict physical verification: PID_CREATED != APPLICATION_OPENED. OPEN_APPLICATION succeeds ONLY when a real, interactive, visible, and stable GUI window exists. Surgically guarantees TYPE_TEXT only sends keystrokes when the verified target window is in the foreground. \"\"\" ACTIONS = [ "MOUSE_MOVE", "MOUSE_CLICK", "MOUSE_DOUBLE_CLICK", "MOUSE_SCROLL", "MOUSE_DRAG", "TYPE_TEXT", "KEY_PRESS", "HOTKEY", "WINDOW_FOCUS", "OPEN_APPLICATION" ] _mission_hwnds: Dict[str, int] = {} _last_target_hwnd: int = 0 def __init__(self): super().__init__("mouse_keyboard", version="1.0.0", available_actions=self.ACTIONS) @classmethod def set_mission_hwnd(cls, mission_id: str, hwnd: int): if hwnd and int(hwnd) > 0: if mission_id: cls._mission_hwnds[str(mission_id)] = int(hwnd) cls._last_target_hwnd = int(hwnd) @classmethod def get_mission_hwnd(cls, mission_id: str) -> int: import ctypes user32 = ctypes.windll.user32 if mission_id and str(mission_id) in cls._mission_hwnds: h = cls._mission_hwnds[str(mission_id)] if h and user32.IsWindow(h) and user32.IsWindowVisible(h): return h if cls._last_target_hwnd: h = cls._last_target_hwnd if h and user32.IsWindow(h) and user32.IsWindowVisible(h): return h return 0 def execute(self, request: ActionRequest) -> ActionResult: self.last_seen = time.time() import ctypes from ctypes import wintypes import shutil import subprocess user32 = ctypes.windll.user32 kernel32 = ctypes.windll.kernel32 shell32 = ctypes.windll.shell32 aname = request.action_type.upper() payload = request.payload # Win64 type safety for user32 user32.IsWindow.argtypes = [wintypes.HWND] user32.IsWindow.restype = wintypes.BOOL user32.IsWindowVisible.argtypes = [wintypes.HWND] user32.IsWindowVisible.restype = wintypes.BOOL user32.GetWindowTextLengthW.argtypes = [wintypes.HWND] user32.GetWindowTextLengthW.restype = ctypes.c_int user32.GetWindowTextW.argtypes = [wintypes.HWND, wintypes.LPWSTR, ctypes.c_int] user32.GetWindowTextW.restype = ctypes.c_int user32.GetClassNameW.argtypes = [wintypes.HWND, wintypes.LPWSTR, ctypes.c_int] user32.GetClassNameW.restype = ctypes.c_int user32.GetWindowThreadProcessId.argtypes = [wintypes.HWND, ctypes.POINTER(wintypes.DWORD)] user32.GetWindowThreadProcessId.restype = wintypes.DWORD user32.GetForegroundWindow.argtypes = [] user32.GetForegroundWindow.restype = wintypes.HWND user32.SetForegroundWindow.argtypes = [wintypes.HWND] user32.SetForegroundWindow.restype = wintypes.BOOL user32.BringWindowToTop.argtypes = [wintypes.HWND] user32.BringWindowToTop.restype = wintypes.BOOL user32.ShowWindow.argtypes = [wintypes.HWND, ctypes.c_int] user32.ShowWindow.restype = wintypes.BOOL user32.SetFocus.argtypes = [wintypes.HWND] user32.SetFocus.restype = wintypes.HWND user32.AttachThreadInput.argtypes = [wintypes.DWORD, wintypes.DWORD, wintypes.BOOL] user32.AttachThreadInput.restype = wintypes.BOOL user32.GetAncestor.argtypes = [wintypes.HWND, ctypes.c_uint] user32.GetAncestor.restype = wintypes.HWND kernel32.GetCurrentThreadId.argtypes = [] kernel32.GetCurrentThreadId.restype = wintypes.DWORD # Win32 Mouse event constants MOUSEEVENTF_MOVE = 0x0001 MOUSEEVENTF_LEFTDOWN = 0x0002 MOUSEEVENTF_LEFTUP = 0x0004 MOUSEEVENTF_RIGHTDOWN = 0x0008 MOUSEEVENTF_RIGHTUP = 0x0010 MOUSEEVENTF_MIDDLEDOWN = 0x0020 MOUSEEVENTF_MIDDLEUP = 0x0040 MOUSEEVENTF_WHEEL = 0x0800 MOUSEEVENTF_ABSOLUTE = 0x8000 # Win32 Keyboard event constants KEYEVENTF_KEYUP = 0x0002 KEYEVENTF_UNICODE = 0x0004 VK_MAP = { "ENTER": 0x0D, "RETURN": 0x0D, "TAB": 0x09, "ESC": 0x1B, "ESCAPE": 0x1B, "SPACE": 0x20, "BACKSPACE": 0x08, "DELETE": 0x2E, "UP": 0x26, "DOWN": 0x28, "LEFT": 0x25, "RIGHT": 0x27, "CONTROL": 0x11, "CTRL": 0x11, "ALT": 0x12, "SHIFT": 0x10, "WIN": 0x5B, "F5": 0x74 } # Ignore non-interactive overlay / tool helper classes EXCLUDED_CLASSES = { "uac_inputindicatoroverlaywnd", "uac input indicator", "tooltips_class32", "msctls_statusbar32", "combobox" } def is_target_foreground(target_h: int) -> bool: if not target_h: return False fg = user32.GetForegroundWindow() or 0 if fg == target_h: return True if fg and user32.GetAncestor(fg, 2) == target_h: return True return False def activate_foreground(target_h: int, max_attempts: int = 5, retry_interval: float = 0.15) -> bool: if not target_h or not user32.IsWindow(target_h) or not user32.IsWindowVisible(target_h): return False if is_target_foreground(target_h): return True cur_thread = kernel32.GetCurrentThreadId() for _ in range(max_attempts): if is_target_foreground(target_h): return True fg_curr = user32.GetForegroundWindow() or 0 fg_thread = user32.GetWindowThreadProcessId(fg_curr, None) if fg_curr else 0 target_thread = user32.GetWindowThreadProcessId(target_h, None) attached_fg = False attached_target = False if fg_thread and fg_thread != cur_thread: attached_fg = bool(user32.AttachThreadInput(cur_thread, fg_thread, True)) if target_thread and target_thread != cur_thread: attached_target = bool(user32.AttachThreadInput(cur_thread, target_thread, True)) try: user32.keybd_event(0x12, 0, 0, 0) user32.ShowWindow(target_h, 9) user32.BringWindowToTop(target_h) user32.SetForegroundWindow(target_h) user32.SetFocus(target_h) user32.keybd_event(0x12, 0, 2, 0) finally: if attached_fg: user32.AttachThreadInput(cur_thread, fg_thread, False) if attached_target: user32.AttachThreadInput(cur_thread, target_thread, False) time.sleep(retry_interval) if is_target_foreground(target_h): return True return is_target_foreground(target_h) try: if aname == "MOUSE_MOVE": x = int(payload.get("x", 0)) y = int(payload.get("y", 0)) user32.SetCursorPos(x, y) evidence = {"action": "MOUSE_MOVE", "target": (x, y), "timestamp": time.time()} return ActionResult(action_id=request.action_id, mission_id=request.mission_id, status="SUCCESS", source="MOUSE_KEYBOARD_EFFECTOR", evidence=evidence) elif aname in ("MOUSE_CLICK", "CLICK"): x = payload.get("x") y = payload.get("y") if x is not None and y is not None: user32.SetCursorPos(int(x), int(y)) btn = (payload.get("button") or "left").lower() if btn == "right": user32.mouse_event(MOUSEEVENTF_RIGHTDOWN, 0, 0, 0, 0) time.sleep(0.05) user32.mouse_event(MOUSEEVENTF_RIGHTUP, 0, 0, 0, 0) else: user32.mouse_event(MOUSEEVENTF_LEFTDOWN, 0, 0, 0, 0) time.sleep(0.05) user32.mouse_event(MOUSEEVENTF_LEFTUP, 0, 0, 0, 0) evidence = {"action": "MOUSE_CLICK", "button": btn, "position": (x, y), "timestamp": time.time()} return ActionResult(action_id=request.action_id, mission_id=request.mission_id, status="SUCCESS", source="MOUSE_KEYBOARD_EFFECTOR", evidence=evidence) elif aname == "MOUSE_DOUBLE_CLICK": x = payload.get("x") y = payload.get("y") if x is not None and y is not None: user32.SetCursorPos(int(x), int(y)) user32.mouse_event(MOUSEEVENTF_LEFTDOWN, 0, 0, 0, 0) user32.mouse_event(MOUSEEVENTF_LEFTUP, 0, 0, 0, 0) time.sleep(0.08) user32.mouse_event(MOUSEEVENTF_LEFTDOWN, 0, 0, 0, 0) user32.mouse_event(MOUSEEVENTF_LEFTUP, 0, 0, 0, 0) evidence = {"action": "MOUSE_DOUBLE_CLICK", "position": (x, y), "timestamp": time.time()} return ActionResult(action_id=request.action_id, mission_id=request.mission_id, status="SUCCESS", source="MOUSE_KEYBOARD_EFFECTOR", evidence=evidence) elif aname == "MOUSE_SCROLL": delta = int(payload.get("delta", 120)) user32.mouse_event(MOUSEEVENTF_WHEEL, 0, 0, delta, 0) evidence = {"action": "MOUSE_SCROLL", "delta": delta, "timestamp": time.time()} return ActionResult(action_id=request.action_id, mission_id=request.mission_id, status="SUCCESS", source="MOUSE_KEYBOARD_EFFECTOR", evidence=evidence) elif aname == "MOUSE_DRAG": x1 = int(payload.get("x1", 0)) y1 = int(payload.get("y1", 0)) x2 = int(payload.get("x2", 0)) y2 = int(payload.get("y2", 0)) user32.SetCursorPos(x1, y1) time.sleep(0.05) user32.mouse_event(MOUSEEVENTF_LEFTDOWN, 0, 0, 0, 0) time.sleep(0.05) user32.SetCursorPos(x2, y2) time.sleep(0.05) user32.mouse_event(MOUSEEVENTF_LEFTUP, 0, 0, 0, 0) evidence = {"action": "MOUSE_DRAG", "from": (x1, y1), "to": (x2, y2), "timestamp": time.time()} return ActionResult(action_id=request.action_id, mission_id=request.mission_id, status="SUCCESS", source="MOUSE_KEYBOARD_EFFECTOR", evidence=evidence) elif aname in ("TYPE_TEXT", "TYPE"): text = str(payload.get("text") or request.target or "") # 2. Propagar ese mismo HWND a TYPE_TEXT target_hwnd = payload.get("hwnd") or self.get_mission_hwnd(request.mission_id) or 0 active_hwnd = 0 # 3. No resolver nuevamente por título si existe HWND válido if target_hwnd and user32.IsWindow(int(target_hwnd)) and user32.IsWindowVisible(int(target_hwnd)): active_hwnd = int(target_hwnd) else: target_hint = str(payload.get("target_window") or "").lower() if target_hint or not target_hwnd: WNDENUMPROC = ctypes.WINFUNCTYPE(ctypes.c_bool, wintypes.HWND, wintypes.LPARAM) def find_top_cb(h, _): nonlocal active_hwnd if user32.IsWindow(h) and user32.IsWindowVisible(h): cb_buf = ctypes.create_unicode_buffer(256) user32.GetClassNameW(h, cb_buf, 256) cls = cb_buf.value.lower() if cls not in EXCLUDED_CLASSES: b = ctypes.create_unicode_buffer(512) user32.GetWindowTextW(h, b, 512) val = b.value.lower() if target_hint: if target_hint in cls or target_hint in val: active_hwnd = int(h) return False else: if "notepad" in cls or any(c in val for c in ["bloc de notas", "notepad", "sin título", "sin titulo", "untitled"]): active_hwnd = int(h) return False return True user32.EnumWindows(WNDENUMPROC(find_top_cb), 0) expected_hwnd = int(active_hwnd) if active_hwnd else 0 fg_hwnd_before = user32.GetForegroundWindow() or 0 # 4. Antes de SendInput: # - IsWindow(hwnd)=TRUE # - IsWindowVisible(hwnd)=TRUE # - GetForegroundWindow() == hwnd is_fg_ok = False if expected_hwnd > 0 and user32.IsWindow(expected_hwnd) and user32.IsWindowVisible(expected_hwnd): if is_target_foreground(expected_hwnd): is_fg_ok = True else: # 5. Si no es foreground: # - intentar ShowWindow/SetForegroundWindow # - esperar breve intervalo # - volver a comprobar. is_fg_ok = activate_foreground(expected_hwnd) fg_hwnd_current = user32.GetForegroundWindow() or 0 b_curr = ctypes.create_unicode_buffer(512) user32.GetWindowTextW(fg_hwnd_current, b_curr, 512) active_window_title = b_curr.value.strip() is_final_target = bool( is_fg_ok and expected_hwnd > 0 and user32.IsWindow(expected_hwnd) and user32.IsWindowVisible(expected_hwnd) and (fg_hwnd_current == expected_hwnd or (fg_hwnd_current and user32.GetAncestor(fg_hwnd_current, 2) == expected_hwnd)) ) # 6. Si sigue sin coincidir: # status=FAILED # verified=False # error=TARGET_WINDOW_NOT_FOREGROUND # NO enviar ninguna tecla. if not is_final_target: evidence = { "action": "TYPE_TEXT", "typed": False, "expected_hwnd": expected_hwnd, "foreground_hwnd_before": fg_hwnd_before, "foreground_hwnd_after": fg_hwnd_current, "active_window_title": active_window_title, "target_validated": False, "chars_count": 0, "preview": text[:40], "error": "TARGET_WINDOW_NOT_FOREGROUND", "timestamp": time.time() } return ActionResult( action_id=request.action_id, mission_id=request.mission_id, status="FAILED", source="MOUSE_KEYBOARD_EFFECTOR", error="TARGET_WINDOW_NOT_FOREGROUND", evidence=evidence ) # Send keystrokes for ch in text: code = ord(ch) user32.keybd_event(0, code, KEYEVENTF_UNICODE, 0) user32.keybd_event(0, code, KEYEVENTF_UNICODE | KEYEVENTF_KEYUP, 0) time.sleep(0.01) time.sleep(0.1) # 7. Después de escribir, registrar: # expected_hwnd # foreground_hwnd_before # foreground_hwnd_after # active_window_title # target_validated=True/False fg_hwnd_after = user32.GetForegroundWindow() or 0 b_after = ctypes.create_unicode_buffer(512) user32.GetWindowTextW(fg_hwnd_after, b_after, 512) active_window_title_after = b_after.value.strip() post_alive = bool( expected_hwnd > 0 and user32.IsWindow(expected_hwnd) and user32.IsWindowVisible(expected_hwnd) and (fg_hwnd_after == expected_hwnd or (fg_hwnd_after and user32.GetAncestor(fg_hwnd_after, 2) == expected_hwnd)) ) evidence = { "action": "TYPE_TEXT", "typed": True, "expected_hwnd": expected_hwnd, "foreground_hwnd_before": fg_hwnd_before, "foreground_hwnd_after": fg_hwnd_after, "active_window_title": active_window_title_after or active_window_title, "target_validated": post_alive, "chars_count": len(text), "preview": text[:40], "timestamp": time.time() } return ActionResult( action_id=request.action_id, mission_id=request.mission_id, status="SUCCESS", source="MOUSE_KEYBOARD_EFFECTOR", evidence=evidence ) elif aname in ("KEY_PRESS", "PRESS_KEY"): key_name = str(payload.get("key") or request.target or "").upper() vk = VK_MAP.get(key_name) or (ord(key_name[0]) if key_name else 0) if vk: user32.keybd_event(vk, 0, 0, 0) time.sleep(0.05) user32.keybd_event(vk, 0, KEYEVENTF_KEYUP, 0) evidence = {"action": "KEY_PRESS", "key": key_name, "vk": vk, "timestamp": time.time()} return ActionResult(action_id=request.action_id, mission_id=request.mission_id, status="SUCCESS", source="MOUSE_KEYBOARD_EFFECTOR", evidence=evidence) elif aname == "HOTKEY": keys = payload.get("keys") or [request.target] vks = [VK_MAP.get(k.upper(), ord(k.upper()[0]) if k else 0) for k in keys if k] for vk in vks: user32.keybd_event(vk, 0, 0, 0) time.sleep(0.05) for vk in reversed(vks): user32.keybd_event(vk, 0, KEYEVENTF_KEYUP, 0) evidence = {"action": "HOTKEY", "keys": keys, "timestamp": time.time()} return ActionResult(action_id=request.action_id, mission_id=request.mission_id, status="SUCCESS", source="MOUSE_KEYBOARD_EFFECTOR", evidence=evidence) elif aname in ("OPEN_APPLICATION", "LAUNCH_APPLICATION", "START_APPLICATION"): app = str(payload.get("application") or payload.get("app") or request.target or "").strip().lower() app_map = { "notepad": { "exe": "notepad.exe", "full_path": "C:\\Windows\\System32\\notepad.exe", "win_titles": ["bloc de notas", "notepad", "sin título", "sin titulo", "untitled"], "class_names": ["notepad", "notepad.exe", "rich-edit", "applicationframewindow"] }, "bloc de notas": { "exe": "notepad.exe", "full_path": "C:\\Windows\\System32\\notepad.exe", "win_titles": ["bloc de notas", "notepad", "sin título", "sin titulo", "untitled"], "class_names": ["notepad", "notepad.exe", "rich-edit", "applicationframewindow"] }, "calc": { "exe": "calc.exe", "full_path": "C:\\Windows\\System32\\calc.exe", "win_titles": ["calculadora", "calculator", "calc"], "class_names": ["applicationframewindow", "calc"] }, "calculadora": { "exe": "calc.exe", "full_path": "C:\\Windows\\System32\\calc.exe", "win_titles": ["calculadora", "calculator", "calc"], "class_names": ["applicationframewindow", "calc"] }, } app_info = app_map.get(app, { "exe": f"{app}.exe" if not app.endswith(".exe") else app, "full_path": f"{app}.exe" if not app.endswith(".exe") else app, "win_titles": [app], "class_names": [app] }) exe_name = app_info["exe"] full_path = app_info.get("full_path") or exe_name win_search = [s.lower() for s in app_info["win_titles"]] class_search = [s.lower() for s in app_info.get("class_names", [])] # Resolve executable binary safely resolved_path = None if os.path.isabs(full_path) and os.path.exists(full_path): resolved_path = full_path elif shutil.which(full_path): resolved_path = shutil.which(full_path) elif shutil.which(exe_name): resolved_path = shutil.which(exe_name) if not resolved_path: return ActionResult( action_id=request.action_id, mission_id=request.mission_id, status="FAILED", source="MOUSE_KEYBOARD_EFFECTOR", error=f"APPLICATION_NOT_FOUND: Application executable '{full_path}' not found on system.", evidence={"application": app, "application_opened": False, "pid": 0, "hwnd": 0} ) # Step 1: Launch application using reliable GUI methods launch_ok = False launched_pid = 0 launch_method = "" launch_errors = [] # Method A: os.startfile try: os.startfile(resolved_path) launch_ok = True launch_method = "os.startfile" except Exception as ex_sf: launch_errors.append(f"os.startfile failed: {ex_sf}") # Method B: subprocess.Popen if not launch_ok: try: proc = subprocess.Popen([resolved_path]) launched_pid = proc.pid launch_ok = True launch_method = "subprocess_popen" except Exception as ex_sub: launch_errors.append(f"subprocess failed: {ex_sub}") # Method C: ShellExecuteW if not launch_ok: try: ret = shell32.ShellExecuteW(None, "open", resolved_path, None, None, 1) if ret > 32: launch_ok = True launch_method = "ShellExecuteW" else: launch_errors.append(f"ShellExecuteW returned code {ret}") except Exception as ex_se: launch_errors.append(f"ShellExecuteW failed: {ex_se}") # Method D: PowerShell Start-Process if not launch_ok: try: subprocess.Popen(["powershell.exe", "-Command", f"Start-Process '{resolved_path}' -WindowStyle Normal"]) launch_ok = True launch_method = "powershell_start_process" except Exception as ex_ps: launch_errors.append(f"PowerShell failed: {ex_ps}") # Step 2: Poll for real, interactive, visible window with valid title WNDENUMPROC = ctypes.WINFUNCTYPE(ctypes.c_bool, wintypes.HWND, wintypes.LPARAM) matched_hwnd = 0 actual_title = "" window_pid = 0 t_end = time.time() + 10.0 while time.time() < t_end and not matched_hwnd: time.sleep(0.2) def enum_cb(h, _): nonlocal matched_hwnd, actual_title, window_pid if user32.IsWindow(h) and user32.IsWindowVisible(h): cb_buf = ctypes.create_unicode_buffer(256) user32.GetClassNameW(h, cb_buf, 256) cls_name = cb_buf.value.lower() # Exclude helper overlays and tooltips if cls_name in EXCLUDED_CLASSES: return True b = ctypes.create_unicode_buffer(512) user32.GetWindowTextW(h, b, 512) val = b.value.strip() # Check window process w_pid = wintypes.DWORD() user32.GetWindowThreadProcessId(h, ctypes.byref(w_pid)) exe_file = "" h_proc = kernel32.OpenProcess(0x1000, False, w_pid.value) if h_proc: buf_path = ctypes.create_unicode_buffer(1024) sz = wintypes.DWORD(1024) if kernel32.QueryFullProcessImageNameW(h_proc, 0, buf_path, ctypes.byref(sz)): exe_file = os.path.basename(buf_path.value).lower() kernel32.CloseHandle(h_proc) # Validate: MUST have a valid title AND match class or title class_matches = any(c in cls_name for c in class_search if c) title_matches = any(s in val.lower() for s in win_search if s) proc_matches = bool(exe_file and exe_name.lower() in exe_file) # For legitimate GUI apps, a visible window must match and have non-empty title if val and (class_matches or title_matches) and (proc_matches or not exe_file): matched_hwnd = int(h) actual_title = val window_pid = w_pid.value return False return True user32.EnumWindows(WNDENUMPROC(enum_cb), 0) # Step 3: Window Stability Check: wait 300ms and re-verify window_stable = False if matched_hwnd > 0: time.sleep(0.3) if user32.IsWindow(matched_hwnd) and user32.IsWindowVisible(matched_hwnd): b = ctypes.create_unicode_buffer(512) user32.GetWindowTextW(matched_hwnd, b, 512) if b.value.strip(): actual_title = b.value.strip() window_stable = True # Step 4: SUCCESS only when HWND > 0, visible, stable, and has valid title if matched_hwnd > 0 and window_stable and actual_title: # 1. Propagar el HWND obtenido por OPEN_APPLICATION self.set_mission_hwnd(request.mission_id, matched_hwnd) # Bring window to foreground activate_foreground(matched_hwnd) evidence = { "action": "OPEN_APPLICATION", "application": app, "application_opened": True, "pid": window_pid or launched_pid, "hwnd": matched_hwnd, "window_title": actual_title, "window_stable": True, "verified": True, "launch_method": launch_method, "timestamp": time.time() } return ActionResult( action_id=request.action_id, mission_id=request.mission_id, status="SUCCESS", source="MOUSE_KEYBOARD_EFFECTOR", evidence=evidence ) else: # Process was spawned (or requested) but NO visible interactive window appeared! return ActionResult( action_id=request.action_id, mission_id=request.mission_id, status="FAILED", source="MOUSE_KEYBOARD_EFFECTOR", error=f"APPLICATION_PROCESS_WITHOUT_VISIBLE_WINDOW: Application '{app}' process spawned (pid={window_pid or launched_pid}) but failed to create a visible, stable interactive window within timeout.", evidence={ "application": app, "application_opened": False, "pid": window_pid or launched_pid, "hwnd": 0, "window_stable": False, "verified": False, "launch_errors": launch_errors } ) elif aname in ("WINDOW_FOCUS", "FOCUS_WINDOW"): # 1. Propagar el HWND obtenido por OPEN_APPLICATION al siguiente WINDOW_FOCUS target_hwnd = payload.get("hwnd") or self.get_mission_hwnd(request.mission_id) or 0 target_title = str(payload.get("title") or payload.get("target_title") or request.target or "").lower() title_candidates = [target_title] if "notepad" in target_title or "bloc de notas" in target_title: title_candidates.extend(["bloc de notas", "notepad", "sin título", "sin titulo", "untitled"]) elif "calc" in target_title or "calculadora" in target_title: title_candidates.extend(["calculadora", "calculator", "calc"]) matched_hwnd = 0 actual_title = "" # 3. No resolver nuevamente por título si existe HWND válido if target_hwnd and user32.IsWindow(int(target_hwnd)) and user32.IsWindowVisible(int(target_hwnd)): matched_hwnd = int(target_hwnd) b = ctypes.create_unicode_buffer(512) user32.GetWindowTextW(matched_hwnd, b, 512) actual_title = b.value.strip() else: def enum_cb(h, _): nonlocal matched_hwnd, actual_title if user32.IsWindow(h) and user32.IsWindowVisible(h): cb_buf = ctypes.create_unicode_buffer(256) user32.GetClassNameW(h, cb_buf, 256) cls = cb_buf.value.lower() if cls in EXCLUDED_CLASSES: return True b = ctypes.create_unicode_buffer(512) user32.GetWindowTextW(h, b, 512) val = b.value.strip() if val: for cand in title_candidates: if cand and cand in val.lower(): matched_hwnd = int(h) actual_title = val return False if "notepad" in target_title and cls == "notepad": matched_hwnd = int(h) actual_title = val return False return True WNDENUMPROC = ctypes.WINFUNCTYPE(ctypes.c_bool, wintypes.HWND, wintypes.LPARAM) user32.EnumWindows(WNDENUMPROC(enum_cb), 0) if matched_hwnd > 0: self.set_mission_hwnd(request.mission_id, matched_hwnd) activate_foreground(matched_hwnd) time.sleep(0.15) fg_hwnd = user32.GetForegroundWindow() or 0 is_fg = bool(fg_hwnd == matched_hwnd or (fg_hwnd and user32.GetAncestor(fg_hwnd, 2) == matched_hwnd)) evidence = { "action": "WINDOW_FOCUS", "hwnd": matched_hwnd, "target_title": target_title, "window_title": actual_title, "focused": is_fg, "is_foreground": is_fg, "found": True, "timestamp": time.time() } return ActionResult( action_id=request.action_id, mission_id=request.mission_id, status="SUCCESS", source="MOUSE_KEYBOARD_EFFECTOR", evidence=evidence ) else: return ActionResult( action_id=request.action_id, mission_id=request.mission_id, status="FAILED", source="MOUSE_KEYBOARD_EFFECTOR", error=f"WINDOW_NOT_FOUND: Could not find visible window matching '{target_title}' or hwnd '{target_hwnd}'.", evidence={"hwnd": 0, "found": False, "focused": False} ) else: return ActionResult(action_id=request.action_id, mission_id=request.mission_id, status="FAILED", source="MOUSE_KEYBOARD_EFFECTOR", error=f"UNKNOWN_ACTION: {aname}") except Exception as e: return ActionResult(action_id=request.action_id, mission_id=request.mission_id, status="FAILED", source="MOUSE_KEYBOARD_EFFECTOR", error=f"EXECUTION_ERROR: {str(e)}")\n\n\n""" final_content = pre_part + new_class + post_part tmp_file = target_file + ".tmp" with open(tmp_file, "w", encoding="utf-8") as f: f.write(final_content) py_compile.compile(tmp_file, doraise=True) print("PY_COMPILE PASSED") shutil.move(tmp_file, target_file) print("UPDATED EFFECTOR_RUNTIME SUCCESSFULLY")