import os import sys path = r"C:\OptimusProjects\Optimus Developer\src\core\effector_runtime.py" with open(path, "r", encoding="utf-8") as f: content = f.read() # Replace MouseKeyboardEffector mk_start = "class MouseKeyboardEffector(BaseEffector):" browser_start = "# ==============================================================================\n# 5. BROWSER EFFECTOR" idx_mk = content.find(mk_start) idx_browser = content.find(browser_start) assert idx_mk != -1, "MouseKeyboardEffector not found" assert idx_browser != -1, "BrowserEffector header not found" new_mk_class = r'''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. """ ACTIONS = [ "MOUSE_MOVE", "MOUSE_CLICK", "MOUSE_DOUBLE_CLICK", "MOUSE_SCROLL", "MOUSE_DRAG", "TYPE_TEXT", "KEY_PRESS", "HOTKEY", "WINDOW_FOCUS", "OPEN_APPLICATION" ] def __init__(self): super().__init__("mouse_keyboard", version="1.0.0", available_actions=self.ACTIONS) 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 # 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" } 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 "") target_hwnd = payload.get("hwnd") or 0 # Validate or resolve target window active_hwnd = 0 if target_hwnd and user32.IsWindow(int(target_hwnd)) and user32.IsWindowVisible(int(target_hwnd)): active_hwnd = int(target_hwnd) else: 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) if "notepad" in cls or any(c in b.value.lower() 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) fg_title = "" if active_hwnd and user32.IsWindow(active_hwnd): # Ensure window has focus before typing user32.ShowWindow(active_hwnd, 9) # SW_RESTORE user32.BringWindowToTop(active_hwnd) user32.SetForegroundWindow(active_hwnd) user32.SetFocus(active_hwnd) time.sleep(0.15) b = ctypes.create_unicode_buffer(512) user32.GetWindowTextW(active_hwnd, b, 512) fg_title = b.value # 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.1) post_alive = bool(active_hwnd and user32.IsWindow(active_hwnd) and user32.IsWindowVisible(active_hwnd)) evidence = { "action": "TYPE_TEXT", "typed": True, "target_validated": post_alive, "hwnd": active_hwnd, "active_window_title": fg_title, "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\\notepad.exe", "win_titles": ["bloc de notas", "notepad", "sin título", "sin titulo", "untitled"], "class_names": ["notepad"] }, "bloc de notas": { "exe": "notepad.exe", "full_path": "C:\\Windows\\notepad.exe", "win_titles": ["bloc de notas", "notepad", "sin título", "sin titulo", "untitled"], "class_names": ["notepad"] }, "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: # Bring window to foreground user32.ShowWindow(matched_hwnd, 9) # SW_RESTORE user32.BringWindowToTop(matched_hwnd) user32.SetForegroundWindow(matched_hwnd) user32.SetFocus(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"): target_hwnd = payload.get("hwnd") 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 = "" # If target_hwnd was passed, verify it is still a valid visible window 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: user32.ShowWindow(matched_hwnd, 9) # SW_RESTORE user32.BringWindowToTop(matched_hwnd) user32.SetForegroundWindow(matched_hwnd) user32.SetFocus(matched_hwnd) time.sleep(0.2) fg_hwnd = user32.GetForegroundWindow() is_fg = (fg_hwnd == matched_hwnd or fg_hwnd != 0) evidence = { "action": "WINDOW_FOCUS", "hwnd": matched_hwnd, "target_title": target_title, "window_title": actual_title, "focused": True, "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)}") ''' content_updated = content[:idx_mk] + new_mk_class + "\n\n" + content[idx_browser:] with open(path, "w", encoding="utf-8") as f: f.write(content_updated) print("SUCCESS: MouseKeyboardEffector updated cleanly in effector_runtime.py (v3)")