""" Module Giám Sát Thời Gian Thực & Khép Kín Vòng Lặp Tự Sửa Lỗi (AgentWatcherSentinel) Kế thừa và tái sử dụng Single Source of Truth từ src.agentic_engine: - InvariantSentinel & 14 Architectural Invariant Error Codes - MultimodalFlightRecorder (MultimodalIncidentRecorder) 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). """ import json import logging import os import time from collections import deque from typing import Any, Deque, Dict, List, Optional, Tuple from src.agentic_engine.flight_recorder import ( MultimodalFlightRecorder, MultimodalIncidentRecorder, ) from src.agentic_engine.invariant_sentinel import ( ALL_INVARIANT_CODES, ERR_BOGUS_VITAL_STRUCT_778, ERR_CORE_DISCONNECTED, ERR_CORE_LOG_FATAL_EXCEPTION, ERR_CURRENT_ES_EXCEEDS_MAX, ERR_ES_OVERFLOW_0x7FFB, ERR_HP_EXCEEDS_MAX, ERR_MANA_BOGUS_STRUCT_779, ERR_MANA_EXCEEDS_MAX, ERR_OPTICAL_RAM_DESYNC, ERR_PLAYER_DEAD_OR_ZERO, ERR_WATCHDOG_HEARTBEAT_LOST, ERR_WATCHDOG_HEARTBEAT_TIMEOUT, ERR_XYZ_FROZEN, ERR_XYZ_FROZEN_AT_ORIGIN, ERR_XYZ_OUT_OF_BOUNDS, ERR_XYZ_TELEPORT_ANOMALY, INVARIANT_NAMES, InvariantSentinel, InvariantViolation, ) logger = logging.getLogger("AgentWatcher") # ============================================================================ # AGENT WATCHER SENTINEL (NGƯỜI GIÁM SÁT THỜI GIAN THỰC) # ============================================================================ class AgentWatcherSentinel: """ Module Giám Sát Thời Gian Thực & Điều Phối Tự Động (Master Sentinel). Chạy song song ở Cold Path để bảo vệ tính bất biến của hệ thống. Kết nối: - CoreShmBridge (đọc Telemetry 30-120Hz) - core_log.txt (quét fatal logs) - ScreenCapturer (chụp hình & RAM snapshot) - OpticalHPSensor (đối soát quang học 2ms) """ def __init__( self, core_log_path: Optional[str] = None, poll_interval: float = 0.5, flight_recorder_interval: float = 5.0, max_dev_snapshots: int = 20, shm_bridge: Optional[Any] = None, ): self.poll_interval = poll_interval self.flight_recorder_interval = flight_recorder_interval self.max_dev_snapshots = max_dev_snapshots self.running = False # Định vị tệp core_log.txt thật nằm cạnh binary Core root_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")) if core_log_path: self.core_log_path = core_log_path else: cands = [ os.path.join(root_dir, "bin", "Release", "core_log.txt"), os.path.join(root_dir, "bin", "core_log.txt"), os.path.join(root_dir, "build", "bin", "Release", "core_log.txt"), ] self.core_log_path = next((p for p in cands if os.path.exists(p)), cands[0]) self.incident_recorder = MultimodalIncidentRecorder() self.sentinel = InvariantSentinel() # Bộ đệm log xoay vòng self.core_log_buffer: Deque[str] = deque(maxlen=100) self.companion_log_buffer: Deque[str] = deque(maxlen=100) self.log_file_pos = 0 # Trạng thái di chuyển và theo dõi kẹt tọa độ self.last_pos = (0.0, 0.0, 0.0) self.stationary_start_time = 0.0 self.last_flight_record_time = 0.0 self.last_telemetry: Dict[str, Any] = {} self.last_heartbeat_ms = 0 # Danh sách snapshot Dev định kỳ để xoay vòng tối đa max_dev_snapshots (20) self.dev_snapshots: Deque[List[str]] = deque(maxlen=self.max_dev_snapshots * 2) # 1. Khởi tạo ScreenCapturer self.screen_capturer = None try: from src.assistant_tool.screen_capturer import ScreenCapturer self.screen_capturer = ScreenCapturer(output_dir="captures") except Exception as e: logger.warning(f"Chưa nạp ScreenCapturer: {e}") # 2. Khởi tạo OpticalHPSensor self.optical_sensor = None try: from src.assistant_tool.optical_hp_sensor import OpticalHPSensor self.optical_sensor = OpticalHPSensor() except Exception as e: logger.debug(f"Chưa nạp OpticalHPSensor: {e}") # 3. Kết nối CoreShmBridge (nếu được cung cấp) self.shm_bridge = shm_bridge if self.shm_bridge: try: self.shm_bridge.register_listener(self.on_shm_snapshot) except Exception as e: logger.warning(f"Lỗi đăng ký SHM listener: {e}") def connect_shm_bridge(self, bridge: Optional[Any] = None) -> bool: """Kết nối vào CoreShmBridge để nhận Telemetry thời gian thực.""" if bridge is not None: self.shm_bridge = bridge elif self.shm_bridge is None: try: from src.assistant_tool.core_shm_bridge import CoreShmBridge self.shm_bridge = CoreShmBridge() self.shm_bridge.start() except Exception as e: logger.warning(f"Không thể khởi tạo CoreShmBridge: {e}") return False if self.shm_bridge: try: self.shm_bridge.register_listener(self.on_shm_snapshot) return True except Exception as e: logger.warning(f"Lỗi gắn callback listener CoreShmBridge: {e}") return False def on_shm_snapshot(self, snapshot: Any) -> None: """Callback nhận Telemetry snapshot mới nhất từ Shared Memory.""" p = getattr(snapshot, "player", None) if not p: return now_ms = getattr(snapshot, "timestamp_ms", int(time.time() * 1000)) self.last_heartbeat_ms = now_ms telemetry: Dict[str, Any] = { "is_online": getattr(snapshot, "is_online", True), "snapshot_id": getattr(snapshot, "snapshot_id", 0), "timestamp_ms": now_ms, "area_name": getattr(snapshot, "area_name", ""), "area_level": getattr(snapshot, "area_level", 0), "entity_count": getattr(snapshot, "entity_count", 0), "monsters_count": getattr(snapshot, "monsters_count", 0), "items_count": getattr(snapshot, "items_count", 0), "player": { "current_hp": getattr(p, "current_hp", 0), "max_hp": getattr(p, "max_hp", 0), "current_mana": getattr(p, "current_mana", 0), "max_mana": getattr(p, "max_mana", 0), "current_es": getattr(p, "current_es", 0), "max_es": getattr(p, "max_es", 0), "current_spirit": getattr(p, "current_spirit", 0), "max_spirit": getattr(p, "max_spirit", 0), "hp_percent": getattr(p, "hp_percentage", 100.0), "pos_x": getattr(p, "pos_x", 0.0), "pos_y": getattr(p, "pos_y", 0.0), "pos_z": getattr(p, "pos_z", 0.0), }, } self.last_telemetry = telemetry opt_pct = self.sample_optical_hp() self.evaluate_live_telemetry(telemetry, optical_hp_pct=opt_pct) def sample_optical_hp(self) -> Optional[float]: """Đo lường lượng HP quang học từ màn hình game bằng OpticalHPSensor.""" if not self.optical_sensor: return None try: _, _, ratio = self.optical_sensor.read_hp() if ratio > 0.0: return round(ratio * 100.0, 1) except Exception: pass return None # Delegated Invariant check methods def check_vitals_invariants(self, player_data: Any) -> List[str]: return self.sentinel.check_vitals_invariants(player_data) def check_xyz_invariants( self, 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]: return self.sentinel.check_xyz_invariants( pos_x=pos_x, pos_y=pos_y, pos_z=pos_z, area_name=area_name, monsters_count=monsters_count, is_moving=is_moving, stationary_duration_sec=stationary_duration_sec, last_pos=last_pos, ) def check_optical_crosscheck( self, ram_hp_pct: float, optical_hp_pct: Optional[float], threshold_pct: float = 10.0, ) -> List[str]: return self.sentinel.check_optical_crosscheck( ram_hp_pct=ram_hp_pct, optical_hp_pct=optical_hp_pct, threshold_pct=threshold_pct, ) def check_watchdog_invariants( self, last_heartbeat_ms: int, core_is_online: bool = True, timeout_ms: int = 3000, ) -> List[str]: return self.sentinel.check_watchdog_invariants( last_heartbeat_ms=last_heartbeat_ms, core_is_online=core_is_online, timeout_ms=timeout_ms, ) def check_log_errors(self, log_lines: List[str]) -> List[str]: return self.sentinel.check_log_errors(log_lines) def ingest_core_log_line(self, line: str) -> None: """Ghi nhận 1 dòng log thời gian thực từ C++ Core.""" clean = line.strip() if not clean: return self.core_log_buffer.append(clean) anomalies = self.check_log_errors([clean]) if anomalies: self._handle_anomaly( incident_type=anomalies[0], violated_invariants=anomalies, telemetry={"log_trigger": clean}, root_cause_hint="Phát hiện thông báo lỗi nghiêm trọng hoặc ngoại lệ trong log C++ Core.", suggested_fix="Kiểm tra call stack của exception và vùng nhớ heap xung quanh.", ) def tail_core_log_file(self) -> None: """Đọc phần mới thêm vào của tệp core_log.txt thật.""" if not os.path.exists(self.core_log_path): return try: with open(self.core_log_path, "r", encoding="utf-8", errors="ignore") as f: f.seek(self.log_file_pos) new_lines = f.readlines() self.log_file_pos = f.tell() for line in new_lines: self.ingest_core_log_line(line) except Exception as e: logger.debug(f"tail_core_log_file: {e}") def evaluate_live_telemetry( self, telemetry: Dict[str, Any], optical_hp_pct: Optional[float] = None, ) -> List[str]: """ Đánh giá toàn bộ ma trận bất biến trên snapshot telemetry thật. Trả về danh sách các vi phạm phát hiện được. Kích hoạt Incident Dossier tức thời 0ms nếu có vi phạm. """ all_violations: List[str] = [] # 1. Thẩm tra Vitals p = telemetry.get("player", telemetry) vitals_errs = self.check_vitals_invariants(p) all_violations.extend(vitals_errs) # 2. Thẩm tra Tọa độ XYZ px = float(p.get("pos_x", 0.0) or 0.0) py = float(p.get("pos_y", 0.0) or 0.0) pz = float(p.get("pos_z", 0.0) or 0.0) area = str(telemetry.get("area_name", telemetry.get("area", ""))) monsters_cnt = int(telemetry.get("monsters_count", 0) or 0) is_moving = bool(telemetry.get("is_moving", False)) now = time.time() if abs(px - self.last_pos[0]) < 0.1 and abs(py - self.last_pos[1]) < 0.1: if self.stationary_start_time == 0.0: self.stationary_start_time = now stat_sec = now - self.stationary_start_time else: self.stationary_start_time = now stat_sec = 0.0 self.last_pos = (px, py, pz) xyz_errs = self.check_xyz_invariants( pos_x=px, pos_y=py, pos_z=pz, area_name=area, monsters_count=monsters_cnt, is_moving=is_moving, stationary_duration_sec=stat_sec, last_pos=self.last_pos if self.last_pos != (px, py, pz) else None, ) all_violations.extend(xyz_errs) # 3. Thẩm tra Optical vs RAM ram_hp = float(p.get("current_hp", p.get("hp", 0)) or 0) ram_max_hp = float(p.get("max_hp", 0) or 0) ram_hp_pct = (ram_hp / ram_max_hp * 100.0) if ram_max_hp > 0 else 100.0 opt_errs = self.check_optical_crosscheck(ram_hp_pct, optical_hp_pct, threshold_pct=10.0) all_violations.extend(opt_errs) # 4. Thẩm tra Watchdog is_online = bool(telemetry.get("is_online", True)) wdg_errs = self.check_watchdog_invariants(self.last_heartbeat_ms, core_is_online=is_online) all_violations.extend(wdg_errs) # Kích hoạt Incident Dossier tức thời 0ms nếu có bất kỳ vi phạm nào if all_violations: primary_err = all_violations[0] hint = f"Vi phạm bất biến: {', '.join(all_violations)}" self._handle_anomaly( incident_type=primary_err, violated_invariants=all_violations, telemetry=telemetry, root_cause_hint=hint, suggested_fix="Khắc phục triệt để tại Single Source of Logic theo Rule 8 và Rule 14.", ) return all_violations def record_dev_snapshot(self) -> Optional[str]: """ Chế độ Dev: Tự động chụp screenshot + RAM snapshot định kỳ (mặc định mỗi 5 giây). Xoay vòng giữ tối đa max_dev_snapshots (20) snapshot gần nhất để tiết kiệm dung lượng. """ now = time.time() if now - self.last_flight_record_time < self.flight_recorder_interval: return None self.last_flight_record_time = now if not self.screen_capturer: return None try: img_path = self.screen_capturer.capture_screenshot( event_name="DEV_PERIODIC", telemetry_data=self.last_telemetry, capture_ram=True, ) if not img_path: return None base_no_ext = os.path.splitext(img_path)[0] group_files = [ img_path, f"{base_no_ext}.json", f"{base_no_ext}_RAM.json", f"{base_no_ext}_CORRELATED.png", ] self.dev_snapshots.append(group_files) while len(self.dev_snapshots) > self.max_dev_snapshots: old_files = self.dev_snapshots.popleft() for f in old_files: try: if os.path.exists(f): os.remove(f) except Exception: pass return img_path except Exception as e: logger.warning(f"Lỗi chụp dev snapshot định kỳ: {e}") return None def _handle_anomaly( self, incident_type: str, violated_invariants: List[str], telemetry: Dict[str, Any], root_cause_hint: str = "", suggested_fix: str = "", ) -> str: """Kích hoạt đóng gói hồ sơ sự cố 0ms.""" raw_png = None corr_png = None ram_json = None if self.screen_capturer: try: raw_png = self.screen_capturer.capture_screenshot( event_name=f"ANOMALY_{incident_type[:15]}", telemetry_data=telemetry, capture_ram=True, ) if raw_png: base_no_ext = os.path.splitext(raw_png)[0] corr_png = f"{base_no_ext}_CORRELATED.png" ram_path = f"{base_no_ext}_RAM.json" if os.path.exists(ram_path): with open(ram_path, "r", encoding="utf-8") as f: ram_json = json.load(f) except Exception as e: logger.warning(f"Lỗi chụp ảnh anomaly: {e}") dossier_path = self.incident_recorder.package_incident( incident_type=incident_type, violated_invariants=violated_invariants, telemetry=telemetry, raw_screenshot_path=raw_png, correlated_image_path=corr_png, ram_snapshot=ram_json, core_log_lines=list(self.core_log_buffer), companion_log_lines=list(self.companion_log_buffer), root_cause_hint=root_cause_hint, suggested_fix=suggested_fix, ) return dossier_path @staticmethod def evaluate_real_capture_file(capture_json_path: str) -> Dict[str, Any]: """ Nạp một tệp capture JSON thực tế từ captures/ và chạy thẩm tra toàn bộ Invariants. Tuân thủ Rule 14: 100% sử dụng dữ liệu thực tế trích xuất từ game. """ if not os.path.exists(capture_json_path): raise FileNotFoundError(f"Không tìm thấy tệp capture thực tế: {capture_json_path}") with open(capture_json_path, "r", encoding="utf-8") as f: data = json.load(f) telemetry = data.get("telemetry", data) p = telemetry.get("player", telemetry) # 1. Đánh giá vitals vitals_violations = InvariantSentinel.check_vitals_invariants(p) # 2. Đánh giá XYZ px = float(p.get("pos_x", 0.0) or 0.0) py = float(p.get("pos_y", 0.0) or 0.0) pz = float(p.get("pos_z", 0.0) or 0.0) area = str(telemetry.get("area", telemetry.get("area_name", ""))) raw_mobs = data.get("monsters", []) mobs_cnt_default = len(raw_mobs) if isinstance(raw_mobs, list) else 0 monsters_cnt = int(telemetry.get("monsters_count", mobs_cnt_default) or 0) xyz_violations = InvariantSentinel.check_xyz_invariants( pos_x=px, pos_y=py, pos_z=pz, area_name=area, monsters_count=monsters_cnt, is_moving=False, stationary_duration_sec=4.0 if (abs(px) < 0.001 and abs(py) < 0.001) else 0.0, ) all_violations = vitals_violations + xyz_violations return { "file": capture_json_path, "timestamp": data.get("timestamp", data.get("timestamp_iso", "Unknown")), "pid": data.get("pid", 34188), "telemetry": telemetry, "is_authentic_data": True, "violations_detected": all_violations, "has_anomaly": len(all_violations) > 0, }