import os import sys path = r"C:\OptimusProjects\Optimus Developer\src\core\effector_runtime.py" with open(path, "r", encoding="utf-8") as f: c = f.read() target_start = ' elif aname in ("TYPE_TEXT", "TYPE"):' target_end = ' return ActionResult(\n action_id=request.action_id,\n mission_id=request.mission_id,\n status="BLOCKED",\n source="MOUSE_KEYBOARD_EFFECTOR",\n error=f"UNSUPPORTED_MOUSE_KEYBOARD_ACTION: {aname}"\n )' start_idx = c.find(target_start) end_idx = c.find(target_end) assert start_idx != -1, "target_start not found" assert end_idx != -1, "target_end not found" new_block = ''' elif aname in ("TYPE_TEXT", "TYPE"): text = str(payload.get("text") or request.target or "") fg_hwnd = user32.GetForegroundWindow() target_win = payload.get("target_window") or payload.get("target") fg_title = "" if fg_hwnd and user32.IsWindow(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) post_alive = True if fg_hwnd: post_alive = bool(user32.IsWindow(fg_hwnd)) evidence = { "action": "TYPE_TEXT", "typed": True, "target_validated": bool(fg_hwnd and post_alive), "hwnd": int(fg_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", "win_title": "bloc de notas", "win_title_alt": "notepad"}, "bloc de notas": {"exe": "notepad.exe", "win_title": "bloc de notas", "win_title_alt": "notepad"}, "calc": {"exe": "calc.exe", "win_title": "calculadora", "win_title_alt": "calculator"}, "calculadora": {"exe": "calc.exe", "win_title": "calculadora", "win_title_alt": "calculator"}, } app_info = app_map.get(app, {"exe": f"{app}.exe" if not app.endswith(".exe") else app, "win_title": app, "win_title_alt": app}) exe_name = app_info["exe"] win_search = [app_info["win_title"].lower(), app_info.get("win_title_alt", "").lower()] # Step 1: Pre-launch snapshot of existing visible windows from ctypes import wintypes 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 GUI Launch via ShellExecuteExW with NO_UI flag shell32 = ctypes.windll.shell32 kernel32 = ctypes.windll.kernel32 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) # SEE_MASK_NOCLOSEPROCESS (0x40) | SEE_MASK_FLAG_NO_UI (0x400) sei.fMask = 0x00000040 | 0x00000400 sei.hwnd = 0 sei.lpVerb = 'open' sei.lpFile = exe_name sei.lpParameters = None sei.lpDirectory = None sei.nShow = 1 # SW_SHOWNORMAL (Interactive GUI Window) launch_ok = bool(shell32.ShellExecuteExW(ctypes.byref(sei))) launched_pid = kernel32.GetProcessId(sei.hProcess) if (launch_ok and sei.hProcess) else 0 # Fallback if ShellExecuteExW failed if not launch_ok: try: import subprocess si = subprocess.STARTUPINFO() si.dwFlags |= subprocess.STARTF_USESHOWWINDOW si.wShowWindow = 1 # SW_SHOWNORMAL proc = subprocess.Popen([exe_name], 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 app matched_hwnd = 0 actual_title = "" window_pid = 0 t_end = time.time() + 4.5 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): 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 = ctypes.c_ulong() user32.GetWindowThreadProcessId(h, ctypes.byref(w_pid)) title_matches = any(s in val.lower() for s in win_search if s) pid_matches = (launched_pid > 0 and w_pid.value == launched_pid) is_new_window = (h_int not in existing_hwnds) if (pid_matches and (title_matches or not val.startswith("MSCTFIME"))) or (is_new_window and 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 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: Process '{exe_name}' spawned (pid={launched_pid}) but failed to create a visible, stable interactive window within timeout.", evidence={ "application": app, "application_opened": False, "pid": launched_pid, "hwnd": 0, "window_stable": False, "verified": False } ) elif aname in ("WINDOW_FOCUS", "FOCUS_WINDOW"): from ctypes import wintypes WNDENUMPROC = ctypes.WINFUNCTYPE(ctypes.c_bool, wintypes.HWND, wintypes.LPARAM) 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", "untitled"]) elif "calc" in target_title or "calculadora" in target_title: title_candidates.extend(["calculadora", "calculator"]) 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 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} ) ''' c = c[:start_idx] + new_block + c[end_idx:] with open(path, "w", encoding="utf-8") as f: f.write(c) print("SUCCESS_PATCHED_EFFECTOR_RUNTIME_V2")