""" AutoPOE2 - Headless Scenario Harness Tests (Rule 11 & Rule 14 Compliant) Kiểm thử 5 kịch bản End-to-End Headless không cần game thật và không cần quyền Admin: 1. test_scenario_swamp_flat_ground_xyz: Giả lập di chuyển 2D (Z = const, X,Y thay đổi), không trigger ERR_XYZ_FROZEN. 2. test_scenario_bogus_entity_778_rejection: Giả lập struct rác HP 778/778 và ES 32763 từ RAM thật, phát hiện ERR_BOGUS_VITAL_STRUCT_778. 3. test_scenario_optical_vs_ram_desync: Giả lập Optical HP 503 vs RAM HP 342, bắt đúng ERR_OPTICAL_RAM_DESYNC. 4. test_scenario_watchdog_passive_degrade: Mất nhịp tim, Core chuyển sang Co-Pilot Passive, không tự sát. 5. test_scenario_multimodal_flight_recorder_incident_export: Anomaly export đầy đủ incident folder, dossier, png, ram dump, LATEST_INCIDENT.json. Tuân thủ Rule 14: Sử dụng dữ liệu thật trích xuất từ captures/ và core_log.txt thực tế của client POE2. """ import json import os import shutil import sys import tempfile import time import unittest sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) from src.agentic_engine.flight_recorder import MultimodalFlightRecorder from src.agentic_engine.invariant_sentinel import ( ERR_BOGUS_VITAL_STRUCT_778, ERR_OPTICAL_RAM_DESYNC, ERR_XYZ_FROZEN, InvariantSentinel, ) from src.agentic_engine.realtime_watcher import RealtimeWatcher from src.assistant_tool.core_shm_bridge import CoreTelemetrySnapshot, PlayerTelemetry from src.assistant_tool.optical_hp_sensor import OpticalVitals class TestHarnessScenarios(unittest.TestCase): """Bộ kiểm thử kịch bản giả lập Headless 100% User-Mode (Zero Admin, Zero Game Dependency).""" def setUp(self): self.test_dir = tempfile.mkdtemp(prefix="poe2_harness_test_") self.sentinel = InvariantSentinel(min_planar_delta=0.8, optical_divergence_threshold=0.25, watchdog_timeout_ms=3000) self.flight_recorder = MultimodalFlightRecorder(output_dir=self.test_dir, capacity=30) self.watcher = RealtimeWatcher(sentinel=self.sentinel, flight_recorder=self.flight_recorder) def tearDown(self): if os.path.exists(self.test_dir): shutil.rmtree(self.test_dir, ignore_errors=True) # ========================================================== # KỊCH BẢN 1: Di Chuyển Mặt Phẳng Đầm Lầy (Swamp Flat Ground - Z = const) # ========================================================== def test_scenario_swamp_flat_ground_xyz(self): """ Kịch bản 1: Giả lập nhân vật di chuyển 2D trên đầm lầy Sandswept Marsh (Z = const = 125.5f, X và Y thay đổi). Khẳng định: Watcher xác nhận XYZ tracking hợp lệ, TUYỆT ĐỐI KHÔNG trigger ERR_XYZ_FROZEN. Đối chứng: Khi is_moving = True nhưng tọa độ kẹt tại (0.0, 0.0, 0.0), Watcher BẮT BUỘC cắm cờ ERR_XYZ_FROZEN. """ # 1. Trạng thái ban đầu: Nhân vật đứng tại đầm lầy Sandswept Marsh (Z = 125.5f) p_init = PlayerTelemetry( current_hp=491, max_hp=491, current_mana=340, max_mana=340, current_es=200, max_es=200, pos_x=1500.0, pos_y=3200.0, pos_z=125.5 ) snap_init = CoreTelemetrySnapshot( is_online=True, snapshot_id=1, timestamp_ms=1000, player=p_init, area_name="Sandswept Marsh" ) self.watcher.process_telemetry(snapshot=snap_init, is_moving=False) # 2. Nhân vật chạy trên mặt nước phẳng: X và Y thay đổi (+25.4f, +30.1f), Z GIỮ NGUYÊN HOÀN TOÀN (125.5f) p_moved = PlayerTelemetry( current_hp=491, max_hp=491, current_mana=340, max_mana=340, current_es=200, max_es=200, pos_x=1525.4, pos_y=3230.1, pos_z=125.5 # delta_x = 25.4, delta_y = 30.1, delta_z = 0.0 ) snap_moved = CoreTelemetrySnapshot( is_online=True, snapshot_id=2, timestamp_ms=1050, player=p_moved, area_name="Sandswept Marsh" ) errors = self.watcher.process_telemetry(snapshot=snap_moved, is_moving=True) # Khẳng định: Watcher chấp nhận chuyển động mặt phẳng 2D, không trigger ERR_XYZ_FROZEN self.assertNotIn(ERR_XYZ_FROZEN, errors, "Watcher không được trigger ERR_XYZ_FROZEN khi nhân vật di chuyển 2D trên đầm lầy") self.assertNotIn(ERR_XYZ_FROZEN, self.watcher.active_errors) self.assertEqual(self.watcher.last_known_xyz, (1525.4, 3230.1, 125.5)) # 3. Phép thử đối chứng nghịch đảo (Adversarial Check): # Nhân vật đang di chuyển (is_moving = True) nhưng tọa độ bị kẹt cứng tại (0.0, 0.0, 0.0) # (Tái hiện lỗi từ capture thực tế captures/20260907_192644_088_PERIODIC_ANALYSIS.json) p_frozen = PlayerTelemetry( current_hp=491, max_hp=491, pos_x=0.0, pos_y=0.0, pos_z=0.0 ) snap_frozen = CoreTelemetrySnapshot( is_online=True, snapshot_id=3, timestamp_ms=1100, player=p_frozen, area_name="Sandswept Marsh" ) errors_frozen = self.watcher.process_telemetry(snapshot=snap_frozen, is_moving=True) self.assertIn(ERR_XYZ_FROZEN, errors_frozen, "Watcher BẮT BUỘC cắm cờ ERR_XYZ_FROZEN khi tọa độ kẹt tại (0,0,0) lúc đang chạy") self.assertIn(ERR_XYZ_FROZEN, self.watcher.active_errors) # ========================================================== # KỊCH BẢN 2: Từ Chối Struct Rác Entity 778/778 & ES 32763 # ========================================================== def test_scenario_bogus_entity_778_rejection(self): """ Kịch bản 2: Giả lập struct rác HP 778/778, ES 32763, Mana 779/779 trong snapshot. Dữ liệu đối chiếu: Trích xuất trực tiếp từ captures/20260907_192644_088_PERIODIC_ANALYSIS_RAM.json và bin/Release/core_log.txt. Khẳng định: Invariant Sentinel phát hiện và cắm cờ ERR_BOGUS_VITAL_STRUCT_778. Đối chứng: Snapshot máu thật (HP 491/491 hoặc 1023/1250) được thông qua 100%. """ # 1. Trích xuất fixture từ dữ liệu thật captures/20260907_192644_088_PERIODIC_ANALYSIS_RAM.json real_incident_file = os.path.join(os.path.dirname(__file__), "..", "captures", "20260907_192644_088_PERIODIC_ANALYSIS_RAM.json") if os.path.exists(real_incident_file): with open(real_incident_file, "r", encoding="utf-8") as f: real_ram = json.load(f) real_p = real_ram["player"] bogus_hp = real_p["hp"] # 778 bogus_max_hp = real_p["max_hp"] # 778 bogus_es = real_p["es"] # 32763 bogus_max_es = real_p["max_es"] # 54 bogus_mana = real_p["mana"] # 779 bogus_max_mana = real_p["max_mana"] # 779 else: # Fallback nếu đường dẫn tương đối khác bogus_hp, bogus_max_hp = 778, 778 bogus_es, bogus_max_es = 32763, 54 bogus_mana, bogus_max_mana = 779, 779 # Tạo snapshot mang dữ liệu rác từ RAM dump thực tế p_bogus = PlayerTelemetry( current_hp=bogus_hp, max_hp=bogus_max_hp, current_es=bogus_es, max_es=bogus_max_es, current_mana=bogus_mana, max_mana=bogus_max_mana ) snap_bogus = CoreTelemetrySnapshot( is_online=True, snapshot_id=42, timestamp_ms=1788784004860, player=p_bogus, area_name="Sandswept Marsh" ) valid, err_code, desc = self.sentinel.evaluate_vitals(snap_bogus) # Khẳng định: Invariant Sentinel phát hiện và cắm cờ ERR_BOGUS_VITAL_STRUCT_778 self.assertFalse(valid, "Struct 778/778 và ES 32763/54 BẮT BUỘC bị Invariant Sentinel từ chối") self.assertEqual(err_code, ERR_BOGUS_VITAL_STRUCT_778) self.assertIn("778", desc) # 2. Phép thử đối chứng: Snapshot máu thật của người chơi # (Từ captures/20260907_125229_242_PERIODIC_ANALYSIS.json: HP 1023/1250, Mana 300/300, Spirit 100/100) p_real = PlayerTelemetry( current_hp=1023, max_hp=1250, current_es=0, max_es=0, current_mana=300, max_mana=300, current_spirit=100, max_spirit=100 ) snap_real = CoreTelemetrySnapshot( is_online=True, snapshot_id=43, timestamp_ms=1788784004900, player=p_real, area_name="The Riverbank" ) valid_real, err_real, _ = self.sentinel.evaluate_vitals(snap_real) self.assertTrue(valid_real, "Vitals thật (HP 1023/1250, Mana 300/300) phải được chấp thuận") self.assertIsNone(err_real) # ========================================================== # KỊCH BẢN 3: Lệch Pha Quang Học vs RAM (Optical vs RAM Desync) # ========================================================== def test_scenario_optical_vs_ram_desync(self): """ Kịch bản 3: Giả lập Optical HP đọc 503 trong khi RAM đọc 342. Độ lệch: |503/503 - 342/503| = |1.0 - 0.68| = 0.32 (32% > 25% tolerance). Khẳng định: Watcher bắt đúng lỗi ERR_OPTICAL_RAM_DESYNC. Đối chứng: Khi Optical = 500 và RAM = 490 (độ lệch 2% <= 25%), không kích hoạt lỗi. """ # 1. Khởi tạo snapshot với RAM HP = 342 p_ram = PlayerTelemetry( current_hp=342, max_hp=503, current_mana=300, max_mana=300, pos_x=1500.0, pos_y=3200.0, pos_z=125.5 ) snap = CoreTelemetrySnapshot( is_online=True, snapshot_id=100, timestamp_ms=2000, player=p_ram, area_name="Sandswept Marsh" ) # 2. Cảm biến quang học quả cầu máu đọc trực tiếp từ pixel màn hình: HP = 503 / 503 (Ratio = 1.0) optical = OpticalVitals( current_hp=503, max_hp=503, hp_ratio=1.0, is_detected=True, detection_method="raster_match" ) errors = self.watcher.process_telemetry(snapshot=snap, optical_vitals=optical, is_moving=False) # Khẳng định: Watcher bắt đúng ERR_OPTICAL_RAM_DESYNC self.assertIn(ERR_OPTICAL_RAM_DESYNC, errors, "Watcher BẮT BUỘC phát hiện lệch pha khi Optical = 503 và RAM = 342") self.assertIn(ERR_OPTICAL_RAM_DESYNC, self.watcher.active_errors) # 3. Phép thử đối chứng: Khi Optical = 500 và RAM = 490 (chênh lệch 2% < 25%) p_coherent = PlayerTelemetry(current_hp=490, max_hp=500) snap_coherent = CoreTelemetrySnapshot(is_online=True, snapshot_id=101, timestamp_ms=2050, player=p_coherent) optical_coherent = OpticalVitals(current_hp=500, max_hp=500, hp_ratio=1.0, is_detected=True) errors_coherent = self.watcher.process_telemetry(snapshot=snap_coherent, optical_vitals=optical_coherent, is_moving=False) self.assertNotIn(ERR_OPTICAL_RAM_DESYNC, errors_coherent, "Không được báo lỗi khi độ lệch pha chỉ 2% <= 25%") self.assertNotIn(ERR_OPTICAL_RAM_DESYNC, self.watcher.active_errors) # ========================================================== # KỊCH BẢN 4: Watchdog Hạ Cấp An Toàn Co-Pilot Passive (INV-WDG-01) # ========================================================== def test_scenario_watchdog_passive_degrade(self): """ Kịch bản 4: Giả lập mất nhịp tim Watchdog (>3000ms). Khẳng định: Core tự động chuyển sang Co-Pilot Passive, TUYỆT ĐỐI KHÔNG tự sát (request_stop == False). Khi nhịp tim phục hồi: Tự động khôi phục về chế độ Tự Hành Đầy Đủ (Full Autonomous). """ t0 = 100000 self.watcher.update_heartbeat(t0) self.assertFalse(self.watcher.is_copilot_passive) self.assertFalse(self.watcher.request_stop) # 1. Sau 1000ms: Nhịp tim bình thường is_healthy = self.watcher.check_watchdog(now_ms=t0 + 1000, timeout_ms=3000) self.assertTrue(is_healthy) self.assertFalse(self.watcher.is_copilot_passive) # 2. Sau 3500ms không có nhịp tim: Watchdog timeout (>3000ms) is_healthy_timeout = self.watcher.check_watchdog(now_ms=t0 + 3500, timeout_ms=3000) self.assertFalse(is_healthy_timeout, "Watchdog phải báo vi phạm khi quá 3500ms không có nhịp tim") self.assertTrue(self.watcher.is_copilot_passive, "Core PHẢI chuyển sang chế độ Co-Pilot Passive") # BẤT BIẾN SINH TỒN QUAN TRỌNG NHẤT: Core CẤM tự sát! self.assertFalse(self.watcher.request_stop, "Core TUYỆT ĐỐI KHÔNG được đặt cờ request_stop khi mất nhịp tim") # 3. Nhịp tim quay trở lại: Phục hồi chế độ tự hành đầy đủ self.watcher.update_heartbeat(t0 + 4000) self.assertFalse(self.watcher.is_copilot_passive, "Khi nhịp tim phục hồi, Co-Pilot Passive phải được tắt") self.assertFalse(self.watcher.request_stop) # ========================================================== # KỊCH BẢN 5: Xuất Hồ Sơ Sự Cố Multimodal Flight Recorder # ========================================================== def test_scenario_multimodal_flight_recorder_incident_export(self): """ Kịch bản 5: Giả lập kích hoạt Anomaly tại thời điểm 0ms. Khẳng định: 1. Tạo ra đầy đủ thư mục incident trong debug_harness/incidents/ 2. Sinh tệp incident_dossier.json đúng schema Doc 17 3. Sinh tệp screenshot.png giả lập hợp lệ 4. Sinh tệp ram_snapshot.json dump dữ liệu RAM 5. Cập nhật tệp chỉ mục LATEST_INCIDENT.json ở thư mục gốc """ # Sử dụng dữ liệu thật từ RAM dump ngày 07/09/2026 p = PlayerTelemetry( current_hp=778, max_hp=778, current_es=32763, max_es=54, current_mana=779, max_mana=779, pos_x=0.0, pos_y=0.0, pos_z=0.0 ) snap = CoreTelemetrySnapshot( is_online=True, snapshot_id=999, timestamp_ms=int(time.time() * 1000), player=p, area_name="Sandswept Marsh" ) optical = OpticalVitals(current_hp=491, max_hp=491, hp_ratio=1.0, is_detected=True) # Ghi nhận vài tick trước sự cố self.flight_recorder.record_tick(snap, optical) # Kích hoạt xuất hồ sơ sự cố 0ms result = self.flight_recorder.export_incident( error_code=ERR_BOGUS_VITAL_STRUCT_778, description="Phát hiện UI Entity Marker 778/778 gây ô nhiễm cache", snapshot=snap, optical_vitals=optical ) incident_dir = result["incident_dir"] dossier_path = result["dossier_path"] screenshot_path = result["screenshot_path"] ram_snapshot_path = result["ram_snapshot_path"] latest_incident_path = result["latest_incident_path"] # 1. Khẳng định thư mục incident tồn tại self.assertTrue(os.path.isdir(incident_dir), f"Thư mục incident phải tồn tại: {incident_dir}") self.assertIn("ERR_BOGUS_VITAL_STRUCT_778", incident_dir) # 2. Khẳng định incident_dossier.json tồn tại và có schema hợp lệ self.assertTrue(os.path.isfile(dossier_path), "incident_dossier.json phải được sinh ra") with open(dossier_path, "r", encoding="utf-8") as f: dossier = json.load(f) self.assertEqual(dossier["violation"]["invariant_code"], ERR_BOGUS_VITAL_STRUCT_778) self.assertEqual(dossier["system_context"]["zone_name"], "Sandswept Marsh") self.assertEqual(dossier["telemetry_state"]["reported_vitals"]["hp"], 778) self.assertEqual(dossier["telemetry_state"]["reported_vitals"]["es"], 32763) self.assertTrue(dossier["memory_diagnostics"]["suspected_pointer_pollution"]) # 3. Khẳng định screenshot.png tồn tại và là tệp PNG hợp lệ self.assertTrue(os.path.isfile(screenshot_path), "screenshot.png phải được lưu trữ") with open(screenshot_path, "rb") as f: header = f.read(8) self.assertEqual(header, b"\x89PNG\r\n\x1a\n", "screenshot.png phải có PNG magic header hợp lệ") # 4. Khẳng định ram_snapshot.json tồn tại và lưu thông tin RAM self.assertTrue(os.path.isfile(ram_snapshot_path), "ram_snapshot.json phải tồn tại") with open(ram_snapshot_path, "r", encoding="utf-8") as f: ram_dump = json.load(f) self.assertIn("in_game_state", ram_dump) # 5. Khẳng định LATEST_INCIDENT.json được cập nhật tại thư mục gốc self.assertTrue(os.path.isfile(latest_incident_path), "LATEST_INCIDENT.json phải tồn tại") with open(latest_incident_path, "r", encoding="utf-8") as f: latest = json.load(f) self.assertEqual(latest["invariant_code"], ERR_BOGUS_VITAL_STRUCT_778) self.assertEqual(latest["incident_dir"], os.path.abspath(incident_dir)) self.assertEqual(latest["dossier_path"], os.path.abspath(dossier_path)) if __name__ == "__main__": unittest.main()