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() start_marker = " def execute(self, request: ActionRequest) -> ActionResult:" end_marker = " except Exception as e:" idx_start = content.find(start_marker) idx_end = content.find(end_marker, idx_start) assert idx_start != -1, "start_marker not found" assert idx_end != -1, "end_marker not found" new_execute_body = r''' def execute(self, request: ActionRequest) -> ActionResult: self.last_seen = time.time() import ctypes from ctypes import wintypes 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 # 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"] }, "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"] }, "calc": { "exe": "calc.exe", "full_path": "C:\\Windows\\System32\\calc.exe", "win_titles": ["calculadora", "calculator", "calc"] }, "calculadora": { "exe": "calc.exe", "full_path": "C:\\Windows\\System32\\calc.exe", "win_titles": ["calculadora", "calculator", "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] }) 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"]] # 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 # Method A: Shell.Application COM (Routes directly via explorer.exe on WinSta0\Default) try: import win32com.client shell_com = win32com.client.Dispatch("Shell.Application") shell_com.ShellExecute(full_path, "", "", "open", 1) launch_ok = True except Exception: pass # Method B: os.startfile fallback if not launch_ok: try: os.startfile(full_path) launch_ok = True except Exception: pass # Method C: ShellExecuteExW fallback if not launch_ok: try: shell32 = ctypes.windll.shell32 class SHELLEXECUTEINFOW(ctypes.Structure): _fields_ = [ ('cbSize', wintypes.DWORD), ('fMask', wintypes.ULONG), ('hwnd', wintypes.HWND), ('lpVerb', wintypes.LPCWSTR), ('lpFile', wintypes.LPCWSTR), ('lpParameters', wintypes.LPCWSTR), ('lpDirectory', wintypes.LPCWSTR), ('nShow', ctypes.c_int), ('hInstApp', wintypes.HINSTANCE), ('lpIDList', wintypes.LPVOID), ('lpClass', wintypes.LPCWSTR), ('hkeyClass', wintypes.HKEY), ('dwHotKey', wintypes.DWORD), ('hIconOrMonitor', wintypes.HANDLE), ('hProcess', wintypes.HANDLE) ] sei = SHELLEXECUTEINFOW() sei.cbSize = ctypes.sizeof(SHELLEXECUTEINFOW) sei.fMask = 0x00000040 sei.hwnd = 0 sei.lpVerb = 'open' sei.lpFile = full_path sei.nShow = 1 if shell32.ShellExecuteExW(ctypes.byref(sei)): launched_pid = kernel32.GetProcessId(sei.hProcess) if sei.hProcess else 0 launch_ok = True except Exception: pass # Method D: subprocess.Popen with lpDesktop fallback 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([full_path], startupinfo=si, shell=False) launched_pid = proc.pid launch_ok = True except Exception as ex_launch: return ActionResult( action_id=request.action_id, mission_id=request.mission_id, status="FAILED", source="MOUSE_KEYBOARD_EFFECTOR", error=f"APPLICATION_OPEN_FAILED: Failed to launch '{exe_name}': {ex_launch}", evidence={"application": app, "application_opened": False, "pid": 0, "hwnd": 0} ) # Step 3: Poll for interactive visible window belonging to the requested app matched_hwnd = 0 actual_title = "" window_pid = 0 t_end = time.time() + 6.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: l = user32.GetWindowTextLengthW(h) + 1 if l > 1: b = ctypes.create_unicode_buffer(l) user32.GetWindowTextW(h, b, l) val = b.value if val: w_pid = wintypes.DWORD() user32.GetWindowThreadProcessId(h, ctypes.byref(w_pid)) # Query process image name of window owner 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 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 proc_matches or title_matches: matched_hwnd = h_int actual_title = val window_pid = w_pid.value return False return True user32.EnumWindows(WNDENUMPROC(enum_cb), 0) # 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 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, "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 } ) 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 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}" ) ''' new_content = content[:idx_start] + new_execute_body + "\n" + content[idx_end:] with open(path, "w", encoding="utf-8") as f: f.write(new_content) print("SUCCESS_SURGICAL_EFFECTOR_FIX_APPLIED")