""" AutoPOE2 - Architectural Invariant Sentinel (Single Source of Truth) Tuân thủ nghiêm ngặt: - Rule 8: Triệt tiêu sửa lỗi triệu chứng (Root-Cause-First). - Rule 11: Phòng ngừa hồi quy (Automated Regression Prevention). - Rule 12: Rà soát toàn cục & Tự đối kháng (Global Invariant Sweep). - Rule 13: Giao thức Watcher thời gian thực & Tự sửa lỗi khép kín (6-Stage Closed-Loop). - Rule 14: Tuyệt đối cấm dùng mock data / bịa data — 100% dữ liệu thật (Real Authentic Data Protocol). Chứa toàn bộ 14 mã lỗi Invariant chuẩn mực: - INV-VIT-01..07 (Vitals sinh mệnh) - INV-XYZ-01..03 (Tọa độ không gian 3D & 2D Planar) - INV-OPT-01 (Đối soát quang học Optical vs RAM) - INV-WDG-01..02 (Watchdog nhịp tim & Core survival) - INV-LOG-01 (Quét ngoại lệ nghiêm trọng trong Core Log) """ import math import time from dataclasses import dataclass from typing import Any, Dict, List, Optional, Tuple # ============================================================================ # 1. 14 MÃ LỖI BẤT BIẾN KIẾN TRÚC CHUẨN MỰC (ARCHITECTURAL INVARIANT CODES) # ============================================================================ # Vitals Invariants (INV-VIT-01..07) ERR_BOGUS_VITAL_STRUCT_778 = "INV-VIT-01_ERR_BOGUS_VITAL_STRUCT_778" ERR_ES_OVERFLOW_0x7FFB = "INV-VIT-02_ERR_ES_OVERFLOW_0x7FFB" ERR_HP_EXCEEDS_MAX = "INV-VIT-03_ERR_HP_EXCEEDS_MAX" ERR_MANA_EXCEEDS_MAX = "INV-VIT-04_ERR_MANA_EXCEEDS_MAX" ERR_MANA_BOGUS_STRUCT_779 = "INV-VIT-05_ERR_MANA_BOGUS_STRUCT_779" ERR_CURRENT_ES_EXCEEDS_MAX = "INV-VIT-06_ERR_CURRENT_ES_EXCEEDS_MAX" ERR_PLAYER_DEAD_OR_ZERO = "INV-VIT-07_ERR_PLAYER_DEAD_OR_ZERO" # Spatial XYZ Invariants (INV-XYZ-01..03) ERR_XYZ_FROZEN_AT_ORIGIN = "INV-XYZ-01_ERR_XYZ_FROZEN_AT_ORIGIN" ERR_XYZ_OUT_OF_BOUNDS = "INV-XYZ-02_ERR_XYZ_OUT_OF_BOUNDS" ERR_XYZ_TELEPORT_ANOMALY = "INV-XYZ-03_ERR_XYZ_TELEPORT_ANOMALY" # Optical Invariants (INV-OPT-01) ERR_OPTICAL_RAM_DESYNC = "INV-OPT-01_ERR_OPTICAL_RAM_DESYNC" # Watchdog & Core Survival Invariants (INV-WDG-01..02) ERR_WATCHDOG_HEARTBEAT_TIMEOUT = "INV-WDG-01_ERR_WATCHDOG_HEARTBEAT_TIMEOUT" ERR_CORE_DISCONNECTED = "INV-WDG-02_ERR_CORE_DISCONNECTED" # Core Log & Fatal Crash Invariants (INV-LOG-01) ERR_CORE_LOG_FATAL_EXCEPTION = "INV-LOG-01_ERR_CORE_LOG_FATAL_EXCEPTION" # Backward compatibility aliases ERR_XYZ_FROZEN = ERR_XYZ_FROZEN_AT_ORIGIN ERR_WATCHDOG_HEARTBEAT_LOST = ERR_WATCHDOG_HEARTBEAT_TIMEOUT ALL_INVARIANT_CODES: List[str] = [ ERR_BOGUS_VITAL_STRUCT_778, ERR_ES_OVERFLOW_0x7FFB, ERR_HP_EXCEEDS_MAX, ERR_MANA_EXCEEDS_MAX, ERR_MANA_BOGUS_STRUCT_779, ERR_CURRENT_ES_EXCEEDS_MAX, ERR_PLAYER_DEAD_OR_ZERO, ERR_XYZ_FROZEN_AT_ORIGIN, ERR_XYZ_OUT_OF_BOUNDS, ERR_XYZ_TELEPORT_ANOMALY, ERR_OPTICAL_RAM_DESYNC, ERR_WATCHDOG_HEARTBEAT_TIMEOUT, ERR_CORE_DISCONNECTED, ERR_CORE_LOG_FATAL_EXCEPTION, ] INVARIANT_NAMES: Dict[str, str] = { ERR_BOGUS_VITAL_STRUCT_778: "INV-VIT-01: Bogus Vital Struct 778/778 (UI Entity Marker)", ERR_ES_OVERFLOW_0x7FFB: "INV-VIT-02: Energy Shield Overflow 32763 (0x7FFB Vertex Trash)", ERR_HP_EXCEEDS_MAX: "INV-VIT-03: Current HP Exceeds Max HP", ERR_MANA_EXCEEDS_MAX: "INV-VIT-04: Current Mana Exceeds Max Mana", ERR_MANA_BOGUS_STRUCT_779: "INV-VIT-05: Mana Bogus Struct 779/779 (UI Entity Marker)", ERR_CURRENT_ES_EXCEEDS_MAX: "INV-VIT-06: Current ES Exceeds Max ES", ERR_PLAYER_DEAD_OR_ZERO: "INV-VIT-07: Player Dead Or Zero HP", ERR_XYZ_FROZEN_AT_ORIGIN: "INV-XYZ-01: Anti-Frozen (0,0,0) On Movement Or In Combat", ERR_XYZ_OUT_OF_BOUNDS: "INV-XYZ-02: XYZ Out Of Bounds / NaN / Inf", ERR_XYZ_TELEPORT_ANOMALY: "INV-XYZ-03: Abnormal Teleport Distance Jump", ERR_OPTICAL_RAM_DESYNC: "INV-OPT-01: Optical Screen HP vs RAM HP Desynchronization", ERR_WATCHDOG_HEARTBEAT_TIMEOUT: "INV-WDG-01: Watchdog Heartbeat Timeout (>3000ms)", ERR_CORE_DISCONNECTED: "INV-WDG-02: Core Process Disconnected / Offline", ERR_CORE_LOG_FATAL_EXCEPTION: "INV-LOG-01: Fatal Crash / Exception In Core Log", } @dataclass class InvariantViolation: """Mô tả một vi phạm bất biến kiến trúc tại thời điểm 0ms.""" invariant_code: str invariant_name: str severity: str message: str timestamp_ms: int # ============================================================================ # 2. BỘ ĐÁNH GIÁ BẤT BIẾN KIẾN TRÚC (ARCHITECTURAL INVARIANT SENTINEL) # ============================================================================ class InvariantSentinel: """ Bộ thẩm tra bất biến kiến trúc thời gian thực (Rule 12, Rule 13 & Rule 14). Hoạt động thuần túy trên dữ liệu thực địa, không dùng số liệu bịa đặt. Là Single Source of Truth duy nhất cho toàn bộ hệ thống AutoPOE2. """ def __init__( self, min_planar_delta: float = 0.8, optical_divergence_threshold: float = 0.25, watchdog_timeout_ms: int = 3000, ): self.min_planar_delta = min_planar_delta self.optical_divergence_threshold = optical_divergence_threshold self.watchdog_timeout_ms = watchdog_timeout_ms @staticmethod def _parse_vital_pair(raw_cur: Any, raw_max: Any) -> Tuple[int, int]: """Trích xuất cặp giá trị (current, max) an toàn từ cả số nguyên lẫn chuỗi định dạng (ví dụ 'HP: 157 / 157').""" cur = 0 max_val = 0 if isinstance(raw_cur, str) and "/" in raw_cur: parts = raw_cur.split("/") cur_digits = "".join(c for c in parts[0] if c.isdigit()) max_digits = "".join(c for c in parts[1].split()[0] if c.isdigit()) cur = int(cur_digits) if cur_digits else 0 max_val = int(max_digits) if max_digits else 0 else: if isinstance(raw_cur, (int, float)): cur = int(raw_cur) elif isinstance(raw_cur, str): digits = "".join(c for c in raw_cur if c.isdigit()) cur = int(digits) if digits else 0 if max_val == 0: if isinstance(raw_max, (int, float)): max_val = int(raw_max) elif isinstance(raw_max, str): digits = "".join(c for c in raw_max if c.isdigit()) max_val = int(digits) if digits else 0 return cur, max_val @staticmethod def check_vitals_invariants(player_data: Any) -> List[str]: """ Kiểm tra bất biến sinh tồn Vitals (INV-VIT-01..07): - Bắt lỗi HP == 778 && max_hp == 778 (bogus entity struct / UI marker). - Bắt lỗi ES > 20000 hoặc ES == 32763 (0x7FFB graphic vertex trash). - Bắt lỗi current_hp > max_hp. - Bắt lỗi current_mana > max_mana. - Bắt lỗi Mana == 779 && max_mana == 779 (UI marker). - Bắt lỗi current_es > max_es (khi max_es > 0). - Bắt lỗi HP <= 0 khi max_hp > 0 (player dead or zero). """ anomalies: List[str] = [] if isinstance(player_data, dict): p = player_data.get("player", player_data) elif hasattr(player_data, "player"): p = getattr(player_data, "player") else: p = player_data if isinstance(p, dict): cur_hp, max_hp = InvariantSentinel._parse_vital_pair( p.get("current_hp", p.get("hp", 0)), p.get("max_hp", 0) ) cur_es, max_es = InvariantSentinel._parse_vital_pair( p.get("current_es", p.get("es", 0)), p.get("max_es", 0) ) cur_mana, max_mana = InvariantSentinel._parse_vital_pair( p.get("current_mana", p.get("mana", 0)), p.get("max_mana", 0) ) else: cur_hp, max_hp = InvariantSentinel._parse_vital_pair( getattr(p, "current_hp", getattr(p, "hp", 0)), getattr(p, "max_hp", 0) ) cur_es, max_es = InvariantSentinel._parse_vital_pair( getattr(p, "current_es", getattr(p, "es", 0)), getattr(p, "max_es", 0) ) cur_mana, max_mana = InvariantSentinel._parse_vital_pair( getattr(p, "current_mana", getattr(p, "mana", 0)), getattr(p, "max_mana", 0) ) # 1. Bắt lỗi struct quái/marker UI 778/778 (INV-VIT-01) if cur_hp == 778 and max_hp == 778: anomalies.append(ERR_BOGUS_VITAL_STRUCT_778) # 2. Bắt lỗi marker UI Mana 779/779 (INV-VIT-05) if cur_mana == 779 and max_mana == 779: anomalies.append(ERR_MANA_BOGUS_STRUCT_779) # 3. Bắt lỗi đọc rác Energy Shield (0x7FFB = 32763 hoặc ES vọt trên trần 20000) (INV-VIT-02) if cur_es > 20000 or cur_es == 32763: anomalies.append(ERR_ES_OVERFLOW_0x7FFB) # 4. Bắt lỗi current_es vượt max_es (INV-VIT-06) if max_es > 0 and cur_es > max_es: if ERR_ES_OVERFLOW_0x7FFB not in anomalies: anomalies.append(ERR_CURRENT_ES_EXCEEDS_MAX) # 5. Bắt lỗi current_hp vượt max_hp (INV-VIT-03) if max_hp > 0 and cur_hp > max_hp: anomalies.append(ERR_HP_EXCEEDS_MAX) # 6. Bắt lỗi current_mana vượt max_mana (INV-VIT-04) if max_mana > 0 and cur_mana > max_mana: anomalies.append(ERR_MANA_EXCEEDS_MAX) # 7. Bắt lỗi nhân vật tử vong hoặc HP về 0 bất thường khi max_hp > 0 (INV-VIT-07) if max_hp > 0 and cur_hp <= 0: anomalies.append(ERR_PLAYER_DEAD_OR_ZERO) return anomalies @staticmethod def check_xyz_invariants( pos_x: float, pos_y: float, pos_z: float, area_name: str = "", monsters_count: int = 0, is_moving: bool = False, stationary_duration_sec: float = 0.0, last_pos: Optional[Tuple[float, float, float]] = None, ) -> List[str]: """ Kiểm tra bất biến tọa độ XYZ (INV-XYZ-01..03): - Bắt lỗi XYZ đứng yên (0.0, 0.0, 0.0) khi đang ở trong zone có quái vật hoặc di chuyển > 3s. - Kiểm tra NaN / Inf. - Kiểm tra biên bản đồ POE2 (abs > 50000). - Kiểm tra vận tốc dịch chuyển phi lý (> 2000 units/tick). """ anomalies: List[str] = [] # 1. Kiểm tra NaN / Inf (INV-XYZ-02) if not (math.isfinite(pos_x) and math.isfinite(pos_y) and math.isfinite(pos_z)): anomalies.append(ERR_XYZ_OUT_OF_BOUNDS) return anomalies # 2. Kiểm tra biên bản đồ POE2 (INV-XYZ-02) if abs(pos_x) > 50000.0 or abs(pos_y) > 50000.0 or abs(pos_z) > 10000.0: anomalies.append(ERR_XYZ_OUT_OF_BOUNDS) # 3. Kiểm tra kẹt tại gốc tọa độ (0.0, 0.0, 0.0) (INV-XYZ-01) is_at_origin = abs(pos_x) < 0.001 and abs(pos_y) < 0.001 and abs(pos_z) < 0.001 is_active_combat_area = bool(area_name and area_name != "--" and "town" not in area_name.lower()) if is_at_origin: if monsters_count > 0: anomalies.append(ERR_XYZ_FROZEN_AT_ORIGIN) elif (is_moving or is_active_combat_area) and stationary_duration_sec >= 3.0: anomalies.append(ERR_XYZ_FROZEN_AT_ORIGIN) # 4. Kiểm tra nhảy bước dịch chuyển phi lý (Teleportation anomaly) (INV-XYZ-03) if last_pos is not None: lx, ly, _ = last_pos dist = math.sqrt((pos_x - lx) ** 2 + (pos_y - ly) ** 2) if dist > 2000.0 and not is_at_origin: anomalies.append(ERR_XYZ_TELEPORT_ANOMALY) return anomalies @staticmethod def check_optical_crosscheck( ram_hp_pct: float, optical_hp_pct: Optional[float], threshold_pct: float = 25.0, ) -> List[str]: """Bắt lỗi chênh lệch giữa Screen HP (từ Optical sensor) và RAM HP (INV-OPT-01). Tự động chuẩn hóa cả thang tỷ lệ [0.0 - 1.0] lẫn phần trăm [0 - 100%]. """ if optical_hp_pct is None: return [] # Tự động quy về thang phần trăm nếu giá trị vào là ratio <= 1.0 và threshold > 1.0 r_val = ram_hp_pct * 100.0 if (ram_hp_pct <= 1.0 and threshold_pct > 1.0) else ram_hp_pct o_val = optical_hp_pct * 100.0 if (optical_hp_pct <= 1.0 and threshold_pct > 1.0) else optical_hp_pct if abs(r_val - o_val) > threshold_pct: return [ERR_OPTICAL_RAM_DESYNC] return [] @staticmethod def check_watchdog_invariants( last_heartbeat_ms: int, core_is_online: bool = True, timeout_ms: int = 3000, now_ms: Optional[int] = None, ) -> List[str]: """Bắt Core bị timeout hoặc ngắt kết nối bất ngờ (INV-WDG-01..02).""" anomalies: List[str] = [] if not core_is_online: anomalies.append(ERR_CORE_DISCONNECTED) if now_ms is None: now_ms = int(time.time() * 1000) if last_heartbeat_ms > 0 and (now_ms - last_heartbeat_ms > timeout_ms): anomalies.append(ERR_WATCHDOG_HEARTBEAT_TIMEOUT) return anomalies @staticmethod def check_log_errors(log_lines: List[str]) -> List[str]: """Bắt các dòng [FATAL], [EXCEPTION], Access Violation, c0000005, Unhandled Exception, v.v. (INV-LOG-01).""" anomalies: List[str] = [] fatal_keywords = [ "[FATAL]", "[EXCEPTION]", "Access Violation", "0xC0000005", "c0000005", "Unhandled Exception", "std::bad_alloc", "CRASH_DUMP", "Assertion failed", "Crash detected", ] for line in log_lines: for kw in fatal_keywords: if kw in line: anomalies.append(f"{ERR_CORE_LOG_FATAL_EXCEPTION}:{kw}") break return anomalies @staticmethod def check_all(data: Dict[str, Any]) -> List[str]: """Kiểm tra toàn bộ bất biến từ dictionary telemetry.""" anomalies = [] anomalies.extend(InvariantSentinel.check_vitals_invariants(data)) p = data.get("player", data) if isinstance(data, dict) else getattr(data, "player", data) if isinstance(p, dict): px = float(p.get("pos_x", data.get("pos_x", 0.0)) or 0.0) py = float(p.get("pos_y", data.get("pos_y", 0.0)) or 0.0) pz = float(p.get("pos_z", data.get("pos_z", 0.0)) or 0.0) else: px = float(getattr(p, "pos_x", getattr(data, "pos_x", 0.0)) or 0.0) py = float(getattr(p, "pos_y", getattr(data, "pos_y", 0.0)) or 0.0) pz = float(getattr(p, "pos_z", getattr(data, "pos_z", 0.0)) or 0.0) anomalies.extend(InvariantSentinel.check_xyz_invariants( pos_x=px, pos_y=py, pos_z=pz, area_name=str(data.get("area_name", data.get("area", ""))), monsters_count=int(data.get("monsters_count", 0) or 0), )) return anomalies # ======================================================================== # INSTANCE METHODS CHO REALTIME WATCHER & HARNESS SCENARIOS # ======================================================================== def evaluate_vitals(self, snapshot_or_data: Any) -> Tuple[bool, Optional[str], str]: """ Kiểm tra tính toàn vẹn của Vitals theo INV-VIT-01..07. Trả về (is_valid, error_code_if_any, description). """ anomalies = self.check_vitals_invariants(snapshot_or_data) if anomalies: first_err = anomalies[0] name = INVARIANT_NAMES.get(first_err, first_err) return ( False, first_err, f"Phát hiện vi phạm bất biến sinh tồn ({first_err}) - {name}: {', '.join(anomalies)}", ) return (True, None, "Vitals thỏa mãn toàn bộ Architectural Invariants") def verify_xyz_movement( self, prev_xyz: Tuple[float, float, float], curr_xyz: Tuple[float, float, float], is_moving: bool = True, ) -> Tuple[bool, Optional[str], str]: """ Kiểm chứng tính hợp lệ của Vector3 Position theo INV-XYZ-01..03. Hỗ trợ nhận diện chuyển động mặt phẳng 2D (Z=const, delta_x/delta_y > 0) trên đầm lầy Sandswept Marsh hoặc sàn phẳng. """ x, y, z = curr_xyz px, py, pz = prev_xyz # 1. Bắt lỗi kẹt cứng tại gốc tọa độ (0, 0, 0) khi đang di chuyển (INV-XYZ-01) if is_moving and abs(x) < 0.001 and abs(y) < 0.001 and abs(z) < 0.001: return ( False, ERR_XYZ_FROZEN_AT_ORIGIN, "Nhân vật đang di chuyển nhưng tọa độ kẹt tại (0.0, 0.0, 0.0) - Vi phạm INV-XYZ-01", ) if not is_moving: return (True, None, "Nhân vật đứng yên hợp lệ") dx = x - px dy = y - py dz = z - pz planar_dist = math.sqrt(dx * dx + dy * dy) dist_3d = math.sqrt(dx * dx + dy * dy + dz * dz) # 2. Nhận diện chuyển động mặt phẳng 2D (Sandswept Marsh Swamp: Z giữ nguyên, X và Y thay đổi) if planar_dist >= self.min_planar_delta: return ( True, None, f"Chuyển động mặt phẳng 2D hợp lệ (planar_delta={planar_dist:.2f}, dz={dz:.2f})", ) # 3. Nhận diện dốc (Y-Z hoặc X-Z) if dist_3d >= self.min_planar_delta: return ( True, None, f"Chuyển động địa hình dốc hợp lệ (dist_3d={dist_3d:.2f})", ) # 4. Khi is_moving = True nhưng tọa độ hoàn toàn không suy suyển quá ngưỡng return ( False, ERR_XYZ_FROZEN_AT_ORIGIN, f"Nhân vật di chuyển nhưng XYZ đứng im (planar_delta={planar_dist:.2f} < {self.min_planar_delta}) - Vi phạm INV-XYZ-01", ) def evaluate_optical_vs_ram( self, optical_hp: float, ram_hp: float, max_hp: float = 0.0, optical_ratio: Optional[float] = None, ram_ratio: Optional[float] = None, ) -> Tuple[bool, Optional[str], str]: """ Kiểm tra độ đồng nhất giữa cảm biến quang học Quả Cầu Máu và RAM theo INV-OPT-01. Phát hiện stale pointer RAM khi chênh lệch vượt quá ngưỡng cho phép (mặc định 25%). """ if max_hp <= 0.0: max_hp = max(float(optical_hp), float(ram_hp), 1.0) opt_r = optical_ratio if optical_ratio is not None else (optical_hp / max_hp) ram_r = ram_ratio if ram_ratio is not None else (ram_hp / max_hp) divergence = abs(opt_r - ram_r) if divergence > self.optical_divergence_threshold: return ( False, ERR_OPTICAL_RAM_DESYNC, f"Lệch pha Optical ({optical_hp} HP, ratio={opt_r:.2f}) vs RAM ({ram_hp} HP, ratio={ram_r:.2f}): " f"độ lệch {divergence * 100:.1f}% > ngưỡng {self.optical_divergence_threshold * 100:.1f}% - Vi phạm INV-OPT-01", ) return (True, None, f"Optical và RAM đồng bộ (độ lệch {divergence * 100:.1f}% <= {self.optical_divergence_threshold * 100:.1f}%)") def evaluate_watchdog( self, heartbeat_timestamp_ms: int, now_ms: int, timeout_ms: Optional[int] = None, ) -> Tuple[bool, Optional[str], str]: """ Kiểm tra nhịp tim Watchdog theo INV-WDG-01. Khi mất nhịp tim (>3000ms), kích hoạt Co-Pilot Passive Mode, cấm hủy tiến trình. """ effective_timeout = timeout_ms if timeout_ms is not None else self.watchdog_timeout_ms age_ms = now_ms - heartbeat_timestamp_ms if age_ms > effective_timeout: return ( False, ERR_WATCHDOG_HEARTBEAT_TIMEOUT, f"Mất nhịp tim Watchdog ({age_ms}ms > {effective_timeout}ms) -> Hạ cấp Co-Pilot Passive, bảo vệ Core độc lập", ) return (True, None, f"Watchdog nhịp tim khỏe mạnh (age={age_ms}ms)") # Backward compatibility aliases evaluate_xyz = check_xyz_invariants evaluate_log_slice = check_log_errors