"""AutoPOE2 - Supervised Real Test Runner & Autonomous Recovery Harness. ======================================================================== Vận hành phiên kiểm thử thực tế (Real Test) trên client game live (2560x1440): 1. Kết hợp đa nguồn dữ liệu thật 100%: Game Window, Client.txt, Sentinel 3s, Optical Scan, IPC Telemetry. 2. Giám sát tự động khép kín từng Stage (Stage 1 -> Stage 6). 3. Cơ chế Khẩn cấp F8 Interlock (INV-F8-MUTES-BRAIN): Tự động phát hiện lỗi/timeout -> ngắt bot, nhả toàn bộ WASD. 4. Đóng gói Incident Dossier tự động khi gặp sự cố theo Rule 6 & SSoT Doc 61. 5. Tạo báo cáo nghiệm thu thực nghiệm đầy đủ (Empirical DoD). """ from __future__ import annotations import argparse import ctypes import json import os import sys import time import traceback from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) try: if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") except Exception: pass from src.assistant_tool.screen_capturer import ScreenCapturer from src.assistant_tool.log_watcher import LogWatcher from src.assistant_tool.core_shm_bridge import CoreShmBridge from src.assistant_tool.telemetry_correlator import MultimodalTelemetryCorrelator import src.common as common from src.common.win32_window import find_poe2_window, ensure_window_focus, get_client_rect from src.common.win32_input import emergency_f8_pause_and_release_all, dismiss_modals from tools.autonomous_mapping_coordinator import AutonomousMappingCoordinator class SupervisedRealTestRunner: def __init__( self, max_maps: int = 1, map_duration_sec: int = 25, start_stage: int = 0, stage_stop: int = 6, dry_run: bool = False, ): self.max_maps = max_maps self.map_duration_sec = map_duration_sec self.start_stage = start_stage self.stage_stop = stage_stop self.dry_run = dry_run self.timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") self.session_dir = os.path.join("debug_harness", "real_tests", f"REAL_TEST_{self.timestamp}") self.sentinel_dir = os.path.join(self.session_dir, "sentinel") self.stages_dir = os.path.join(self.session_dir, "stages") self.logs_dir = os.path.join(self.session_dir, "logs") os.makedirs(self.sentinel_dir, exist_ok=True) os.makedirs(self.stages_dir, exist_ok=True) os.makedirs(self.logs_dir, exist_ok=True) self.client_log_path = r"C:\Program Files (x86)\Grinding Gear Games\Path of Exile 2\logs\Client.txt" self.capturer = ScreenCapturer(output_dir=self.stages_dir) self.shm_bridge = CoreShmBridge() self.log_watcher = LogWatcher(self.client_log_path) if os.path.exists(self.client_log_path) else None self.log_file = os.path.join(self.logs_dir, "real_test_execution.log") self.execution_events: List[Dict[str, Any]] = [] def log(self, msg: str, level: str = "INFO"): ts = datetime.now().strftime("%H:%M:%S.%f")[:-3] formatted = f"[{ts}][{level}][RealTest] {msg}" print(formatted, flush=True) try: with open(self.log_file, "a", encoding="utf-8") as f: f.write(formatted + "\n") except Exception: pass def check_preconditions(self) -> Dict[str, Any]: """Kiểm tra tiền điều kiện khởi chạy phiên Real Test trên client game thật.""" self.log("--- BƯỚC 1: KIỂM TRA TIỀN ĐIỀU KIỆN CLIENT GAME THẬT ---") status: Dict[str, Any] = { "window_found": False, "focused": False, "rect": [], "initial_zone": "Unknown", "log_exists": os.path.exists(self.client_log_path), } hwnd = find_poe2_window() if not hwnd: self.log("❌ Không tìm thấy cửa sổ game Path of Exile 2! Hãy mở game trước.", level="ERROR") return status status["window_found"] = True focused = ensure_window_focus(hwnd) status["focused"] = focused rect = get_client_rect(hwnd) status["rect"] = list(rect) self.log(f"✅ Game Window HWND: 0x{hwnd:X} | Rect: {rect[2]-rect[0]}x{rect[3]-rect[1]} | Focused: {focused}") if self.log_watcher: seeded = self.log_watcher.seed_latest_playable_zone() if seeded and seeded.area_name: status["initial_zone"] = seeded.area_name self.log(f"📍 Khu vực hiện tại trong game: '{status['initial_zone']}'") # Chụp ảnh hiện trường ban đầu (Baseline) baseline_img = self.capturer.capture_screenshot(event_name="BASELINE_INITIAL_STATE", capture_ram=True) status["baseline_image"] = baseline_img self.log(f"📸 Đã chụp ảnh Baseline hiện trường: {baseline_img}") return status def handle_incident(self, stage_name: str, error_msg: str, exc: Optional[Exception] = None): """Kích hoạt F8 Pause khẩn cấp, chụp ảnh sự cố và đóng gói Incident Dossier (Rule 6).""" self.log(f"🚨 PHÁT HIỆN SỰ CỐ TẠI {stage_name}: {error_msg}", level="ERROR") self.log("🛑 KÍCH HOẠT F8 KHẨN CẤP (INV-F8-MUTES-BRAIN) & GIẢI PHÓNG TOÀN BỘ PHÍM BẤM...", level="WARN") emergency_f8_pause_and_release_all() # Tạo thư mục Incident Dossier ts_inc = datetime.now().strftime("%Y%m%d_%H%M%S") incident_dir = os.path.join("debug_harness", "incidents", f"INCIDENT_{ts_inc}_{stage_name}") os.makedirs(incident_dir, exist_ok=True) # Chụp ảnh hiện trường thất bại fail_img = self.capturer.capture_screenshot(event_name=f"INCIDENT_FAIL_{stage_name}", capture_ram=False) if fail_img and os.path.exists(fail_img): import shutil shutil.copy(fail_img, os.path.join(incident_dir, "incident_screenshot.png")) # Trích xuất 100 dòng log Client.txt gần nhất recent_client_log = [] if os.path.exists(self.client_log_path): try: with open(self.client_log_path, "r", encoding="utf-8", errors="ignore") as f: recent_client_log = f.readlines()[-100:] with open(os.path.join(incident_dir, "client_recent.log"), "w", encoding="utf-8") as f: f.writelines(recent_client_log) except Exception as e: self.log(f"Lỗi ghi incident client log: {e}", level="WARN") # Ghi báo cáo Incident Dossier incident_report = { "timestamp": ts_inc, "stage_failed": stage_name, "error_message": error_msg, "traceback": traceback.format_exc() if exc else "", "recent_client_log_tail": [l.strip() for l in recent_client_log[-10:]], "incident_dir": incident_dir, } with open(os.path.join(incident_dir, "incident_report.json"), "w", encoding="utf-8") as f: json.dump(incident_report, f, indent=2, ensure_ascii=False) self.log(f"📁 Incident Dossier đã được lưu tại: {incident_dir}", level="INFO") return incident_report def run(self) -> Dict[str, Any]: """Vận hành toàn bộ phiên Real Test tự hành có giám sát.""" start_time = time.time() self.log("==================================================================") self.log(f" BẮT ĐẦU PHIÊN REAL TEST TỰ HÀNH TOÀN DIỆN (Phiên: {self.timestamp})") self.log("==================================================================") preconditions = self.check_preconditions() if not preconditions.get("window_found"): return { "status": "FAILED", "error": "Game POE2 window not found", "preconditions": preconditions, } # Khởi tạo Coordinator tự hành coordinator = AutonomousMappingCoordinator( capture_dir=self.sentinel_dir, map_duration_sec=self.map_duration_sec, client_log_path=self.client_log_path, auto_detect_resolution=True, ) test_summary: Dict[str, Any] = { "session_id": self.timestamp, "status": "FAILED", "stages_completed": [], "incident": None, "evidence_images": [], "sentinel_images": [], "elapsed_seconds": 0.0, "preconditions": preconditions, } # Bắt đầu Sentinel 3s coordinator.start_3s_sentinel() try: # Quyết định stage bắt đầu actual_start_stage = self.start_stage baseline_img = preconditions.get("baseline_image") if actual_start_stage == 0: correlator = MultimodalTelemetryCorrelator(client_log_path=self.client_log_path) telemetry_report = correlator.correlate(img_path=baseline_img) rec = telemetry_report.get("recommendation", {}) actual_start_stage = rec.get("suggested_stage", 1) self.log(f"🧠 Viễn trắc đề xuất bắt đầu từ STAGE {actual_start_stage}: {rec.get('reason')}") test_summary["initial_telemetry"] = telemetry_report selected_slot = None if actual_start_stage <= 1 and self.stage_stop >= 1: # STAGE 1: Túi đồ & Waystone Scan self.log("\n>>> THỰC THI STAGE 1: Quét Túi Đồ & Xác Định Waystone...") s1 = coordinator.execute_stage1_precheck(target_slot_idx=0) if s1.get("status") != "SUCCESS": test_summary["incident"] = self.handle_incident("STAGE1_PRECHECK", s1.get("error", "Stage 1 failed")) return test_summary selected_slot = s1.get("selected_slot") test_summary["stages_completed"].append("STAGE1_PRECHECK") self.log(f"✅ STAGE 1 HOÀN TẤT: Đã chọn Waystone tại slot {selected_slot}") if self.stage_stop <= 1 or self.dry_run: self.log("Dừng phiên sau Stage 1 theo cấu hình.") test_summary["status"] = "SUCCESS" return test_summary if actual_start_stage <= 2 and self.stage_stop >= 2: # STAGE 2: Tiếp cận & Mở Map Device self.log("\n>>> THỰC THI STAGE 2: Mở Giao Diện Map Device / Atlas...") s2 = coordinator.execute_stage2_map_device() if s2.get("status") != "SUCCESS": test_summary["incident"] = self.handle_incident("STAGE2_MAP_DEVICE", s2.get("error", "Stage 2 failed")) return test_summary test_summary["stages_completed"].append("STAGE2_MAP_DEVICE") self.log("✅ STAGE 2 HOÀN TẤT: Giao diện Atlas đã mở và chọn tab ENDGAME.") if self.stage_stop <= 2: test_summary["status"] = "SUCCESS" return test_summary if actual_start_stage <= 3 and self.stage_stop >= 3: # STAGE 3: Chọn Node & Nạp Waystone & Mở Portal self.log("\n>>> THỰC THI STAGE 3: Nạp Waystone & Kích Hoạt Portal...") if not selected_slot: selected_slot = { "col": 0, "row": 0, "center_x": int(1731 * coordinator.scale_x), "center_y": int(822 * coordinator.scale_y), "avg_rgb": 58.9, } s3 = coordinator.execute_stage3_atlas_activation(selected_slot) if s3.get("status") != "SUCCESS": test_summary["incident"] = self.handle_incident("STAGE3_ATLAS_ACTIVATION", s3.get("error", "Stage 3 failed")) return test_summary test_summary["stages_completed"].append("STAGE3_ATLAS_ACTIVATION") self.log("✅ STAGE 3 HOÀN TẤT: Đã nạp Waystone và click TRAVERSE.") if self.stage_stop <= 3: test_summary["status"] = "SUCCESS" return test_summary entered_zone = preconditions.get("initial_zone", "Unknown Map") if actual_start_stage <= 4 and self.stage_stop >= 4: # STAGE 4: Bước vào Portal chuyển cảnh Map self.log("\n>>> THỰC THI STAGE 4: Bước Vào Portal & Chuyển Cảnh...") s4 = coordinator.execute_stage4_map_entrance() if s4.get("status") != "SUCCESS": test_summary["incident"] = self.handle_incident("STAGE4_MAP_ENTRANCE", s4.get("error", "Stage 4 failed")) return test_summary test_summary["stages_completed"].append("STAGE4_MAP_ENTRANCE") entered_zone = s4.get("entered_zone", "Unknown Map") self.log(f"✅ STAGE 4 HOÀN TẤT: Đã vào Map '{entered_zone}'!") if self.stage_stop <= 4: test_summary["status"] = "SUCCESS" return test_summary runner_instance = None if actual_start_stage <= 5 and self.stage_stop >= 5: # STAGE 5: In-Map Autonomous Clear Sequence self.log(f"\n>>> THỰC THI STAGE 5: Vận Hành Tự Trị Trong Map ({self.map_duration_sec}s)...") s5 = coordinator.execute_stage5_map_clear() if s5.get("status") != "SUCCESS": test_summary["incident"] = self.handle_incident("STAGE5_MAP_CLEAR", s5.get("error", "Stage 5 failed")) return test_summary test_summary["stages_completed"].append("STAGE5_MAP_CLEAR") runner_instance = s5.get("runner_instance") self.log("✅ STAGE 5 HOÀN TẤT: Đã dọn map an toàn.") if self.stage_stop <= 5: test_summary["status"] = "SUCCESS" return test_summary if actual_start_stage <= 6 and self.stage_stop >= 6: # STAGE 6: Town Portal Exit & Về Hideout self.log("\n>>> THỰC THI STAGE 6: Mở Portal Phím '.' & Quay Về Hideout...") s6 = coordinator.execute_stage6_portal_return(runner=runner_instance, map_zone=entered_zone) if s6.get("status") != "SUCCESS": test_summary["incident"] = self.handle_incident("STAGE6_PORTAL_RETURN", s6.get("error", "Stage 6 failed")) return test_summary test_summary["stages_completed"].append("STAGE6_PORTAL_RETURN") self.log("✅ STAGE 6 HOÀN TẤT: Đã quay về Hideout an toàn!") test_summary["status"] = "SUCCESS" except Exception as e: test_summary["incident"] = self.handle_incident("GLOBAL_EXCEPTION", str(e), exc=e) finally: coordinator.stop_3s_sentinel() test_summary["evidence_images"] = list(coordinator.captured_evidence) test_summary["sentinel_images"] = list(coordinator.sentinel_captures) test_summary["elapsed_seconds"] = round(time.time() - start_time, 2) report_file = os.path.join(self.session_dir, "session_report.json") with open(report_file, "w", encoding="utf-8") as f: json.dump(test_summary, f, indent=2, ensure_ascii=False, default=str) self.log(f"📊 Báo cáo tổng kết phiên Real Test đã ghi tại: {report_file}") self.log("==================================================================") self.log(f" KẾT THÚC PHIÊN REAL TEST: Trạng thái = {test_summary['status']} ({test_summary['elapsed_seconds']}s)") self.log("==================================================================") return test_summary if __name__ == "__main__": parser = argparse.ArgumentParser(description="AutoPOE2 Supervised Real Test Runner") parser.add_argument("--max-maps", type=int, default=1, help="Số lượng map cần test") parser.add_argument("--map-duration", type=int, default=25, help="Thời lượng in-map clear (giây)") parser.add_argument("--start-stage", type=int, default=0, help="Bắt đầu từ stage N (0: tự động phát hiện, 1-6)") parser.add_argument("--stage-stop", type=int, default=6, help="Dừng sau stage N (1-6)") parser.add_argument("--dry-run", action="store_true", help="Chỉ kiểm tra precheck không tương tác Map Device") args = parser.parse_args() runner = SupervisedRealTestRunner( max_maps=args.max_maps, map_duration_sec=args.map_duration, start_stage=args.start_stage, stage_stop=args.stage_stop, dry_run=args.dry_run, ) res = runner.run() print(json.dumps(res, indent=2, ensure_ascii=False, default=str))