"""AutoPOE2 - Bộ Gác Cổng & Giám Sát Thời Gian Thực (Live Run Sentinel) Phân hệ: Lead QA & Governance Sentinel (Sư đoàn 4) Nhiệm vụ: - Chạy nền thụ động tuyệt đối (Zero-Input Guarantee: CẤM phát lệnh phím/chuột). - Giám sát đa kênh đồng thời: 1. Client.txt: Zone transitions, player deaths, fatal exceptions, portal errors. 2. Windows Focus: Trạng thái focus tiền cảnh của tiến trình game (PID 22632). 3. Live Invariants: HP/ES vitals, XYZ planar boundaries, Watchdog lag (>1500ms warning, >3000ms error). - Tự động đóng gói Incident Dossier chuẩn 6 thành phần vào debug_harness/incidents/ khi phát hiện bất thường. Tuân thủ nghiêm ngặt: - Rule 4: Tự động khóa chặn hồi quy. - Rule 5: Cấm mock ảo, 100% dữ liệu thực nghiệm. - Rule 8: Kiểm toán toàn diện Master Invariant Matrix (18 Invariants). """ from __future__ import annotations import logging import os import re import threading import time from dataclasses import dataclass, field from datetime import datetime from typing import Any, Callable, Dict, List, Optional, Tuple from src.agentic_engine.flight_recorder import MultimodalFlightRecorder from src.agentic_engine.invariant_sentinel import ( 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_OPTICAL_RAM_DESYNC, ERR_PLAYER_DEAD_OR_ZERO, ERR_WATCHDOG_HEARTBEAT_TIMEOUT, ERR_XYZ_FROZEN_AT_ORIGIN, ERR_XYZ_OUT_OF_BOUNDS, ERR_XYZ_TELEPORT_ANOMALY, InvariantSentinel, ) from src.assistant_tool.config import Config logger = logging.getLogger("LiveRunSentinel") # Mã lỗi gác cổng mới cho phiên thử nghiệm thực địa ERR_FOCUS_LOST_WHILE_AUTONOMOUS = "INV-F8-01_ERR_FOCUS_LOST_WHILE_AUTONOMOUS" ERR_WATCHDOG_HEARTBEAT_LAG = "INV-WDG-03_ERR_WATCHDOG_HEARTBEAT_LAG_1500MS" ERR_CLIENT_LOG_PORTAL_DISABLED = "INV-KEY-02_ERR_PORTALS_DISABLED_IN_HIDEOUT" @dataclass class LiveSentinelStatus: """Ảnh chụp trạng thái giám sát thời gian thực.""" is_healthy: bool = True target_pid: int = 0 is_game_focused: bool = True current_area: str = "" last_heartbeat_ms: int = 0 heartbeat_age_ms: int = 0 autonomous_active: bool = False incidents_recorded: int = 0 active_anomalies: List[str] = field(default_factory=list) recent_log_events: List[str] = field(default_factory=list) class LiveRunSentinel: """ Bộ gác cổng thụ động giám sát phiên chạy thực địa trên Client game POE2. Đảm bảo an toàn tài khoản và đóng gói chứng cứ sự cố ngay tức thì. """ def __init__( self, target_pid: Optional[int] = None, log_path: Optional[str] = None, incidents_dir: str = "debug_harness/incidents", heartbeat_warning_ms: int = 1500, heartbeat_timeout_ms: int = 3000, poll_interval_s: float = 0.05, ): self.target_pid = target_pid or 0 self.log_path = log_path or Config.find_poe2_client_log() self.heartbeat_warning_ms = heartbeat_warning_ms self.heartbeat_timeout_ms = heartbeat_timeout_ms self.poll_interval_s = poll_interval_s # Thành phần hộp đen ghi nhận sự cố self.flight_recorder = MultimodalFlightRecorder(output_dir=incidents_dir) # Trạng thái theo dõi nội tại self._running = False self._thread: Optional[threading.Thread] = None self._lock = threading.Lock() # Dữ liệu động self.autonomous_active = False self.last_heartbeat_ms = int(time.time() * 1000) self.has_received_heartbeat = False self._last_dossier_time: Dict[str, float] = {} self.current_area = "" self.last_telemetry: Dict[str, Any] = {} self.recent_log_lines: List[str] = [] self.active_anomalies: List[str] = [] self.incidents_recorded = 0 # Con trỏ đọc log Client.txt self._log_file_pos = 0 if self.log_path and os.path.isfile(self.log_path): try: self._log_file_pos = os.path.getsize(self.log_path) except Exception: self._log_file_pos = 0 # Callbacks mở rộng self.on_incident_callbacks: List[Callable[[str, List[str], Dict[str, Any]], None]] = [] # Mẫu Regex bắt lỗi Client.txt self._re_scene = re.compile(r"^(\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2}).+?\[SCENE\] Set Source \[(.+?)\]") self._re_portals_disabled = re.compile(r"Portals are disabled here", re.IGNORECASE) self._re_fatal = re.compile(r"(\[FATAL\]|\[EXCEPTION\]|0xC0000005|Access Violation|Unhandled Exception|Crash detected)", re.IGNORECASE) def register_incident_callback(self, callback: Callable[[str, List[str], Dict[str, Any]], None]) -> None: """Đăng ký callback nhận thông báo khi phát hiện sự cố.""" self.on_incident_callbacks.append(callback) def set_autonomous_mode(self, active: bool) -> None: """Cập nhật cờ tự hành F8 đang BẬT hay TẮT.""" with self._lock: self.autonomous_active = active def update_telemetry(self, telemetry: Dict[str, Any]) -> None: """Cập nhật telemetry mới nhất từ Shared Memory IPC.""" with self._lock: self.last_telemetry = telemetry self.has_received_heartbeat = True now_ms = int(time.time() * 1000) self.last_heartbeat_ms = telemetry.get("timestamp_ms", now_ms) if "area_name" in telemetry and telemetry["area_name"]: self.current_area = telemetry["area_name"] # Ghi tick vào Flight Recorder self.flight_recorder.record_tick(telemetry) def is_game_window_focused(self) -> bool: """Kiểm tra cửa sổ tiền cảnh của Windows có khớp tiến trình game không.""" try: import ctypes user32 = ctypes.windll.user32 fg_hwnd = user32.GetForegroundWindow() if not fg_hwnd: return False pid = ctypes.c_ulong() user32.GetWindowThreadProcessId(fg_hwnd, ctypes.byref(pid)) current_fg_pid = int(pid.value) if self.target_pid > 0: return current_fg_pid == self.target_pid # Fallback nếu chưa biết PID cụ thể from src.common.win32_window import find_poe2_window poe_hwnd = find_poe2_window() return fg_hwnd == poe_hwnd except Exception: return True def check_live_invariants(self, now_ms: Optional[int] = None) -> List[str]: """Kiểm toán toàn diện các bất biến sống tại thời điểm hiện tại.""" if now_ms is None: now_ms = int(time.time() * 1000) anomalies: List[str] = [] # 1. Kiểm tra Focus Interlock khi đang tự hành (INV-F8-MUTES-BRAIN) if self.autonomous_active and not self.is_game_window_focused(): anomalies.append(ERR_FOCUS_LOST_WHILE_AUTONOMOUS) # 2. Kiểm tra Nhịp tim Watchdog IPC (INV-WDG-01) - chỉ kiểm tra khi đã kết nối và nhận nhịp tim if self.has_received_heartbeat: heartbeat_age = now_ms - self.last_heartbeat_ms if heartbeat_age > self.heartbeat_timeout_ms: anomalies.append(ERR_WATCHDOG_HEARTBEAT_TIMEOUT) elif heartbeat_age > self.heartbeat_warning_ms: anomalies.append(ERR_WATCHDOG_HEARTBEAT_LAG) # 3. Kiểm tra Telemetry Vitals & XYZ (INV-VIT, INV-XYZ) if self.last_telemetry: # Kiểm tra Vitals vital_errs = InvariantSentinel.check_vitals_invariants(self.last_telemetry) anomalies.extend(vital_errs) # Kiểm tra XYZ Out of bounds / frozen pos_x = float(self.last_telemetry.get("pos_x", 0.0) or 0.0) pos_y = float(self.last_telemetry.get("pos_y", 0.0) or 0.0) pos_z = float(self.last_telemetry.get("pos_z", 0.0) or 0.0) xyz_errs = InvariantSentinel.check_xyz_invariants( pos_x=pos_x, pos_y=pos_y, pos_z=pos_z, area_name=self.current_area, monsters_count=int(self.last_telemetry.get("monsters_count", 0) or 0), ) anomalies.extend(xyz_errs) return anomalies def poll_client_log(self) -> List[str]: """Đọc và trích xuất các dòng mới từ Client.txt.""" if not self.log_path or not os.path.isfile(self.log_path): return [] new_lines: List[str] = [] try: current_size = os.path.getsize(self.log_path) if current_size < self._log_file_pos: # Log bị rotate hoặc truncate self._log_file_pos = 0 if current_size > self._log_file_pos: with open(self.log_path, "r", encoding="utf-8", errors="replace") as f: f.seek(self._log_file_pos) chunk = f.read(current_size - self._log_file_pos) self._log_file_pos = f.tell() for line in chunk.splitlines(): clean = line.strip() if clean: new_lines.append(clean) self.recent_log_lines.append(clean) if len(self.recent_log_lines) > 50: self.recent_log_lines.pop(0) except Exception as e: logger.debug(f"Lỗi đọc Client.txt: {e}") return new_lines def process_tick(self, now_ms: Optional[int] = None) -> List[str]: """Thực thi 1 chu kỳ giám sát thụ động (Zero Input).""" if now_ms is None: now_ms = int(time.time() * 1000) # 1. Đọc log mới new_log_lines = self.poll_client_log() # 2. Kiểm tra bất biến log log_anomalies: List[str] = [] for line in new_log_lines: # Phát hiện chuyển cảnh m_scene = self._re_scene.search(line) if m_scene: zone = m_scene.group(2).strip() if zone.lower() not in {"act 1", "act 2", "act 3", "atlas", "world map", "(null)"}: self.current_area = zone # Phát hiện lỗi portal trong Hideout (phá bất biến INV-KEY-PORTAL) if self._re_portals_disabled.search(line): log_anomalies.append(ERR_CLIENT_LOG_PORTAL_DISABLED) # Phát hiện crash / fatal if self._re_fatal.search(line): log_anomalies.append(f"{ERR_CORE_LOG_FATAL_EXCEPTION}:{line}") # 3. Kiểm toán bất biến sống live_anomalies = self.check_live_invariants(now_ms=now_ms) total_anomalies = list(set(log_anomalies + live_anomalies)) self.active_anomalies = total_anomalies # 4. Tự động đóng gói Incident Dossier nếu có vi phạm nghiêm trọng critical_anomalies = [ a for a in total_anomalies if a != ERR_WATCHDOG_HEARTBEAT_LAG # Lag cảnh báo không đóng gói dossier ] if critical_anomalies: self._trigger_incident_packaging(critical_anomalies, now_ms=now_ms) return total_anomalies def _trigger_incident_packaging(self, anomalies: List[str], now_ms: int) -> str: """Đóng gói hồ sơ sự cố (Incident Dossier).""" first_err = anomalies[0] now_sec = time.time() if now_sec - self._last_dossier_time.get(first_err, 0.0) < 10.0: return "" self._last_dossier_time[first_err] = now_sec logger.error(f"🚨 [LiveRunSentinel] PHÁT HIỆN SỰ CỐ: {first_err} | Vi phạm: {anomalies}") telemetry_payload = dict(self.last_telemetry) telemetry_payload.setdefault("area_name", self.current_area) telemetry_payload.setdefault("timestamp_ms", now_ms) telemetry_payload["is_game_focused"] = self.is_game_window_focused() telemetry_payload["autonomous_active"] = self.autonomous_active dossier_path = self.flight_recorder.package_incident( incident_type=first_err, violated_invariants=anomalies, telemetry=telemetry_payload, core_log_lines=list(self.recent_log_lines[-20:]), companion_log_lines=[f"[LiveRunSentinel] Detected {first_err} at {now_ms}"], root_cause_hint="Live Run Sentinel phát hiện bất thường trong phiên chạy thực nghiệm.", suggested_fix="Kiểm tra trạng thái Client và tọa độ nhân vật tại thời điểm sự cố.", ) self.incidents_recorded += 1 for cb in self.on_incident_callbacks: try: cb(first_err, anomalies, telemetry_payload) except Exception as e: logger.warning(f"Lỗi gọi incident callback: {e}") return dossier_path def get_status(self) -> LiveSentinelStatus: """Lấy toàn bộ thông số báo cáo trạng thái giám sát.""" now_ms = int(time.time() * 1000) return LiveSentinelStatus( is_healthy=len(self.active_anomalies) == 0, target_pid=self.target_pid, is_game_focused=self.is_game_window_focused(), current_area=self.current_area, last_heartbeat_ms=self.last_heartbeat_ms, heartbeat_age_ms=now_ms - self.last_heartbeat_ms, autonomous_active=self.autonomous_active, incidents_recorded=self.incidents_recorded, active_anomalies=list(self.active_anomalies), recent_log_events=list(self.recent_log_lines[-5:]), ) def start_monitoring(self) -> None: """Khởi động luồng giám sát thụ động nền.""" with self._lock: if self._running: return self._running = True self._thread = threading.Thread(target=self._run_loop, daemon=True) self._thread.start() logger.info(f"✅ [LiveRunSentinel] Đã khởi chạy luồng giám sát thụ động PID={self.target_pid}") def stop_monitoring(self) -> None: """Dừng luồng giám sát an toàn.""" self._running = False if self._thread and self._thread.is_alive(): self._thread.join(timeout=1.5) logger.info("🛑 [LiveRunSentinel] Đã dừng luồng giám sát an toàn.") # Aliases for coordinator compatibility start = start_monitoring stop = stop_monitoring def _run_loop(self) -> None: """Vòng lặp giám sát liên tục.""" while self._running: try: self.process_tick() except Exception as e: logger.error(f"Lỗi trong vòng lặp giám sát: {e}") time.sleep(self.poll_interval_s)