""" AutoPOE2 - Realtime Watcher Điều phối viên giám sát thời gian thực kết nối Invariant Sentinel, Multimodal Flight Recorder và bảo toàn sinh mệnh độc lập (Co-Pilot Passive Mode). """ import logging import time from typing import Any, List, Optional, Tuple from src.agentic_engine.flight_recorder import MultimodalFlightRecorder from src.agentic_engine.invariant_sentinel import ( ERR_OPTICAL_RAM_DESYNC, ERR_WATCHDOG_HEARTBEAT_LOST, ERR_XYZ_FROZEN, InvariantSentinel, ) from src.assistant_tool.core_shm_bridge import CoreTelemetrySnapshot logger = logging.getLogger("RealtimeWatcher") class RealtimeWatcher: """ Người giám sát thời gian thực tự hành (Autonomous Closed-Loop Watcher). """ def __init__( self, sentinel: Optional[InvariantSentinel] = None, flight_recorder: Optional[MultimodalFlightRecorder] = None, ): self.sentinel = sentinel or InvariantSentinel() self.flight_recorder = flight_recorder or MultimodalFlightRecorder() self.last_known_xyz: Tuple[float, float, float] = (0.0, 0.0, 0.0) self.is_moving: bool = False self.last_heartbeat_ms: int = int(time.time() * 1000) self.is_copilot_passive: bool = False self.request_stop: bool = False self.active_errors: List[str] = [] self._last_incident_time: Dict[str, float] = {} self.incident_debounce_s: float = 5.0 def _should_export_incident(self, err_code: str) -> bool: """Kiểm tra thời gian hồi debounce trước khi xuất incident mới (tránh flood I/O).""" now = time.time() last = self._last_incident_time.get(err_code, 0.0) if now - last < self.incident_debounce_s: return False self._last_incident_time[err_code] = now return True def set_moving(self, is_moving: bool) -> None: """Cập nhật cờ di chuyển của nhân vật.""" self.is_moving = is_moving def update_heartbeat(self, now_ms: Optional[int] = None) -> None: """Cập nhật mốc nhịp tim mới nhất từ agent/client.""" self.last_heartbeat_ms = now_ms if now_ms is not None else int(time.time() * 1000) if self.is_copilot_passive: logger.info("[Watcher] Nhịp tim phục hồi -> Khôi phục chế độ Tự Hành Đầy Đủ (Full Autonomous)") self.is_copilot_passive = False def check_watchdog(self, now_ms: Optional[int] = None, timeout_ms: int = 3000) -> bool: """ Kiểm tra trạng thái nhịp tim theo INV-WDG-01. Nếu timeout (>3000ms), chuyển sang Co-Pilot Passive, TUYỆT ĐỐI KHÔNG set request_stop. """ current_time = now_ms if now_ms is not None else int(time.time() * 1000) valid, err_code, desc = self.sentinel.evaluate_watchdog( self.last_heartbeat_ms, current_time, timeout_ms ) if not valid: self.is_copilot_passive = True # Bất biến INV-WDG-01: CẤM TỰ SÁT HOẶC YÊU CẦU DỪNG CORE self.request_stop = False if err_code and err_code not in self.active_errors: self.active_errors.append(err_code) return False self.is_copilot_passive = False if ERR_WATCHDOG_HEARTBEAT_LOST in self.active_errors: self.active_errors.remove(ERR_WATCHDOG_HEARTBEAT_LOST) return True def process_telemetry( self, snapshot: CoreTelemetrySnapshot, optical_vitals: Optional[Any] = None, is_moving: Optional[bool] = None, trigger_incident_export: bool = True, ) -> List[str]: """ Đối soát toàn diện 1 khung Telemetry Snapshot với Invariant Matrix: 1. Vitals Bound & Anti-778 2. XYZ Vector3 Tracking (hỗ trợ chuyển động mặt phẳng 2D Sandswept Marsh) 3. Optical vs RAM Coherence """ if is_moving is not None: self.is_moving = is_moving detected_errors: List[str] = [] # 1. Ghi nhận tick vào hộp đen self.flight_recorder.record_tick(snapshot, optical_vitals) # 2. Kiểm tra Vitals (INV-VIT-01 & INV-VIT-02) vitals_ok, vitals_err, vitals_desc = self.sentinel.evaluate_vitals(snapshot) if not vitals_ok and vitals_err: detected_errors.append(vitals_err) if vitals_err not in self.active_errors: self.active_errors.append(vitals_err) if trigger_incident_export and self._should_export_incident(vitals_err): self.flight_recorder.export_incident( error_code=vitals_err, description=vitals_desc, snapshot=snapshot, optical_vitals=optical_vitals, ) # 3. Kiểm tra XYZ Movement (INV-XYZ-01 & INV-XYZ-02) p = snapshot.player curr_xyz = (p.pos_x, p.pos_y, p.pos_z) # Bỏ qua nếu đây là frame khởi tạo đầu tiên (last_known_xyz chưa có giá trị) if self.last_known_xyz != (0.0, 0.0, 0.0) or self.is_moving: xyz_ok, xyz_err, xyz_desc = self.sentinel.verify_xyz_movement( prev_xyz=self.last_known_xyz, curr_xyz=curr_xyz, is_moving=self.is_moving, ) if not xyz_ok and xyz_err: detected_errors.append(xyz_err) if xyz_err not in self.active_errors: self.active_errors.append(xyz_err) if trigger_incident_export and self._should_export_incident(xyz_err): self.flight_recorder.export_incident( error_code=xyz_err, description=xyz_desc, snapshot=snapshot, optical_vitals=optical_vitals, ) else: # Cập nhật tọa độ mới nhất nếu hợp lệ if ERR_XYZ_FROZEN in self.active_errors: self.active_errors.remove(ERR_XYZ_FROZEN) self.last_known_xyz = curr_xyz # 4. Kiểm tra Optical vs RAM (INV-OPT-01) if optical_vitals is not None and getattr(optical_vitals, "current_hp", 0) > 0: opt_hp = optical_vitals.current_hp ram_hp = p.current_hp max_hp = max(p.max_hp, getattr(optical_vitals, "max_hp", 0)) opt_ok, opt_err, opt_desc = self.sentinel.evaluate_optical_vs_ram( optical_hp=opt_hp, ram_hp=ram_hp, max_hp=max_hp, optical_ratio=getattr(optical_vitals, "hp_ratio", None), ) if not opt_ok and opt_err: detected_errors.append(opt_err) if opt_err not in self.active_errors: self.active_errors.append(opt_err) if trigger_incident_export and self._should_export_incident(opt_err): self.flight_recorder.export_incident( error_code=opt_err, description=opt_desc, snapshot=snapshot, optical_vitals=optical_vitals, ) else: if ERR_OPTICAL_RAM_DESYNC in self.active_errors: self.active_errors.remove(ERR_OPTICAL_RAM_DESYNC) return detected_errors