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() # 1. 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. """ 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 user32 = ctypes.windll.user32 kernel32 = ctypes.windll.kernel32 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.GetWindowThreadProcessId.argtypes = [wintypes.HWND, ctypes.POINTER(wintypes.DWORD)] user32.GetWindowThreadProcessId.restype = wintypes.DWORD # Attach calling thread to interactive input desktop (WinSta0\Default) try: h_input = user32.OpenInputDesktop(0, False, 0x01FF) if h_input: user32.SetThreadDesktop(h_input) except Exception: pass # Initialize COM on calling thread try: import pythoncom pythoncom.CoInitialize() except Exception: pass # 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 } 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 "") fg_hwnd = user32.GetForegroundWindow() # Verify target window is valid and visible before typing fg_title = "" if fg_hwnd and user32.IsWindow(fg_hwnd) and user32.IsWindowVisible(fg_hwnd): l = user32.GetWindowTextLengthW(fg_hwnd) + 1 b = ctypes.create_unicode_buffer(l) user32.GetWindowTextW(fg_hwnd, b, l) 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(fg_hwnd and user32.IsWindow(fg_hwnd) and user32.IsWindowVisible(fg_hwnd)) evidence = { "action": "TYPE_TEXT", "typed": True, "target_validated": post_alive, "hwnd": int(fg_hwnd) if fg_hwnd else 0, "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\\System32\\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\\System32\\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 without triggering Windows error dialogs 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: # Non-existent application requested -> fail cleanly without hanging 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: Pre-launch snapshot of existing visible windows on Default desktop WNDENUMPROC = ctypes.WINFUNCTYPE(ctypes.c_bool, wintypes.HWND, wintypes.LPARAM) existing_hwnds = set() try: def snap_cb(h, _): if user32.IsWindowVisible(h): existing_hwnds.add(int(h)) return True user32.EnumWindows(WNDENUMPROC(snap_cb), 0) except Exception: pass # Step 2: Interactive launch to ensure window appears on user interactive desktop launch_ok = False launched_pid = 0 launch_method = "" launch_errors = [] # Method A: Shell.Application COM (Routes directly via explorer.exe on WinSta0\Default) try: import pythoncom pythoncom.CoInitialize() import win32com.client shell_com = win32com.client.Dispatch("Shell.Application") shell_com.ShellExecute(resolved_path, "", "", "open", 1) launch_ok = True launch_method = "Shell.Application" except Exception as ex_a: launch_errors.append(f"Method A (COM) failed: {ex_a}") # Method B: PowerShell Start-Process in user session if not launch_ok: try: import subprocess subprocess.Popen(["powershell.exe", "-WindowStyle", "Hidden", "-Command", f"Start-Process '{resolved_path}'"]) launch_ok = True launch_method = "powershell_start_process" except Exception as ex_b: launch_errors.append(f"Method B (PowerShell) failed: {ex_b}") # Method C: explorer.exe launcher fallback if not launch_ok: try: import subprocess subprocess.Popen(["explorer.exe", resolved_path]) launch_ok = True launch_method = "explorer_launcher" except Exception as ex_c: launch_errors.append(f"Method C (Explorer) failed: {ex_c}") # Method D: subprocess with explicit WinSta0\Default desktop station if not launch_ok: try: import subprocess si = subprocess.STARTUPINFO() si.dwFlags |= subprocess.STARTF_USESHOWWINDOW si.wShowWindow = 1 si.lpDesktop = r"WinSta0\Default" proc = subprocess.Popen([resolved_path], startupinfo=si, shell=False) launched_pid = proc.pid launch_ok = True launch_method = "subprocess_winsta0" except Exception as ex_d: launch_errors.append(f"Method D (subprocess) failed: {ex_d}") # Step 3: Poll for interactive visible window belonging to the requested app 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 h_int = int(h) if user32.IsWindow(h) and user32.IsWindowVisible(h): if h_int not in existing_hwnds: # Check window title l = user32.GetWindowTextLengthW(h) + 1 b = ctypes.create_unicode_buffer(l if l > 1 else 256) user32.GetWindowTextW(h, b, len(b)) val = b.value # Check window class name cb_buf = ctypes.create_unicode_buffer(256) user32.GetClassNameW(h, cb_buf, 256) cls_name = cb_buf.value.lower() # Check process image name of window owner 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 match against class, title, or process name 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) if class_matches or title_matches or proc_matches: matched_hwnd = h_int actual_title = val window_pid = w_pid.value return False return True user32.EnumWindows(WNDENUMPROC(enum_cb), 0) # If matched window has empty title yet, wait up to 2 seconds for title to populate if matched_hwnd > 0 and not actual_title: t_title = time.time() + 2.0 while time.time() < t_title and not actual_title: time.sleep(0.1) l = user32.GetWindowTextLengthW(matched_hwnd) + 1 if l > 1: b = ctypes.create_unicode_buffer(l) user32.GetWindowTextW(matched_hwnd, b, l) actual_title = b.value # Step 4: 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): l = user32.GetWindowTextLengthW(matched_hwnd) + 1 if l > 1: b = ctypes.create_unicode_buffer(l) user32.GetWindowTextW(matched_hwnd, b, l) actual_title = b.value or actual_title window_stable = True # Step 5: SUCCESS only if window exists, is visible, and stable if matched_hwnd > 0 and window_stable: 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) l = user32.GetWindowTextLengthW(matched_hwnd) + 1 b = ctypes.create_unicode_buffer(l) user32.GetWindowTextW(matched_hwnd, b, l) actual_title = b.value else: def enum_cb(h, _): nonlocal matched_hwnd, actual_title if user32.IsWindow(h) and user32.IsWindowVisible(h): l = user32.GetWindowTextLengthW(h) + 1 if l > 1: b = ctypes.create_unicode_buffer(l) user32.GetWindowTextW(h, b, l) val = b.value for cand in title_candidates: if cand and cand in val.lower(): matched_hwnd = int(h) actual_title = val return False # Also check class name for notepad cb_buf = ctypes.create_unicode_buffer(256) user32.GetClassNameW(h, cb_buf, 256) if "notepad" in target_title and cb_buf.value.lower() == "notepad": matched_hwnd = int(h) l = user32.GetWindowTextLengthW(h) + 1 b = ctypes.create_unicode_buffer(l if l > 1 else 256) user32.GetWindowTextW(h, b, len(b)) actual_title = b.value 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.SetForegroundWindow(matched_hwnd) time.sleep(0.3) fg_hwnd = user32.GetForegroundWindow() is_fg = (fg_hwnd == matched_hwnd) 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: No visible window matching '{target_title}' or HWND {target_hwnd}", evidence={"action": "WINDOW_FOCUS", "focused": False, "found": False} ) return ActionResult( action_id=request.action_id, mission_id=request.mission_id, status="BLOCKED", source="MOUSE_KEYBOARD_EFFECTOR", error=f"UNSUPPORTED_MOUSE_KEYBOARD_ACTION: {aname}" ) except Exception as e: self.last_error = str(e) return ActionResult( action_id=request.action_id, mission_id=request.mission_id, status="FAILED", source="MOUSE_KEYBOARD_EFFECTOR", error=f"MOUSE_KEYBOARD_ERROR: {e}" ) ''' content = content[:idx_mk] + new_mk_class + content[idx_browser:] # 2. Allow status inquiry in execute_action when status is UNAVAILABLE, and return FAILED for non-status actions old_unavail_marker = 'if effector.status == STATUS_UNAVAILABLE:' pos_unavail = content.find(old_unavail_marker) if pos_unavail != -1: pos_end = content.find('return ActionResult(', pos_unavail) pos_close = content.find(')\n', pos_end) + 2 replacement = '''if effector.status == STATUS_UNAVAILABLE: if request.action_type.upper() not in ("CAMERA_STATUS", "STATUS", "STREAMING_STATUS"): return ActionResult( action_id=request.action_id, mission_id=request.mission_id, status="FAILED", source="EFFECTOR_REGISTRY", error=f"CAPABILITY_UNAVAILABLE: Effector '{cap}' status is UNAVAILABLE ({effector.last_error})." ) ''' content = content[:pos_unavail] + replacement + content[pos_close:] with open(path, 'w', encoding='utf-8') as f: f.write(content) print("DEFINITIVE_SURGICAL_FIX_APPLIED_SUCCESSFULLY")