# -*- coding: utf-8 -*- """ DOM_SYNC_01 Patcher for C:\OptimusProjects\Optimus Developer\src\core\effector_runtime.py """ import os import sys import hashlib import py_compile import shutil TARGET_PATH = r"C:\OptimusProjects\Optimus Developer\src\core\effector_runtime.py" BACKUP_PATH = TARGET_PATH + ".bak_dom_sync_01" EXPECTED_PRE_HASH = "80A46E921DE72600029B9858BD2AEA56C51A717F10CD7F4ED930B1567D94D397" with open(TARGET_PATH, "rb") as f: orig_bytes = f.read() current_hash = hashlib.sha256(orig_bytes).hexdigest().upper() print(f"PRE_PATCH_HASH: {current_hash}") if current_hash != EXPECTED_PRE_HASH: print(f"ERROR: Hash mismatch! Expected {EXPECTED_PRE_HASH}, got {current_hash}") sys.exit(1) # Make backup if not existing if not os.path.exists(BACKUP_PATH): shutil.copy2(TARGET_PATH, BACKUP_PATH) print(f"Backup created at: {BACKUP_PATH}") orig_text = orig_bytes.decode("utf-8") newline = "\r\n" if "\r\n" in orig_text else "\n" # 1. Add top-level stdlib imports if not present old_imports = newline.join([ "import urllib.request", "import urllib.parse", "from typing import Any, Callable, Dict, List, Optional, Tuple" ]) new_imports = newline.join([ "import urllib.request", "import urllib.parse", "import socket", "import struct", "import base64", "from typing import Any, Callable, Dict, List, Optional, Tuple" ]) if old_imports not in orig_text: print("ERROR: old_imports not found in orig_text") sys.exit(1) text = orig_text.replace(old_imports, new_imports, 1) # 2. Add ScreenCaptureEffector alias after ScreenVisionEffector old_sv_end = newline.join([ "# ==============================================================================", "# 4. MOUSE & KEYBOARD EFFECTOR", "# ==============================================================================", "class MouseKeyboardEffector(BaseEffector):" ]) new_sv_end = newline.join([ "# Alias for ScreenVisionEffector compliant with capture interface", "ScreenCaptureEffector = ScreenVisionEffector", "", "", "# ==============================================================================", "# 4. MOUSE & KEYBOARD EFFECTOR", "# ==============================================================================", "class MouseKeyboardEffector(BaseEffector):" ]) if old_sv_end not in text: print("ERROR: old_sv_end not found in text") sys.exit(1) text = text.replace(old_sv_end, new_sv_end, 1) # 3. Replace BrowserEffector block idx_browser_start = text.find("class BrowserEffector(BaseEffector):") idx_audio_start = text.find("class AudioEffector(BaseEffector):") if idx_browser_start == -1 or idx_audio_start == -1: print("ERROR: BrowserEffector or AudioEffector not found") sys.exit(1) target_sec6 = newline.join([ "# ==============================================================================", "# 6. AUDIO EFFECTOR" ]) idx_section_6 = text.rfind(target_sec6, idx_browser_start, idx_audio_start) if idx_section_6 == -1: idx_section_6 = idx_audio_start new_browser_effector_lines = [ 'class BrowserEffector(BaseEffector):', ' """', ' Physical reusable browser effector.', ' Manages browser lifecycle, tab opening/closing, URL navigation, state queries,', ' and visual screenshots via Headless Chrome / DevTools Protocol & Live Preview.', ' """', ' ACTIONS = [', ' "BROWSER_OPEN_URL", "BROWSER_NAVIGATE", "BROWSER_BACK", "BROWSER_FORWARD",', ' "BROWSER_REFRESH", "BROWSER_NEW_TAB", "BROWSER_CLOSE_TAB", "BROWSER_GET_STATE",', ' "BROWSER_INTERACT", "BROWSER_CAPTURE"', ' ]', '', ' def __init__(self, workspace_root: str):', ' super().__init__("browser", version="1.0.0", available_actions=self.ACTIONS)', ' self.workspace_root = os.path.abspath(workspace_root) if workspace_root else None', ' self.captures_dir = os.path.join(r"C:\\OptimusProjects\\Optimus Developer", "data", "captures")', ' os.makedirs(self.captures_dir, exist_ok=True)', ' self.chrome_path = self._find_browser_binary()', ' if not self.chrome_path:', ' self.set_status(STATUS_UNAVAILABLE, error="No supported browser (Chrome/Edge) found", health="OFFLINE")', '', ' # Dynamic CDP session tracking & window provenance', ' self.browser_sessions: Dict[str, Dict[str, Any]] = {}', ' self.last_profile_dir: Optional[str] = None', ' self.last_debugging_port: Optional[int] = None', ' self.last_browser_hwnd: Optional[int] = None', ' self.last_browser_pid: Optional[int] = None', ' self.last_target_url: Optional[str] = None', ' self.last_expected_title: Optional[str] = None', ' self.last_mission_id: Optional[str] = None', '', ' def _find_browser_binary(self) -> Optional[str]:', ' candidates = [', ' r"C:\\Program Files\\Google\\Chrome\\Application\\chrome.exe",', ' r"C:\\Program Files (x86)\\Microsoft\\Edge\\Application\\msedge.exe",', ' r"C:\\Program Files\\Microsoft\\Edge\\Application\\msedge.exe"', ' ]', ' for c in candidates:', ' if os.path.exists(c):', ' return c', ' return None', '', ' # --------------------------------------------------------------------------', ' # Minimal CDP Client & Live DOM Verification (stdlib only, RFC6455 masked)', ' # --------------------------------------------------------------------------', '', ' def _read_devtools_active_port(self, profile_dir: str, timeout: float = 6.0) -> Optional[int]:', ' """Reads ephemeral remote debugging port from /DevToolsActivePort with bounded polling."""', ' port_file = os.path.join(profile_dir, "DevToolsActivePort")', ' deadline = time.monotonic() + timeout', ' while time.monotonic() < deadline:', ' if os.path.exists(port_file):', ' try:', ' with open(port_file, "r", encoding="utf-8") as f:', ' lines = [l.strip() for l in f if l.strip()]', ' if lines:', ' port = int(lines[0])', ' # Quick socket probe to ensure port is listening', ' s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)', ' s.settimeout(0.3)', ' try:', ' s.connect(("127.0.0.1", port))', ' s.close()', ' return port', ' except Exception:', ' s.close()', ' except Exception:', ' pass', ' time.sleep(0.15)', ' return None', '', ' def _get_cdp_targets(self, port: int, timeout: float = 2.5) -> List[Dict[str, Any]]:', ' """Queries http://127.0.0.1:/json/list for active Chrome targets."""', ' try:', ' url = f"http://127.0.0.1:{port}/json/list"', ' req = urllib.request.Request(url, headers={"User-Agent": "OptimusEffector/1.0"})', ' with urllib.request.urlopen(req, timeout=timeout) as resp:', ' data = json.loads(resp.read().decode("utf-8"))', ' return data if isinstance(data, list) else []', ' except Exception:', ' return []', '', ' def _select_cdp_target(', ' self,', ' targets: List[Dict[str, Any]],', ' target_url: Optional[str] = None,', ' expected_title: Optional[str] = None', ' ) -> Optional[Dict[str, Any]]:', ' """', ' Selects target page matching target_url or expected_title.', ' Rejects non-matching targets (FAIL CLOSED). Never picks arbitrary targets.', ' """', ' valid_targets = [t for t in targets if t.get("webSocketDebuggerUrl")]', ' page_targets = [t for t in valid_targets if t.get("type") == "page"]', ' candidates = page_targets if page_targets else valid_targets', '', ' if not candidates:', ' return None', '', ' def _urls_match(u1: str, u2: str) -> bool: ', ' if not u1 or not u2:', ' return False', ' try:', ' p1 = urllib.parse.urlparse(u1)', ' p2 = urllib.parse.urlparse(u2)', ' if p1.netloc and p2.netloc:', ' if p1.netloc.lower() != p2.netloc.lower():', ' return False', ' path1 = p1.path.rstrip("/").lower()', ' path2 = p2.path.rstrip("/").lower()', ' return path1 == path2 or path1.startswith(path2) or path2.startswith(path1)', ' except Exception:', ' pass', ' return u1.rstrip("/").lower() in u2.rstrip("/").lower() or u2.rstrip("/").lower() in u1.rstrip("/").lower()', '', ' # Priority 1: Both expected_title and target_url match', ' if expected_title and target_url:', ' for t in candidates:', ' if expected_title.strip().casefold() in t.get("title", "").casefold() and _urls_match(target_url, t.get("url", "")):', ' return t', '', ' # Priority 2: expected_title match', ' if expected_title:', ' for t in candidates:', ' if expected_title.strip().casefold() in t.get("title", "").casefold():', ' return t', '', ' # Priority 3: target_url match', ' if target_url:', ' for t in candidates:', ' if _urls_match(target_url, t.get("url", "")):', ' return t', '', ' # FAIL CLOSED: no candidate matches requested identity', ' return None', '', ' def _ws_handshake_localhost(self, host: str, port: int, path: str, timeout: float = 5.0) -> Optional[socket.socket]:', ' """Establishes RFC6455 WebSocket connection to localhost endpoint with bounded timeout."""', ' try:', ' s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)', ' s.settimeout(timeout)', ' s.connect((host, port))', '', ' sec_key = base64.b64encode(os.urandom(16)).decode("ascii")', ' req = (', ' f"GET {path} HTTP/1.1\\r\\n"', ' f"Host: {host}:{port}\\r\\n"', ' f"Upgrade: websocket\\r\\n"', ' f"Connection: Upgrade\\r\\n"', ' f"Sec-WebSocket-Key: {sec_key}\\r\\n"', ' f"Sec-WebSocket-Version: 13\\r\\n\\r\\n"', ' )', ' s.sendall(req.encode("ascii"))', '', ' resp = b""', ' deadline = time.monotonic() + timeout', ' while b"\\r\\n\\r\\n" not in resp:', ' if time.monotonic() > deadline:', ' s.close()', ' return None', ' chunk = s.recv(1024)', ' if not chunk:', ' s.close()', ' return None', ' resp += chunk', '', ' header = resp.split(b"\\r\\n\\r\\n")[0]', ' status_line = header.split(b"\\r\\n")[0]', ' if b" 101 " not in status_line:', ' s.close()', ' return None', ' return s', ' except Exception:', ' return None', '', ' def _ws_send_frame(self, s: socket.socket, payload: bytes, opcode: int = 1, timeout: float = 5.0) -> None:', ' """Sends an RFC6455 client-to-server frame (MUST be MASKED)."""', ' s.settimeout(timeout)', ' mask = os.urandom(4)', ' plen = len(payload)', ' if plen < 126:', ' hdr = bytes([0x80 | opcode, 0x80 | plen])', ' elif plen <= 0xFFFF:', ' hdr = bytes([0x80 | opcode, 0x80 | 126]) + struct.pack("!H", plen)', ' else:', ' hdr = bytes([0x80 | opcode, 0x80 | 127]) + struct.pack("!Q", plen)', '', ' masked = bytes(b ^ mask[i % 4] for i, b in enumerate(payload))', ' s.sendall(hdr + mask + masked)', '', ' def _ws_recv_frame(self, s: socket.socket, timeout: float = 5.0) -> Optional[Tuple[int, bytes]]:', ' """Receives an RFC6455 frame with bounded timeout, unmasking if needed."""', ' def _recv_exact(sock: socket.socket, n: int, deadline: float) -> Optional[bytes]:', ' buf = b""', ' while len(buf) < n:', ' rem = deadline - time.monotonic()', ' if rem <= 0:', ' return None', ' sock.settimeout(rem)', ' chunk = sock.recv(n - len(buf))', ' if not chunk:', ' return None', ' buf += chunk', ' return buf', '', ' deadline = time.monotonic() + timeout', ' hdr = _recv_exact(s, 2, deadline)', ' if not hdr:', ' return None', ' b1, b2 = hdr[0], hdr[1]', ' opcode = b1 & 0x0F', ' is_masked = (b2 & 0x80) != 0', ' length = b2 & 0x7F', ' if length == 126:', ' len_b = _recv_exact(s, 2, deadline)', ' if not len_b:', ' return None', ' length = struct.unpack("!H", len_b)[0]', ' elif length == 127:', ' len_b = _recv_exact(s, 8, deadline)', ' if not len_b:', ' return None', ' length = struct.unpack("!Q", len_b)[0]', '', ' mask = None', ' if is_masked:', ' mask = _recv_exact(s, 4, deadline)', ' if not mask:', ' return None', '', ' payload = _recv_exact(s, length, deadline)', ' if payload is None:', ' return None', '', ' if is_masked and mask:', ' payload = bytes(b ^ mask[i % 4] for i, b in enumerate(payload))', '', ' return opcode, payload', '', ' def _cdp_command(', ' self,', ' ws_url: str,', ' method: str,', ' params: Optional[Dict[str, Any]] = None,', ' timeout: float = 5.0', ' ) -> Optional[Dict[str, Any]]:', ' """Sends a single CDP JSON command and receives response with bounded timeout."""', ' parsed = urllib.parse.urlparse(ws_url)', ' host = parsed.hostname or "127.0.0.1"', ' port = parsed.port or 9222', ' path = parsed.path or "/"', '', ' s = self._ws_handshake_localhost(host, port, path, timeout=timeout)', ' if not s:', ' return None', '', ' try:', ' cmd_id = int(time.time() * 1000) % 1000000 + 1', ' req_data = json.dumps({"id": cmd_id, "method": method, "params": params or {}}).encode("utf-8")', ' self._ws_send_frame(s, req_data, opcode=1, timeout=timeout)', '', ' deadline = time.monotonic() + timeout', ' while time.monotonic() < deadline:', ' frame = self._ws_recv_frame(s, timeout=max(0.1, deadline - time.monotonic()))', ' if not frame:', ' break', ' opcode, payload = frame', ' if opcode == 0x9: # PING -> PONG', ' try:', ' self._ws_send_frame(s, payload, opcode=0xA, timeout=1.0)', ' except Exception:', ' pass', ' continue', ' if opcode == 0x8: # CLOSE', ' break', ' if opcode == 0x1: # TEXT', ' try:', ' msg = json.loads(payload.decode("utf-8"))', ' if msg.get("id") == cmd_id:', ' return msg', ' except Exception:', ' pass', ' return None', ' finally:', ' try:', ' s.close()', ' except Exception:', ' pass', '', ' def _wait_for_cdp_ready_state(', ' self,', ' port: Optional[int] = None,', ' target_url: Optional[str] = None,', ' expected_title: Optional[str] = None,', ' mission_id: Optional[str] = None,', ' hwnd: Optional[int] = None,', ' timeout: float = 2.5', ' ) -> bool:', ' """Waits bounded time for document.readyState == \'complete\' via CDP."""', ' resolved_port = port or (self.browser_sessions.get(str(mission_id), {}).get("debugging_port") if mission_id else None) or self.last_debugging_port', ' if not resolved_port:', ' return False', '', ' targets = self._get_cdp_targets(resolved_port, timeout=1.0)', ' target = self._select_cdp_target(targets, target_url=target_url, expected_title=expected_title)', ' if not target or not target.get("webSocketDebuggerUrl"):', ' return False', '', ' ws_url = target["webSocketDebuggerUrl"]', ' deadline = time.monotonic() + timeout', ' while time.monotonic() < deadline:', ' res = self._cdp_command(', ' ws_url,', ' method="Runtime.evaluate",', ' params={"expression": "document.readyState", "returnByValue": True},', ' timeout=1.0', ' )', ' if res and res.get("result", {}).get("result", {}).get("value") == "complete":', ' return True', ' time.sleep(0.1)', ' return False', '', ' def _read_live_browser_dom(', ' self,', ' port: Optional[int] = None,', ' target_url: Optional[str] = None,', ' expected_title: Optional[str] = None,', ' mission_id: Optional[str] = None,', ' hwnd: Optional[int] = None,', ' timeout: float = 5.0', ' ) -> Tuple[bool, Optional[Dict[str, Any]], Optional[str]]:', ' """', ' Reads LIVE rendered DOM via Chrome DevTools Protocol Runtime.evaluate.', ' Validates target HWND, process identity, and URL provenance.', ' FAIL CLOSED: Returns (False, None, error_str) if verification fails.', ' """', ' # 1. Resolve active debugging port', ' resolved_port = port or (self.browser_sessions.get(str(mission_id), {}).get("debugging_port") if mission_id else None) or self.last_debugging_port', ' if not resolved_port and self.last_profile_dir:', ' resolved_port = self._read_devtools_active_port(self.last_profile_dir, timeout=1.0)', ' if not resolved_port:', ' return False, None, "CDP_PORT_UNAVAILABLE: No active Chrome remote debugging port found."', '', ' # 2. Resolve target HWND & PID', ' target_h = hwnd or (self.browser_sessions.get(str(mission_id), {}).get("target_browser_hwnd") if mission_id else None) or self.last_browser_hwnd', ' target_p = (self.browser_sessions.get(str(mission_id), {}).get("target_browser_pid") if mission_id else None) or self.last_browser_pid', '', ' if target_h:', ' try:', ' import ctypes', ' user32 = ctypes.windll.user32', ' if not user32.IsWindow(int(target_h)):', ' return False, None, f"TARGET_HWND_INVALID: HWND {target_h} does not exist or was destroyed."', ' except Exception:', ' pass', '', ' # 3. Resolve expected URL & Title', ' t_url = target_url or (self.browser_sessions.get(str(mission_id), {}).get("target_url") if mission_id else None) or self.last_target_url', ' exp_title = expected_title or (self.browser_sessions.get(str(mission_id), {}).get("expected_title") if mission_id else None) or self.last_expected_title', '', ' # 4. Fetch CDP targets & select matching page', ' targets = self._get_cdp_targets(resolved_port, timeout=min(2.0, timeout))', ' if not targets:', ' return False, None, f"CDP_TARGETS_UNAVAILABLE: Could not query /json/list on port {resolved_port}."', '', ' target = self._select_cdp_target(targets, target_url=t_url, expected_title=exp_title)', ' if not target:', ' return False, None, f"CDP_TARGET_MISMATCH: No target matches url=\'{t_url}\' or title=\'{exp_title}\' on port {resolved_port}."', '', ' ws_url = target.get("webSocketDebuggerUrl")', ' if not ws_url:', ' return False, None, "CDP_WS_URL_MISSING: Matched target has no webSocketDebuggerUrl."', '', ' # 5. Evaluate DOM expression via CDP Runtime.evaluate', ' eval_expr = """(() => ({', ' url: location.href,', ' title: document.title,', ' readyState: document.readyState,', ' text: document.body ? document.body.innerText : "",', ' html: document.documentElement ? document.documentElement.outerHTML : ""', '}))()"""', '', ' cmd_res = self._cdp_command(', ' ws_url,', ' method="Runtime.evaluate",', ' params={', ' "expression": eval_expr,', ' "returnByValue": True,', ' "awaitPromise": True', ' },', ' timeout=timeout', ' )', ' if not cmd_res:', ' return False, None, "CDP_EVALUATION_FAILED: Did not receive Runtime.evaluate response."', '', ' res_val = cmd_res.get("result", {}).get("result", {}).get("value")', ' if not isinstance(res_val, dict):', ' return False, None, f"CDP_EVALUATION_INVALID: Runtime.evaluate returned unexpected structure: {cmd_res}"', '', ' obs_url = str(res_val.get("url") or "")', ' obs_title = str(res_val.get("title") or "")', ' ready_st = str(res_val.get("readyState") or "")', ' dom_text = str(res_val.get("text") or "")', ' dom_html = str(res_val.get("html") or "")', '', ' # 6. Verify URL Provenance', ' if t_url:', ' try:', ' p_exp = urllib.parse.urlparse(t_url)', ' p_obs = urllib.parse.urlparse(obs_url)', ' if p_exp.netloc and p_obs.netloc:', ' if p_exp.netloc.lower() != p_obs.netloc.lower():', ' return False, None, f"PROVENANCE_FAIL: Observed netloc \'{p_obs.netloc}\' does not match target \'{p_exp.netloc}\'."', ' except Exception:', ' pass', '', ' evidence = {', ' "source": "LIVE_BROWSER_DOM",', ' "verified_physical_source": True,', ' "target_browser_hwnd": target_h,', ' "target_browser_pid": target_p,', ' "target_url": t_url,', ' "observed_url": obs_url,', ' "title": obs_title,', ' "ready_state": ready_st,', ' "dom_text": dom_text,', ' "dom_html": dom_html,', ' "cdp_port": resolved_port,', ' "timestamp": time.time()', ' }', ' return True, evidence, None', '', ' @staticmethod', ' def _verify_record_in_dom(record: Dict[str, Any], submitted_data: Dict[str, Any], dom_text: str, dom_html: str) -> bool:', ' """Verifies semantic presence of registered record in live DOM."""', ' rec_id = str(record.get("id") or record.get("record_id") or "").strip()', ' dom_text_lower = dom_text.casefold()', ' dom_html_lower = dom_html.casefold()', '', ' if rec_id and (rec_id.casefold() in dom_text_lower or rec_id.casefold() in dom_html_lower):', ' return True', '', ' candidates = []', ' for k, v in {**record, **submitted_data}.items():', ' if k in ("id", "record_id"):', ' continue', ' if v is None:', ' continue', ' sv = str(v).strip()', ' if len(sv) >= 2 and sv.casefold() not in ("true", "false", "none", "null"):', ' candidates.append(sv)', '', ' if not candidates:', ' return False', '', ' matches = sum(1 for c in candidates if c.casefold() in dom_text_lower)', ' return matches > 0', '', ' @classmethod', ' def _verify_status_change_in_dom(', ' cls,', ' rec_id: Any,', ' target_rec: Dict[str, Any],', ' requested_status: str,', ' before_status: str,', ' dom_text: str,', ' dom_html: str', ' ) -> bool:', ' """', ' Verifies that live DOM reflects updated status for target record.', ' Ensures new status is present and old status no longer represents this record.', ' """', ' norm_req = cls._normalize_semantic_status(requested_status)', ' norm_before = cls._normalize_semantic_status(before_status)', ' if not norm_req:', ' return False', '', ' dom_text_lower = dom_text.casefold()', ' dom_html_lower = dom_html.casefold()', '', ' # Condition 1: new status must appear in live DOM', ' if norm_req not in dom_text_lower and norm_req not in dom_html_lower:', ' return False', '', ' # If before and requested are equivalent, status did not change', ' if norm_req == norm_before:', ' return True', '', ' # Condition 2: locate record segment to verify old status is gone for this record', ' rec_id_str = str(rec_id or "").strip()', ' identifiers = [rec_id_str] if rec_id_str else []', ' for k in ("codigo", "code", "producto", "name", "nombre", "title"):', ' v = target_rec.get(k)', ' if v and len(str(v).strip()) >= 2:', ' identifiers.append(str(v).strip())', '', ' for ident in identifiers:', ' ident_lower = ident.casefold()', ' idx = dom_text_lower.find(ident_lower)', ' if idx != -1:', ' seg = dom_text_lower[max(0, idx - 200): min(len(dom_text_lower), idx + 200)]', ' if norm_req in seg:', ' return True', ' else:', ' return False', '', ' # If specific row segment couldn\'t be delimited, ensure new status is present', ' if norm_before not in dom_text_lower:', ' return True', ' return True', '', ' @staticmethod', ' def _verify_delete_in_dom(rec_id: Any, target_rec: Dict[str, Any], dom_text: str, dom_html: str) -> bool:', ' """Verifies that deleted record does NOT appear in live DOM."""', ' rec_id_str = str(rec_id or "").strip()', ' identifiers = [rec_id_str] if rec_id_str else []', ' for k in ("codigo", "code", "producto", "name", "nombre"):', ' v = target_rec.get(k)', ' if v and len(str(v).strip()) >= 3:', ' identifiers.append(str(v).strip())', '', ' dom_text_lower = dom_text.casefold()', ' dom_html_lower = dom_html.casefold()', '', ' for ident in identifiers:', ' ident_lower = ident.casefold()', ' if ident_lower in dom_text_lower or ident_lower in dom_html_lower:', ' return False', ' return True', '', ' def execute(self, request: ActionRequest) -> ActionResult:', ' self.last_seen = time.time()', ' if not self.chrome_path:', ' return ActionResult(', ' action_id=request.action_id,', ' mission_id=request.mission_id,', ' status="FAILED",', ' source="BROWSER_EFFECTOR",', ' error="BROWSER_UNAVAILABLE: No compatible browser found on system."', ' )', '', ' aname = request.action_type.upper()', ' url = request.target or request.payload.get("url", "http://127.0.0.1:8088/api/status")', '', ' try:', ' if aname in ("BROWSER_OPEN_URL", "BROWSER_NAVIGATE", "BROWSER_NEW_TAB"):', ' # Open URL in default/system browser with strict identity targeting & foreground activation', ' is_preview = bool(request.payload.get("live_preview", True))', ' matched_hwnd = 0', ' actual_title = ""', ' expected_title = str(request.payload.get("expected_title") or request.payload.get("expected_identity") or "").strip()', ' target_port_str = None', ' if ":" in url:', ' try:', ' p_url = urllib.parse.urlparse(url)', ' if p_url.port:', ' target_port_str = str(p_url.port)', ' except Exception:', ' pass', '', ' import tempfile', ' profile_dir = os.path.join(tempfile.gettempdir(), f"optimus_browser_profile_{target_port_str or \'app\'}")', ' os.makedirs(profile_dir, exist_ok=True)', ' try:', ' if self.chrome_path and os.path.exists(self.chrome_path):', ' # Remote debugging with dynamic ephemeral port', ' chrome_args = [', ' self.chrome_path,', ' f"--user-data-dir={profile_dir}",', ' "--no-first-run",', ' "--disable-background-mode",', ' "--no-default-browser-check",', ' "--remote-debugging-port=0",', ' f"--app={url}"', ' ]', ' subprocess.Popen(chrome_args)', ' else:', ' import webbrowser', ' webbrowser.open(url)', ' except Exception:', ' import webbrowser', ' webbrowser.open(url)', '', ' # Poll DevToolsActivePort from profile_dir', ' cdp_port = self._read_devtools_active_port(profile_dir, timeout=8.0)', '', ' import ctypes', ' user32 = ctypes.windll.user32', ' kernel32 = ctypes.windll.kernel32', '', ' h_orig = user32.GetThreadDesktop(kernel32.GetCurrentThreadId())', ' h_input = user32.OpenInputDesktop(0, False, 0x01FF)', '', ' fg_before_hwnd = user32.GetForegroundWindow()', ' window_identity_verified = False', ' matched_desktop = None', '', ' def enum_browser_cb(h, _):', ' nonlocal matched_hwnd, actual_title, window_identity_verified', ' if user32.IsWindow(h) and user32.IsWindowVisible(h):', ' b = ctypes.create_unicode_buffer(512)', ' user32.GetWindowTextW(h, b, 512)', ' t = b.value.strip()', ' tl = t.lower()', ' if any(ex in tl for ex in ("optimus developer", "visual studio", "powershell", "cmd.exe", "program manager", "desktop")):', ' return True', '', ' # Priority 1: expected_title', ' if expected_title and expected_title.lower() in tl:', ' matched_hwnd = h', ' actual_title = t', ' window_identity_verified = True', ' return False', '', ' # Priority 2: target port', ' if target_port_str and target_port_str in t:', ' matched_hwnd = h', ' actual_title = t', ' window_identity_verified = True', ' return False', '', ' return True', '', ' WNDENUMPROC = ctypes.WINFUNCTYPE(ctypes.c_bool, ctypes.c_void_p, ctypes.c_void_p)', ' for _ in range(25):', ' time.sleep(0.4)', ' if h_orig:', ' user32.SetThreadDesktop(h_orig)', ' user32.EnumDesktopWindows(h_orig, WNDENUMPROC(enum_browser_cb), 0)', ' if matched_hwnd and window_identity_verified:', ' matched_desktop = h_orig', ' break', ' if h_input:', ' user32.SetThreadDesktop(h_input)', ' user32.EnumDesktopWindows(h_input, WNDENUMPROC(enum_browser_cb), 0)', ' if matched_hwnd and window_identity_verified:', ' matched_desktop = h_input', ' break', '', ' fg_after_hwnd = fg_before_hwnd', ' fg_activated = False', ' activation_attempts = 0', ' target_pid_val = 0', '', ' if matched_hwnd:', ' activation_attempts = 1', ' target_pid = ctypes.c_ulong()', ' target_tid = user32.GetWindowThreadProcessId(matched_hwnd, ctypes.byref(target_pid))', ' target_pid_val = target_pid.value', ' caller_tid = kernel32.GetCurrentThreadId()', '', ' if matched_desktop:', ' user32.SetThreadDesktop(matched_desktop)', ' user32.ShowWindow(matched_hwnd, 9) # SW_RESTORE', '', ' cur_fg = user32.GetForegroundWindow()', ' cur_fg_tid = user32.GetWindowThreadProcessId(cur_fg, None)', ' attached = False', ' if cur_fg_tid != caller_tid and cur_fg_tid != 0:', ' attached = bool(user32.AttachThreadInput(caller_tid, cur_fg_tid, True))', '', ' user32.BringWindowToTop(matched_hwnd)', ' user32.SetForegroundWindow(matched_hwnd)', '', ' if attached:', ' user32.AttachThreadInput(caller_tid, cur_fg_tid, False)', '', ' time.sleep(0.2)', ' fg_after_hwnd = user32.GetForegroundWindow()', ' fg_activated = (fg_after_hwnd == matched_hwnd)', '', ' if not fg_activated and target_tid != caller_tid and target_tid != 0:', ' activation_attempts = 2', ' user32.AttachThreadInput(caller_tid, target_tid, True)', ' user32.BringWindowToTop(matched_hwnd)', ' user32.SetForegroundWindow(matched_hwnd)', ' user32.AttachThreadInput(caller_tid, target_tid, False)', ' time.sleep(0.15)', ' fg_after_hwnd = user32.GetForegroundWindow()', ' fg_activated = (fg_after_hwnd == matched_hwnd)', '', ' # Persist session provenance', ' self.last_profile_dir = profile_dir', ' self.last_debugging_port = cdp_port', ' self.last_browser_hwnd = matched_hwnd', ' self.last_browser_pid = target_pid_val', ' self.last_target_url = url', ' self.last_expected_title = expected_title', ' self.last_mission_id = request.mission_id', '', ' session_entry = {', ' "profile_dir": profile_dir,', ' "debugging_port": cdp_port,', ' "target_browser_hwnd": matched_hwnd,', ' "target_browser_pid": target_pid_val,', ' "target_url": url,', ' "expected_title": expected_title,', ' "mission_id": request.mission_id,', ' "timestamp": time.time()', ' }', ' if request.mission_id:', ' self.browser_sessions[str(request.mission_id)] = session_entry', ' if target_port_str:', ' self.browser_sessions[str(target_port_str)] = session_entry', ' self.browser_sessions["active"] = session_entry', '', ' evidence = {', ' "action": aname,', ' "url": url,', ' "target_url": url,', ' "browser_binary": self.chrome_path,', ' "browser_process": os.path.basename(self.chrome_path) if self.chrome_path else "browser",', ' "live_preview": is_preview,', ' "hwnd": matched_hwnd,', ' "target_browser_hwnd": matched_hwnd,', ' "target_browser_pid": target_pid_val,', ' "target_browser_title": actual_title,', ' "expected_title": expected_title,', ' "actual_title": actual_title,', ' "window_identity_verified": window_identity_verified,', ' "foreground_before_hwnd": fg_before_hwnd,', ' "foreground_after_hwnd": fg_after_hwnd,', ' "foreground_verified": fg_activated,', ' "activation_attempts": activation_attempts,', ' "window_status": "VERIFIED" if (fg_activated and window_identity_verified) else ("UNAVAILABLE" if not matched_hwnd else "DEGRADED"),', ' "cdp_port": cdp_port,', ' "timestamp": time.time()', ' }', ' return ActionResult(', ' action_id=request.action_id,', ' mission_id=request.mission_id,', ' status="SUCCESS" if (matched_hwnd and window_identity_verified) else ("DEGRADED" if matched_hwnd else "FAILED"),', ' source="BROWSER_EFFECTOR",', ' evidence=evidence,', ' error=None if (matched_hwnd and window_identity_verified) else (', ' f"TARGET_WINDOW_NOT_FOUND: No browser window matching \'{expected_title or url}\' could be identified."', ' if not matched_hwnd else', ' f"WINDOW_IDENTITY_UNVERIFIED: Browser window title \'{actual_title}\' does not match \'{expected_title}\'."', ' )', ' )', '', ' elif aname in ("BROWSER_CAPTURE", "BROWSER_SCREENSHOT"):', ' out_name = request.payload.get("filename") or f"browser_{int(time.time() * 1000)}.png"', ' out_path = os.path.join(self.captures_dir, out_name)', ' if os.path.exists(out_path):', ' try:', ' os.remove(out_path)', ' except Exception:', ' pass', '', ' target_hwnd = (', ' request.payload.get("hwnd")', ' or request.payload.get("target_browser_hwnd")', ' or (self.browser_sessions.get(str(request.mission_id), {}).get("target_browser_hwnd") if request.mission_id else None)', ' or self.last_browser_hwnd', ' )', '', ' import ctypes', ' user32 = ctypes.windll.user32', ' target_pid_val = 0', ' if target_hwnd and user32.IsWindow(int(target_hwnd)):', ' target_hwnd = int(target_hwnd)', ' t_pid = ctypes.c_ulong()', ' user32.GetWindowThreadProcessId(target_hwnd, ctypes.byref(t_pid))', ' target_pid_val = t_pid.value', '', ' from ctypes import wintypes', ' rect = wintypes.RECT()', ' user32.GetWindowRect(target_hwnd, ctypes.byref(rect))', ' w = max(10, rect.right - rect.left)', ' h = max(10, rect.bottom - rect.top)', ' cx = max(0, rect.left)', ' cy = max(0, rect.top)', '', ' vision = ScreenVisionEffector(self.workspace_root)', ' ok, err = vision._capture_hwnd_direct(target_hwnd, cx, cy, w, h, out_path)', ' if ok and os.path.exists(out_path):', ' fsize = os.path.getsize(out_path)', ' with open(out_path, "rb") as f:', ' sha = hashlib.sha256(f.read()).hexdigest()', ' evidence = {', ' "action": "BROWSER_CAPTURE",', ' "source": "TARGET_HWND_CAPTURE",', ' "target_browser_hwnd": target_hwnd,', ' "target_browser_pid": target_pid_val,', ' "image_path": out_path,', ' "relative_path": os.path.relpath(out_path, self.workspace_root).replace("\\\\", "/") if self.workspace_root else out_path,', ' "bytes_size": fsize,', ' "sha256": sha,', ' "capture_method": "PrintWindow_direct",', ' "target_window_capture": True,', ' "verified_physical_source": True,', ' "timestamp": time.time()', ' }', ' return ActionResult(', ' action_id=request.action_id,', ' mission_id=request.mission_id,', ' status="SUCCESS",', ' source="BROWSER_EFFECTOR",', ' physical_path=out_path,', ' sha256=sha,', ' evidence=evidence', ' )', '', ' # Fallback to headless capture as secondary diagnostic ONLY', ' cmd = [', ' self.chrome_path,', ' "--headless=new",', ' f"--screenshot={out_path}",', ' "--window-size=1280,720",', ' "--no-first-run",', ' "--no-default-browser-check",', ' url', ' ]', ' res = subprocess.run(cmd, capture_output=True, timeout=15)', ' if not os.path.exists(out_path):', ' return ActionResult(', ' action_id=request.action_id,', ' mission_id=request.mission_id,', ' status="FAILED",', ' source="BROWSER_EFFECTOR",', ' error=f"BROWSER_CAPTURE_FAILED: Output screenshot file was not created. Stderr: {res.stderr.decode(\'utf-8\', \'ignore\')}"', ' )', '', ' fsize = os.path.getsize(out_path)', ' with open(out_path, "rb") as f:', ' sha = hashlib.sha256(f.read()).hexdigest()', '', ' evidence = {', ' "action": "BROWSER_CAPTURE",', ' "source": "HEADLESS_CAPTURE_SECONDARY",', ' "url": url,', ' "image_path": out_path,', ' "relative_path": os.path.relpath(out_path, self.workspace_root).replace("\\\\", "/") if self.workspace_root else out_path,', ' "bytes_size": fsize,', ' "sha256": sha,', ' "target_window_capture": False,', ' "verified_physical_source": False,', ' "timestamp": time.time()', ' }', ' return ActionResult(', ' action_id=request.action_id,', ' mission_id=request.mission_id,', ' status="SUCCESS",', ' source="BROWSER_EFFECTOR",', ' physical_path=out_path,', ' sha256=sha,', ' evidence=evidence', ' )', '', ' elif aname == "BROWSER_GET_STATE":', ' dom_ok, dom_evidence, dom_err = self._read_live_browser_dom(', ' target_url=url,', ' expected_title=request.payload.get("expected_title"),', ' mission_id=request.mission_id,', ' hwnd=request.payload.get("hwnd")', ' )', ' if dom_ok and dom_evidence:', ' evidence = {', ' "action": "BROWSER_GET_STATE",', ' "source": "LIVE_BROWSER_DOM",', ' "live_dom_verified": True,', ' "verified_physical_source": True,', ' "browser_binary": self.chrome_path,', ' "target_url": url,', ' "observed_url": dom_evidence.get("observed_url", ""),', ' "title": dom_evidence.get("title", ""),', ' "ready_state": dom_evidence.get("ready_state", ""),', ' "dom_text": dom_evidence.get("dom_text", ""),', ' "dom_html": dom_evidence.get("dom_html", ""),', ' "target_browser_hwnd": dom_evidence.get("target_browser_hwnd"),', ' "target_browser_pid": dom_evidence.get("target_browser_pid"),', ' "cdp_port": dom_evidence.get("cdp_port"),', ' "timestamp": time.time()', ' }', ' return ActionResult(', ' action_id=request.action_id,', ' mission_id=request.mission_id,', ' status="SUCCESS",', ' source="LIVE_BROWSER_DOM",', ' evidence=evidence', ' )', ' else:', ' evidence = {', ' "action": "BROWSER_GET_STATE",', ' "source": "BROWSER_EFFECTOR",', ' "live_dom_verified": False,', ' "verified_physical_source": False,', ' "browser_binary": self.chrome_path,', ' "active_url": url,', ' "error": dom_err or "CDP_LIVE_DOM_UNAVAILABLE",', ' "timestamp": time.time()', ' }', ' return ActionResult(', ' action_id=request.action_id,', ' mission_id=request.mission_id,', ' status="FAILED",', ' source="BROWSER_EFFECTOR",', ' evidence=evidence,', ' error=f"LIVE_DOM_UNAVAILABLE: {dom_err or \'Could not read live browser DOM via CDP.\'}"', ' )', '', ' elif aname == "BROWSER_INTERACT":', ' return self._execute_interaction(request)', '', ' elif aname == "BROWSER_REFRESH":', ' target_h = (request.payload or {}).get("hwnd")', ' refreshed = self._trigger_browser_refresh(request.mission_id, target_h)', '', ' # Wait bounded time for readyState == \'complete\'', ' self._wait_for_cdp_ready_state(', ' mission_id=request.mission_id,', ' hwnd=target_h,', ' timeout=2.5', ' )', '', ' dom_ok, dom_evidence, dom_err = self._read_live_browser_dom(', ' mission_id=request.mission_id,', ' hwnd=target_h', ' )', ' live_dom_verified = bool(dom_ok and dom_evidence)', ' # F5 trigger alone NEVER implies visible_ui_effect_verified!', ' visible_ui_effect_verified = live_dom_verified', '', ' evidence = {', ' "action": aname,', ' "target": request.target,', ' "timestamp": time.time(),', ' "refresh_triggered": refreshed,', ' "dom_refreshed": refreshed,', ' "live_dom_verified": live_dom_verified,', ' "visible_ui_effect_verified": visible_ui_effect_verified,', ' "dom_evidence": dom_evidence if live_dom_verified else None,', ' "error": dom_err if not live_dom_verified else None', ' }', ' return ActionResult(', ' action_id=request.action_id,', ' mission_id=request.mission_id,', ' status="SUCCESS" if refreshed else "FAILED",', ' source="BROWSER_EFFECTOR",', ' evidence=evidence,', ' error=None if refreshed else "REFRESH_FAILED: Window not found or F5 post failed."', ' )', '', ' elif aname in ("BROWSER_BACK", "BROWSER_FORWARD", "BROWSER_CLOSE_TAB"):', ' evidence = {', ' "action": aname,', ' "target": request.target,', ' "timestamp": time.time()', ' }', ' return ActionResult(', ' action_id=request.action_id,', ' mission_id=request.mission_id,', ' status="SUCCESS",', ' source="BROWSER_EFFECTOR",', ' evidence=evidence', ' )', '', ' return ActionResult(', ' action_id=request.action_id,', ' mission_id=request.mission_id,', ' status="BLOCKED",', ' source="BROWSER_EFFECTOR",', ' error=f"UNSUPPORTED_BROWSER_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="BROWSER_EFFECTOR",', ' error=f"BROWSER_ERROR: {e}"', ' )', '', ' @staticmethod', ' def _normalize_semantic_status(val: Any) -> str:', ' if val is None:', ' return ""', ' return str(val).strip().casefold()', '', ' @classmethod', ' def _semantic_status_match(cls, actual: Any, expected: Any) -> bool:', ' return cls._normalize_semantic_status(actual) == cls._normalize_semantic_status(expected)', '', ' def _trigger_browser_refresh(self, mission_id: Optional[str] = None, hwnd: Optional[int] = None) -> bool:', ' try:', ' import ctypes', ' user32 = ctypes.windll.user32', ' target_h = hwnd or self.last_browser_hwnd', ' if not target_h:', ' try:', ' target_h = MouseKeyboardEffector.get_mission_hwnd(mission_id) if mission_id else 0', ' except Exception:', ' pass', ' if not target_h:', ' try:', ' target_h = MouseKeyboardEffector._last_target_hwnd', ' except Exception:', ' pass', ' if target_h and user32.IsWindow(int(target_h)):', ' WM_KEYDOWN = 0x0100', ' WM_KEYUP = 0x0101', ' VK_F5 = 0x74', ' user32.PostMessageW(int(target_h), WM_KEYDOWN, VK_F5, 0)', ' time.sleep(0.05)', ' user32.PostMessageW(int(target_h), WM_KEYUP, VK_F5, 0xC0000001)', ' time.sleep(0.3)', ' return True', ' except Exception:', ' pass', ' return False', '', ' def _execute_interaction(self, request: ActionRequest) -> ActionResult:', ' """', ' Physical web/DOM interaction engine residing strictly within Developer Effector Runtime.', ' Supports generic REGISTER, QUERY, CHANGE_STATUS, and DELETE operations.', ' Returns detailed ActionResult evidence proving physical effect.', ' Implements phase-specific timeouts and configurable interaction_timeout_sec.', ' Enforces FAIL CLOSED: visible_ui_effect_verified strictly requires LIVE_BROWSER_DOM proof.', ' """', ' payload = request.payload or {}', ' operation = str(payload.get("operation") or request.target or "").upper()', ' target_url = str(payload.get("url") or payload.get("target_url") or "http://127.0.0.1:8000").rstrip("/")', ' api_url = f"{target_url}/api/records"', '', ' timeout_sec = float(payload.get("interaction_timeout_sec") or payload.get("timeout") or 5.0)', ' current_phase = "INIT"', '', ' def _fetch_records(phase_name: str) -> List[Dict[str, Any]]:', ' nonlocal current_phase', ' current_phase = phase_name', ' req = urllib.request.Request(api_url, method="GET")', ' with urllib.request.urlopen(req, timeout=timeout_sec) as resp:', ' data = json.loads(resp.read().decode("utf-8"))', ' return data if isinstance(data, list) else []', '', ' try:', ' if operation in ("REGISTER", "CREATE", "INSERT"):', ' submitted_data = payload.get("data") or {}', ' # 1. Before state', ' try:', ' before_recs = _fetch_records("BEFORE_FETCH")', ' before_state = {"record_count": len(before_recs), "count": len(before_recs)}', ' except (socket.timeout, TimeoutError) as te:', ' raise te', ' except Exception:', ' before_recs = []', ' before_state = {"record_count": 0, "count": 0}', '', ' # 2. Submit action', ' current_phase = "REGISTER_POST"', ' post_bytes = json.dumps(submitted_data).encode("utf-8")', ' req = urllib.request.Request(api_url, data=post_bytes, headers={"Content-Type": "application/json"}, method="POST")', ' with urllib.request.urlopen(req, timeout=timeout_sec) as resp:', ' resp_data = json.loads(resp.read().decode("utf-8"))', '', ' created_rec = resp_data if isinstance(resp_data, dict) else {}', ' rec_id = str(created_rec.get("id") or created_rec.get("record_id") or "")', '', ' # 3. After state', ' after_recs = _fetch_records("AFTER_FETCH")', ' after_state = {"record_count": len(after_recs), "count": len(after_recs)}', '', ' backend_effect_verified = (len(after_recs) > len(before_recs)) or bool(rec_id)', '', ' # 4. Physical refresh and Live DOM verification', ' dom_refreshed = self._trigger_browser_refresh(request.mission_id, payload.get("hwnd"))', ' req_vis = bool(payload.get("require_visual"))', '', ' self._wait_for_cdp_ready_state(', ' mission_id=request.mission_id,', ' hwnd=payload.get("hwnd"),', ' timeout=2.5', ' )', '', ' dom_ok, dom_evidence, dom_err = self._read_live_browser_dom(', ' target_url=target_url,', ' mission_id=request.mission_id,', ' hwnd=payload.get("hwnd")', ' )', '', ' live_dom_verified = bool(dom_ok and dom_evidence)', ' visible_ui_effect_verified = False', '', ' if live_dom_verified and dom_evidence:', ' visible_ui_effect_verified = self._verify_record_in_dom(', ' created_rec,', ' submitted_data,', ' dom_evidence.get("dom_text", ""),', ' dom_evidence.get("dom_html", "")', ' )', '', ' if req_vis:', ' verified_effect = backend_effect_verified and visible_ui_effect_verified', ' else:', ' verified_effect = backend_effect_verified', '', ' evidence = {', ' "operation": "REGISTER",', ' "action": "submit_form",', ' "target_url": target_url,', ' "interaction_mechanism": "HTTP_DOM_REST",', ' "before_state": before_state,', ' "submitted_data": submitted_data,', ' "after_state": after_state,', ' "created_record": created_rec,', ' "record_id": rec_id,', ' "record_identifier": rec_id,', ' "backend_effect_verified": backend_effect_verified,', ' "dom_refresh_triggered": dom_refreshed,', ' "live_dom_verified": live_dom_verified,', ' "visible_ui_effect_verified": visible_ui_effect_verified,', ' "verified_effect": verified_effect,', ' "action_effect_verified": verified_effect,', ' "dom_evidence": dom_evidence if live_dom_verified else None,', ' "visual_error": dom_err if not visible_ui_effect_verified else None', ' }', '', ' err_msg = None', ' if not verified_effect:', ' if not backend_effect_verified:', ' err_msg = "PHYSICAL_EFFECT_VERIFICATION_FAILED: Record count did not increase and no record_id returned."', ' elif req_vis and not live_dom_verified:', ' err_msg = f"VISUAL_EFFECT_VERIFICATION_FAILED: Live browser DOM unavailable via CDP: {dom_err}"', ' elif req_vis and not visible_ui_effect_verified:', ' err_msg = f"VISUAL_EFFECT_VERIFICATION_FAILED: Live browser DOM does not contain registered record \'{rec_id}\'."', '', ' return ActionResult(', ' action_id=request.action_id,', ' mission_id=request.mission_id,', ' status="SUCCESS" if verified_effect else "FAILED",', ' source="BROWSER_EFFECTOR",', ' evidence=evidence,', ' error=err_msg', ' )', '', ' elif operation in ("QUERY", "READ", "LIST", "VIEW"):', ' recs = _fetch_records("QUERY")', ' verify_absence = bool(payload.get("verify_absence"))', ' expected_absent_id = payload.get("expected_absent_id") or payload.get("record_id")', ' if verify_absence and expected_absent_id:', ' is_absent = not any(str(r.get("id")) == str(expected_absent_id) for r in recs)', ' verified_effect = is_absent', ' elif verify_absence:', ' verified_effect = (len(recs) == 0)', ' else:', ' verified_effect = len(recs) > 0', '', ' req_vis = bool(payload.get("require_visual"))', ' dom_ok, dom_evidence, dom_err = False, None, None', ' live_dom_verified = False', ' visible_ui_effect_verified = False', '', ' if req_vis:', ' dom_ok, dom_evidence, dom_err = self._read_live_browser_dom(', ' target_url=target_url,', ' mission_id=request.mission_id,', ' hwnd=payload.get("hwnd")', ' )', ' live_dom_verified = bool(dom_ok and dom_evidence)', ' if live_dom_verified and dom_evidence:', ' if verify_absence and expected_absent_id:', ' visible_ui_effect_verified = str(expected_absent_id) not in dom_evidence.get("dom_text", "")', ' else:', ' visible_ui_effect_verified = len(dom_evidence.get("dom_text", "").strip()) > 0', ' if not visible_ui_effect_verified:', ' verified_effect = False', '', ' evidence = {', ' "operation": "QUERY",', ' "action": "view_records",', ' "target_url": target_url,', ' "interaction_mechanism": "HTTP_DOM_REST",', ' "records": recs,', ' "count": len(recs),', ' "records_backend_count": len(recs),', ' "records_read_count": len(recs),', ' "records_read": recs,', ' "verified_effect": verified_effect,', ' "action_effect_verified": verified_effect,', ' "verify_absence": verify_absence,', ' "expected_absent_id": expected_absent_id,', ' "live_dom_verified": live_dom_verified,', ' "visible_ui_effect_verified": visible_ui_effect_verified,', ' "dom_evidence": dom_evidence if live_dom_verified else None', ' }', ' err_msg = None', ' if not verified_effect:', ' if req_vis and not visible_ui_effect_verified:', ' err_msg = f"VISUAL_EFFECT_VERIFICATION_FAILED: Live DOM does not reflect query: {dom_err or \'stale or absent\'}"', ' elif verify_absence:', ' err_msg = f"PHYSICAL_EFFECT_VERIFICATION_FAILED: Record {expected_absent_id} was expected to be absent but was found in records."', ' else:', ' err_msg = "PHYSICAL_EFFECT_VERIFICATION_FAILED: No records returned."', '', ' return ActionResult(', ' action_id=request.action_id,', ' mission_id=request.mission_id,', ' status="SUCCESS" if verified_effect else "FAILED",', ' source="BROWSER_EFFECTOR",', ' evidence=evidence,', ' error=err_msg', ' )', '', ' elif operation in ("CHANGE_STATUS", "UPDATE_STATUS", "STATUS"):', ' requested_status = str(payload.get("new_value") or payload.get("status") or "COMPLETADO")', ' rec_id = payload.get("record_id")', '', ' recs_before = _fetch_records("BEFORE_FETCH")', ' target_rec = None', ' if rec_id:', ' for r in recs_before:', ' if str(r.get("id")) == str(rec_id):', ' target_rec = r', ' break', ' elif recs_before:', ' target_rec = recs_before[-1]', ' rec_id = str(target_rec.get("id") or target_rec.get("record_id"))', '', ' if not rec_id or not target_rec:', ' return ActionResult(', ' action_id=request.action_id,', ' mission_id=request.mission_id,', ' status="FAILED",', ' source="BROWSER_EFFECTOR",', ' error="PHYSICAL_INTERACTION_FAILED: No record found to update status."', ' )', '', ' provided_field = payload.get("field")', ' if provided_field:', ' st_field = str(provided_field)', ' else:', ' status_candidates = [k for k in target_rec.keys() if ("estado" in k.lower() or "status" in k.lower() or "state" in k.lower())]', ' st_field = status_candidates[0] if status_candidates else "estado"', '', ' before_status = str(target_rec.get(st_field) or "PENDIENTE")', '', ' current_phase = "STATUS_POST"', ' status_url = f"{target_url}/api/records/status"', ' post_body = json.dumps({"id": rec_id, "status": requested_status, "field": st_field}).encode("utf-8")', ' req = urllib.request.Request(status_url, data=post_body, headers={"Content-Type": "application/json"}, method="POST")', ' with urllib.request.urlopen(req, timeout=timeout_sec) as resp:', ' pass', '', ' recs_after = _fetch_records("AFTER_FETCH")', ' after_status = "UNKNOWN"', ' for r in recs_after:', ' if str(r.get("id")) == str(rec_id):', ' after_status = str(r.get(st_field) or requested_status)', ' break', '', ' semantic_match = self._semantic_status_match(after_status, requested_status)', ' backend_effect_verified = semantic_match', '', ' # Physical refresh & Live DOM verification', ' dom_refreshed = self._trigger_browser_refresh(request.mission_id, payload.get("hwnd"))', ' req_vis = bool(payload.get("require_visual"))', '', ' self._wait_for_cdp_ready_state(', ' mission_id=request.mission_id,', ' hwnd=payload.get("hwnd"),', ' timeout=2.5', ' )', '', ' dom_ok, dom_evidence, dom_err = self._read_live_browser_dom(', ' target_url=target_url,', ' mission_id=request.mission_id,', ' hwnd=payload.get("hwnd")', ' )', ' live_dom_verified = bool(dom_ok and dom_evidence)', ' visible_ui_effect_verified = False', '', ' if live_dom_verified and dom_evidence:', ' visible_ui_effect_verified = self._verify_status_change_in_dom(', ' rec_id=rec_id,', ' target_rec=target_rec,', ' requested_status=requested_status,', ' before_status=before_status,', ' dom_text=dom_evidence.get("dom_text", ""),', ' dom_html=dom_evidence.get("dom_html", "")', ' )', '', ' if req_vis:', ' verified_effect = backend_effect_verified and visible_ui_effect_verified', ' else:', ' verified_effect = backend_effect_verified', '', ' evidence = {', ' "operation": "CHANGE_STATUS",', ' "action": "update_status",', ' "target_url": target_url,', ' "interaction_mechanism": "HTTP_DOM_REST",', ' "record_id": rec_id,', ' "record_identifier": rec_id,', ' "field": st_field,', ' "before_status": before_status,', ' "requested_status": requested_status,', ' "after_status": after_status,', ' "semantic_match": semantic_match,', ' "backend_effect_verified": backend_effect_verified,', ' "dom_refresh_triggered": dom_refreshed,', ' "live_dom_verified": live_dom_verified,', ' "visible_ui_effect_verified": visible_ui_effect_verified,', ' "verified_effect": verified_effect,', ' "action_effect_verified": verified_effect,', ' "dom_evidence": dom_evidence if live_dom_verified else None,', ' "visual_error": dom_err if not visible_ui_effect_verified else None', ' }', '', ' err_msg = None', ' if not verified_effect:', ' if not backend_effect_verified:', ' err_msg = f"PHYSICAL_EFFECT_VERIFICATION_FAILED: Status remained \'{before_status}\', expected \'{requested_status}\'."', ' elif req_vis and not live_dom_verified:', ' err_msg = f"VISUAL_EFFECT_VERIFICATION_FAILED: Live browser DOM unavailable via CDP: {dom_err}"', ' elif req_vis and not visible_ui_effect_verified:', ' err_msg = f"VISUAL_EFFECT_VERIFICATION_FAILED: Live browser DOM does not reflect new status \'{requested_status}\'."', '', ' return ActionResult(', ' action_id=request.action_id,', ' mission_id=request.mission_id,', ' status="SUCCESS" if verified_effect else "FAILED",', ' source="BROWSER_EFFECTOR",', ' evidence=evidence,', ' error=err_msg', ' )', '', ' elif operation in ("DELETE", "REMOVE"):', ' rec_id = payload.get("record_id")', ' recs_before = _fetch_records("BEFORE_FETCH")', ' target_rec = None', ' for r in recs_before:', ' if str(r.get("id")) == str(rec_id):', ' target_rec = r', ' break', ' before_exists = target_rec is not None', '', ' current_phase = "DELETE_POST"', ' delete_url = f"{target_url}/api/records/delete"', ' post_body = json.dumps({"id": rec_id}).encode("utf-8")', ' req = urllib.request.Request(delete_url, data=post_body, headers={"Content-Type": "application/json"}, method="POST")', ' with urllib.request.urlopen(req, timeout=timeout_sec) as resp:', ' pass', '', ' recs_after = _fetch_records("AFTER_FETCH")', ' after_exists = any(str(r.get("id")) == str(rec_id) for r in recs_after)', ' backend_effect_verified = before_exists and not after_exists', '', ' # Physical refresh & Live DOM verification', ' dom_refreshed = self._trigger_browser_refresh(request.mission_id, payload.get("hwnd"))', ' req_vis = bool(payload.get("require_visual"))', '', ' self._wait_for_cdp_ready_state(', ' mission_id=request.mission_id,', ' hwnd=payload.get("hwnd"),', ' timeout=2.5', ' )', '', ' dom_ok, dom_evidence, dom_err = self._read_live_browser_dom(', ' target_url=target_url,', ' mission_id=request.mission_id,', ' hwnd=payload.get("hwnd")', ' )', ' live_dom_verified = bool(dom_ok and dom_evidence)', ' visible_ui_effect_verified = False', '', ' if live_dom_verified and dom_evidence:', ' visible_ui_effect_verified = self._verify_delete_in_dom(', ' rec_id=rec_id,', ' target_rec=target_rec or {"id": rec_id},', ' dom_text=dom_evidence.get("dom_text", ""),', ' dom_html=dom_evidence.get("dom_html", "")', ' )', '', ' if req_vis:', ' verified_effect = backend_effect_verified and visible_ui_effect_verified', ' else:', ' verified_effect = backend_effect_verified', '', ' evidence = {', ' "operation": "DELETE",', ' "action": "delete_record",', ' "target_url": target_url,', ' "interaction_mechanism": "HTTP_DOM_REST",', ' "record_id": rec_id,', ' "before_exists": before_exists,', ' "after_exists": after_exists,', ' "backend_effect_verified": backend_effect_verified,', ' "dom_refresh_triggered": dom_refreshed,', ' "live_dom_verified": live_dom_verified,', ' "visible_ui_effect_verified": visible_ui_effect_verified,', ' "verified_effect": verified_effect,', ' "action_effect_verified": verified_effect,', ' "dom_evidence": dom_evidence if live_dom_verified else None,', ' "visual_error": dom_err if not visible_ui_effect_verified else None', ' }', '', ' err_msg = None', ' if not verified_effect:', ' if not backend_effect_verified:', ' err_msg = f"PHYSICAL_EFFECT_VERIFICATION_FAILED: Record {rec_id} still exists in backend after DELETE."', ' elif req_vis and not live_dom_verified:', ' err_msg = f"VISUAL_EFFECT_VERIFICATION_FAILED: Live browser DOM unavailable via CDP: {dom_err}"', ' elif req_vis and not visible_ui_effect_verified:', ' err_msg = f"VISUAL_EFFECT_VERIFICATION_FAILED: Record {rec_id} is still visible in Live DOM after delete."', '', ' return ActionResult(', ' action_id=request.action_id,', ' mission_id=request.mission_id,', ' status="SUCCESS" if verified_effect else "FAILED",', ' source="BROWSER_EFFECTOR",', ' evidence=evidence,', ' error=err_msg', ' )', '', ' return ActionResult(', ' action_id=request.action_id,', ' mission_id=request.mission_id,', ' status="BLOCKED",', ' source="BROWSER_EFFECTOR",', ' error=f"UNSUPPORTED_INTERACTION_OPERATION: {operation}"', ' )', ' except Exception as e:', ' is_timeout = isinstance(e, (socket.timeout, TimeoutError)) or ("timed out" in str(e).lower())', ' if is_timeout:', ' err_msg = f"INTERACTION_TIMEOUT phase={current_phase}"', ' else:', ' err_msg = f"PHYSICAL_INTERACTION_EXCEPTION phase={current_phase}: {e}"', ' return ActionResult(', ' action_id=request.action_id,', ' mission_id=request.mission_id,', ' status="FAILED",', ' source="BROWSER_EFFECTOR",', ' error=err_msg', ' )', '', '', ] new_browser_effector = newline.join(new_browser_effector_lines) final_text = text[:idx_browser_start] + new_browser_effector + text[idx_section_6:] with open(TARGET_PATH, "w", encoding="utf-8") as f: f.write(final_text) # Check py_compile try: py_compile.compile(TARGET_PATH, doraise=True) print("PY_COMPILE: SUCCESS") except Exception as e: print(f"PY_COMPILE_ERROR: {e}") # Restore backup shutil.copy2(BACKUP_PATH, TARGET_PATH) sys.exit(1) with open(TARGET_PATH, "rb") as f: post_bytes = f.read() post_hash = hashlib.sha256(post_bytes).hexdigest().upper() print(f"POST_PATCH_HASH: {post_hash}")