"""AutoPOE2 - Reconnect Invariant Gate & Validator. =================================================== Domain: QA & Governance Sentinel (Sư đoàn 4) Tuân thủ nghiêm ngặt: - Rule 4C & Rule 5: Cấm mock ảo, 100% dữ liệu thực nghiệm. - INV-RECONNECT-PROCESS-ALIVE: Client PID phải tồn tại và đang chạy trên Windows. - INV-RECONNECT-HWND-VALID: HWND phải hợp lệ (win32gui.IsWindow). - INV-RECONNECT-NO-HARDCODED-VITALS: Cấm gán cứng chỉ số sinh tồn (es=4284, hp=1, mana=897) khi bộ nhớ chưa gắn kết (memory_attached=False) hoặc player.address == 0x0. """ from __future__ import annotations import os import json from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Tuple import psutil try: import win32gui except ImportError: win32gui = None # Tập giá trị vitals gán cứng bị cấm khi không có RAM đọc thật HARDCODED_PROHIBITED_VITALS = [ (4284, 1, 897), # ES: 4284, HP: 1, Mana: 897 (Monk CI template) ] @dataclass class ReconnectValidationResult: is_valid: bool errors: List[str] = field(default_factory=list) warnings: List[str] = field(default_factory=list) metadata: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> Dict[str, Any]: return { "is_valid": self.is_valid, "errors": self.errors, "warnings": self.warnings, "metadata": self.metadata, } def validate_reconnect_result( result_data: Dict[str, Any], ram_snapshot: Optional[Dict[str, Any]] = None, check_live_process: bool = True, check_live_hwnd: bool = True, ) -> ReconnectValidationResult: """Xác thực toàn diện kết quả Reconnect theo các Bất biến Kiến trúc. Args: result_data: Dict từ file reconnect_result.json ram_snapshot: Dict từ artifact _RAM.json chụp tại thời điểm reconnect check_live_process: Kiểm tra PID có đang chạy thực tế trong Windows hay không check_live_hwnd: Kiểm tra HWND có tồn tại qua IsWindow hay không Returns: ReconnectValidationResult chứa trạng thái và danh sách vi phạm. """ errors: List[str] = [] warnings: List[str] = [] meta: Dict[str, Any] = {} # 1. Schema Validation cơ bản required_fields = ["status", "pid", "hwnd"] for field_name in required_fields: if field_name not in result_data: errors.append(f"Missing required field: '{field_name}'") if errors: return ReconnectValidationResult(is_valid=False, errors=errors, metadata=meta) status = str(result_data.get("status", "")).upper() pid = result_data.get("pid") hwnd_raw = result_data.get("hwnd") # 2. INV-RECONNECT-PROCESS-ALIVE parsed_pid: Optional[int] = None try: parsed_pid = int(pid) meta["parsed_pid"] = parsed_pid if parsed_pid <= 0: errors.append(f"INV-RECONNECT-PROCESS-ALIVE: Invalid PID ({parsed_pid})") except (ValueError, TypeError): errors.append(f"INV-RECONNECT-PROCESS-ALIVE: Non-numeric PID ({pid})") if parsed_pid and check_live_process: if not psutil.pid_exists(parsed_pid): errors.append( f"INV-RECONNECT-PROCESS-ALIVE: PID {parsed_pid} does not exist in Windows process table. " f"Status '{status}' cannot be claimed for a non-existent process." ) else: try: proc = psutil.Process(parsed_pid) proc_name = proc.name() meta["process_name"] = proc_name if proc.status() == psutil.STATUS_ZOMBIE: errors.append(f"INV-RECONNECT-PROCESS-ALIVE: PID {parsed_pid} is a zombie process.") except (psutil.NoSuchProcess, psutil.AccessDenied) as e: errors.append(f"INV-RECONNECT-PROCESS-ALIVE: Cannot inspect PID {parsed_pid}: {e}") # 3. INV-RECONNECT-HWND-VALID parsed_hwnd: Optional[int] = None if isinstance(hwnd_raw, int): parsed_hwnd = hwnd_raw elif isinstance(hwnd_raw, str): try: parsed_hwnd = int(hwnd_raw, 16) if hwnd_raw.startswith(("0x", "0X")) else int(hwnd_raw) except ValueError: errors.append(f"INV-RECONNECT-HWND-VALID: Cannot parse HWND '{hwnd_raw}'") else: errors.append(f"INV-RECONNECT-HWND-VALID: HWND must be str or int, got {type(hwnd_raw).__name__}") meta["parsed_hwnd"] = parsed_hwnd if parsed_hwnd and check_live_hwnd and win32gui is not None: if parsed_hwnd <= 0 or not win32gui.IsWindow(parsed_hwnd): errors.append( f"INV-RECONNECT-HWND-VALID: HWND {hex(parsed_hwnd)} is not a valid open window." ) # 4. INV-RECONNECT-NO-HARDCODED-VITALS & Rule 4C/5 Vitals Integrity es_val = result_data.get("es", 0) hp_val = result_data.get("hp", 0) mana_val = result_data.get("mana", 0) try: vitals_tuple = (int(es_val), int(hp_val), int(mana_val)) except (ValueError, TypeError): vitals_tuple = (0, 0, 0) warnings.append("Vitals values are non-numeric.") meta["vitals"] = { "es": es_val, "hp": hp_val, "mana": mana_val } if ram_snapshot is not None: player_ram = ram_snapshot.get("player", {}) ram_addr = str(player_ram.get("address", "0x0")).lower() mem_attached = ram_snapshot.get("memory_attached", False) meta["ram_attached"] = mem_attached meta["player_ram_address"] = ram_addr # Nếu RAM chưa đọc được (địa chỉ 0x0) hoặc memory_attached = False is_ram_dead = (not mem_attached) or (ram_addr in ("0x0", "0", "")) if is_ram_dead: # Phát hiện gán cứng template CI if vitals_tuple in HARDCODED_PROHIBITED_VITALS: errors.append( f"INV-RECONNECT-NO-HARDCODED-VITALS: Detected hardcoded CI template vitals " f"(es={es_val}, hp={hp_val}, mana={mana_val}) while RAM snapshot has player.address={ram_addr} " f"and memory_attached={mem_attached}. Violates Rule 4C & Rule 5!" ) elif status == "SUCCESS" and (int(es_val) > 0 or int(hp_val) > 0): errors.append( f"INV-RECONNECT-NO-HARDCODED-VITALS: Reconnect result claims SUCCESS with active vitals " f"(es={es_val}, hp={hp_val}) but RAM snapshot shows unattached memory (address={ram_addr})." ) else: # Nếu RAM sống, đối chiếu tính nhất quán giữa RAM và Result ram_hp = player_ram.get("hp", 0) ram_es = player_ram.get("es", 0) ram_mana = player_ram.get("mana", 0) if int(hp_val) != int(ram_hp) or int(es_val) != int(ram_es): warnings.append( f"Vitals mismatch between Reconnect Result (hp={hp_val}, es={es_val}) " f"and RAM Snapshot (hp={ram_hp}, es={ram_es})." ) is_valid = (len(errors) == 0) return ReconnectValidationResult( is_valid=is_valid, errors=errors, warnings=warnings, metadata=meta, )