#!/usr/bin/env python3 """AutoPOE2 - Daily 3:00 AM Automated Audit Engine (Rule 7 AGENTS.md). ====================================================================== Hệ thống tự động rà soát đồng bộ định kỳ lúc 3:00 AM (03:00:00): 1. Đối chiếu chéo mã nguồn C++ Core (Tier 1) và Python Companion (Tier 2). 2. Thẩm định cấu trúc nhị phân (Binary Struct Alignment & Offsets). 3. Kiểm tra tính toàn vẹn của phím tắt dừng khẩn cấp (Panic Killswitch SSoT). 4. Kiểm tra tần số nhịp vòng lặp (120Hz Hot Path vs tài liệu docs). 5. Xác thực 20 Bất biến Kiến trúc SSoT (Zero-Day Gate & Master Invariants). 6. Khóa chặn dữ liệu giả lập (Rule 14 Zero Mock Data Audit). 7. Xuất báo cáo kiểm toán vào docs/development/daily_audits/ và cập nhật LATEST_AUDIT.md. """ from __future__ import annotations import argparse from dataclasses import dataclass, field from datetime import datetime import json import os import re import subprocess import sys from typing import Any, Dict, List, Optional, Tuple if hasattr(sys.stdout, "reconfigure"): try: sys.stdout.reconfigure(encoding="utf-8", errors="replace") sys.stderr.reconfigure(encoding="utf-8", errors="replace") except Exception: pass ROOT_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) DOCS_AUDIT_DIR = os.path.join(ROOT_DIR, "docs", "development", "daily_audits") @dataclass class AuditFinding: code: str severity: str # CRITICAL, HIGH, MEDIUM, LOW file_path: str description: str is_passed: bool details: str = "" @dataclass class AuditReport: timestamp: str execution_time_s: float total_checks: int = 0 passed_checks: int = 0 failed_checks: int = 0 findings: List[AuditFinding] = field(default_factory=list) @property def is_success(self) -> bool: return self.failed_checks == 0 class DailyAuditEngine: """Bộ máy kiểm toán tự động AutoPOE2 lúc 3:00 AM.""" def __init__(self, verbose: bool = False): self.verbose = verbose self.findings: List[AuditFinding] = [] def log(self, msg: str) -> None: if self.verbose: print(f"[DailyAudit] {msg}") def audit_binary_struct_alignment(self) -> None: """Thẩm định cấu trúc nhị phân giữa protocol.hpp và core_shm_bridge.py.""" protocol_path = os.path.join(ROOT_DIR, "src", "core", "common", "protocol.hpp") bridge_path = os.path.join(ROOT_DIR, "src", "assistant_tool", "core_shm_bridge.py") if not os.path.exists(protocol_path) or not os.path.exists(bridge_path): self.findings.append(AuditFinding( code="AUDIT-BIN-01", severity="CRITICAL", file_path=protocol_path, description="Không tìm thấy tệp giao thức nhị phân protocol.hpp hoặc core_shm_bridge.py", is_passed=False )) return with open(protocol_path, "r", encoding="utf-8", errors="ignore") as f: proto_content = f.read() with open(bridge_path, "r", encoding="utf-8", errors="ignore") as f: bridge_content = f.read() # Kiểm tra kích thước cố định expected_sizes = [ ("SharedMemoryHeader", 64), ("CommandPacket", 36), ("TelemetryPacket", 39112), ("PlayerTelemetryData", 84), ("EntityTelemetryData", 88), ("AreaTelemetryData", 96), ] for struct_name, expected_size in expected_sizes: pattern = rf"sizeof\({struct_name}\)\s*==\s*(\d+)" m = re.search(pattern, proto_content) actual_size = int(m.group(1)) if m else None is_ok = (actual_size == expected_size) self.findings.append(AuditFinding( code=f"AUDIT-STRUCT-{struct_name.upper()}", severity="CRITICAL", file_path=protocol_path, description=f"Thẩm định kích thước struct nhị phân {struct_name}: {expected_size} bytes", is_passed=is_ok, details=f"Expected: {expected_size}B, Found in static assert: {actual_size}B" )) # Kiểm tra sự hiện diện của unpack_telemetry_packet SSoT trong core_shm_bridge.py has_ssot_unpack = "def unpack_telemetry_packet" in bridge_content self.findings.append(AuditFinding( code="AUDIT-SHM-SSOT-UNPACK", severity="HIGH", file_path=bridge_path, description="Kiểm tra hàm unpack_telemetry_packet SSoT duy nhất trong core_shm_bridge.py", is_passed=has_ssot_unpack, details="Found SSoT unpack" if has_ssot_unpack else "Missing unpack_telemetry_packet" )) def audit_panic_killswitch_hotkeys(self) -> None: """Thẩm định phím dừng khẩn cấp Panic Killswitch trên toàn bộ tài liệu và code.""" main_cpp = os.path.join(ROOT_DIR, "src", "core", "main.cpp") control_py = os.path.join(ROOT_DIR, "src", "assistant_tool", "control_center.py") readme_md = os.path.join(ROOT_DIR, "README.md") tool_readme = os.path.join(ROOT_DIR, "docs", "assistant_tool", "README.md") # Kiểm tra trong main.cpp with open(main_cpp, "r", encoding="utf-8", errors="ignore") as f: main_code = f.read() has_ctrl_shift_f12 = "VK_F12" in main_code and "VK_CONTROL" in main_code and "VK_SHIFT" in main_code has_pause = "VK_PAUSE" in main_code self.findings.append(AuditFinding( code="AUDIT-PANIC-CPP-MAIN", severity="CRITICAL", file_path=main_cpp, description="Core Engine bắt buộc hỗ trợ VK_PAUSE và tổ hợp Ctrl+Shift+VK_F12", is_passed=has_ctrl_shift_f12 and has_pause, details="VK_PAUSE and Ctrl+Shift+VK_F12 verified" )) # Kiểm tra tài liệu không được ghi phím F12 đơn lẻ if os.path.exists(tool_readme): with open(tool_readme, "r", encoding="utf-8", errors="ignore") as f: tool_doc = f.read() # Cấm mẫu 'Dừng khẩn cấp: Pause / F12' has_bad_single_f12 = bool(re.search(r"Dừng khẩn cấp:.*?Pause\s*/\s*F12(?!\w)", tool_doc)) self.findings.append(AuditFinding( code="AUDIT-PANIC-DOC-SSOT", severity="HIGH", file_path=tool_readme, description="Tài liệu không được chứa phím F12 đơn lẻ gây xung đột chụp ảnh Steam", is_passed=not has_bad_single_f12, details="Doc specifies Pause or Ctrl+Shift+F12 correctly" if not has_bad_single_f12 else "Found single F12 in doc!" )) def audit_loop_frequencies(self) -> None: """Thẩm định chu kỳ vòng lặp 120Hz (~8.33ms) thực tế và phát hiện lệch pha docs.""" main_cpp = os.path.join(ROOT_DIR, "src", "core", "main.cpp") with open(main_cpp, "r", encoding="utf-8", errors="ignore") as f: main_code = f.read() has_120hz_comment = "120Hz" in main_code has_8ms_sleep = "sleep_for(std::chrono::milliseconds(8))" in main_code or "sleep_for(std::chrono::microseconds(8333))" in main_code self.findings.append(AuditFinding( code="AUDIT-FREQ-CPP-120HZ", severity="HIGH", file_path=main_cpp, description="Core Engine vận hành ở nhịp Hot Path 120Hz (~8.33ms)", is_passed=has_120hz_comment and has_8ms_sleep, details="120Hz loop verified with ~8ms pacing" )) def audit_master_invariants(self) -> None: """Thẩm định danh mục 20 Bất biến Kiến trúc SSoT.""" gate_path = os.path.join(ROOT_DIR, "src", "assistant_tool", "zero_day_gate.py") if not os.path.exists(gate_path): return with open(gate_path, "r", encoding="utf-8", errors="ignore") as f: content = f.read() # Kiểm tra danh mục MASTER_INVARIANT_CODES m = re.search(r"MASTER_INVARIANT_CODES\s*=\s*\[(.*?)\]", content, re.DOTALL) if m: codes_block = m.group(1) codes = re.findall(r'"(INV-[^"]+)"', codes_block) count = len(codes) is_ok = (count >= 20) self.findings.append(AuditFinding( code="AUDIT-INV-COUNT-20", severity="HIGH", file_path=gate_path, description=f"Danh mục ZeroDayInvariantGate chứa đủ tối thiểu 20 bất biến SSoT: Tìm thấy {count}/20", is_passed=is_ok, details=f"Total codes: {count}. Codes: {', '.join(codes[:5])}..." )) def audit_zero_mock_data(self) -> None: """Thẩm định không chứa số liệu giả lập bịa đặt (Rule 14).""" sentinel_path = os.path.join(ROOT_DIR, "src", "agentic_engine", "master_invariants_sentinel.py") entity_path = os.path.join(ROOT_DIR, "src", "core", "memory", "entity_manager.cpp") # 1. Kiểm tra sentinel if os.path.exists(sentinel_path): with open(sentinel_path, "r", encoding="utf-8", errors="ignore") as f: sentinel_code = f.read() has_fake_4284 = '"current_es": 4284' in sentinel_code or '"max_es": 4284' in sentinel_code self.findings.append(AuditFinding( code="AUDIT-ZERO-MOCK-SENTINEL", severity="CRITICAL", file_path=sentinel_path, description="MasterInvariantsSentinel không được sinh dữ liệu giả lập (ES 4284)", is_passed=not has_fake_4284, details="Clean from mock telemetry" if not has_fake_4284 else "Contains forbidden fake ES 4284!" )) # 2. Kiểm tra entity manager if os.path.exists(entity_path): with open(entity_path, "r", encoding="utf-8", errors="ignore") as f: entity_code = f.read() has_hp_magic_boss = 'mMaxHp > 15000) ? "Elite Boss"' in entity_code self.findings.append(AuditFinding( code="AUDIT-ZERO-MOCK-ENTITY", severity="CRITICAL", file_path=entity_path, description="EntityManager không gán nhãn Boss giả lập theo ngưỡng HP > 15000", is_passed=not has_hp_magic_boss, details="Clean from fake boss labeling" if not has_hp_magic_boss else "Contains magic HP > 15000 boss label!" )) def audit_watchdog_timeout(self) -> None: """Thẩm định Watchdog Timeout 5000ms chuẩn xác.""" main_cpp = os.path.join(ROOT_DIR, "src", "core", "main.cpp") with open(main_cpp, "r", encoding="utf-8", errors="ignore") as f: main_code = f.read() has_5000ms = "kAgentWatchdogTimeoutMs = 5000" in main_code self.findings.append(AuditFinding( code="AUDIT-WATCHDOG-5000MS", severity="MEDIUM", file_path=main_cpp, description="Watchdog Timeout đồng bộ mức 5000ms (kAgentWatchdogTimeoutMs)", is_passed=has_5000ms, details="kAgentWatchdogTimeoutMs = 5000 verified" if has_5000ms else "Watchdog timeout mismatch" )) def run_automated_unit_tests(self) -> None: """Chạy nhanh bộ kiểm thử hồi quy trọng yếu.""" test_files = [ "tests/test_audit_invariants_fixes.py", "tests/test_coordinate_isometric_parity.py", "tests/test_ocr_coordinate_contract.py", "tests/test_waystone_crafter.py", ] cmd = [sys.executable, "-m", "pytest"] + test_files + ["-q"] try: res = subprocess.run(cmd, cwd=ROOT_DIR, capture_output=True, text=True, timeout=30) is_ok = (res.returncode == 0) self.findings.append(AuditFinding( code="AUDIT-PYTEST-REGRESSION", severity="HIGH", file_path="tests/", description="Thực thi bộ unit test hồi quy Python (Invariants, Coordinate Parity, Waystone Crafter)", is_passed=is_ok, details=res.stdout.strip().split("\n")[-1] if res.stdout else res.stderr )) except Exception as e: self.findings.append(AuditFinding( code="AUDIT-PYTEST-REGRESSION", severity="HIGH", file_path="tests/", description="Lỗi khi chạy pytest", is_passed=False, details=str(e) )) def run_all(self) -> AuditReport: """Chạy toàn bộ quy trình kiểm toán 3:00 AM.""" start_time = datetime.now() self.log("Bắt đầu kiểm toán hệ thống AutoPOE2...") self.audit_binary_struct_alignment() self.audit_panic_killswitch_hotkeys() self.audit_loop_frequencies() self.audit_master_invariants() self.audit_zero_mock_data() self.audit_watchdog_timeout() self.run_automated_unit_tests() end_time = datetime.now() duration = (end_time - start_time).total_seconds() passed = sum(1 for f in self.findings if f.is_passed) failed = sum(1 for f in self.findings if not f.is_passed) report = AuditReport( timestamp=start_time.strftime("%d/%m/%Y %H:%M:%S"), execution_time_s=duration, total_checks=len(self.findings), passed_checks=passed, failed_checks=failed, findings=self.findings ) return report def save_report(self, report: AuditReport) -> Tuple[str, str]: """Lưu báo cáo kiểm toán ra tệp Markdown.""" os.makedirs(DOCS_AUDIT_DIR, exist_ok=True) ts_slug = datetime.now().strftime("%Y%m%d_%H%M%S") report_file = os.path.join(DOCS_AUDIT_DIR, f"AUDIT_{ts_slug}.md") latest_file = os.path.join(DOCS_AUDIT_DIR, "LATEST_AUDIT.md") status_badge = "🟢 PASS (100% SSoT)" if report.is_success else f"🔴 FAILED ({report.failed_checks} checks failed)" md_content = f"""# Báo Cáo Kiểm Toán Tự Động Định Kỳ 3:00 AM (Rule 7 AGENTS.md) **Mốc thời gian thực thi**: `{report.timestamp}` **Thời gian xử lý**: `{report.execution_time_s:.2f}s` **Kết quả chung**: **{status_badge}** **Tổng số mục kiểm tra**: `{report.total_checks}` (Đạt: `{report.passed_checks}`, Không đạt: `{report.failed_checks}`) --- ## 1. Chi Tiết Các Hạng Mục Kiểm Toán | Mã Kiểm Tra | Mức Độ | Trạng Thái | Tệp Liên Quan | Mô Tả & Chi Tiết | | :--- | :--- | :---: | :--- | :--- | """ for f in report.findings: st = "✅ PASS" if f.is_passed else f"❌ **{f.severity}**" md_content += f"| `{f.code}` | `{f.severity}` | {st} | `{f.file_path}` | {f.description} ({f.details}) |\n" md_content += f""" --- ## 2. Kết Luận & Hành Động Tiếp Theo - Hệ thống C++ Core và Python Companion đạt tỷ lệ đồng bộ: `{report.passed_checks / report.total_checks * 100.0:.1f}%`. - {"Tất cả các bất biến và hợp đồng giao tiếp đều bảo đảm 100% Zero-Drift." if report.is_success else "Cần rà soát và khắc phục ngay các mục thất bại ở trên."} """ with open(report_file, "w", encoding="utf-8") as f: f.write(md_content) with open(latest_file, "w", encoding="utf-8") as f: f.write(md_content) return report_file, latest_file def main(): parser = argparse.ArgumentParser(description="AutoPOE2 - Daily 3:00 AM Automated Audit") parser.add_argument("--json", action="store_true", help="Xuất kết quả dưới dạng JSON") parser.add_argument("--verbose", action="store_true", help="In chi tiết log kiểm toán") args = parser.parse_args() engine = DailyAuditEngine(verbose=args.verbose) report = engine.run_all() report_file, latest_file = engine.save_report(report) if args.json: data = { "timestamp": report.timestamp, "execution_time_s": report.execution_time_s, "total_checks": report.total_checks, "passed_checks": report.passed_checks, "failed_checks": report.failed_checks, "is_success": report.is_success, "report_file": report_file, "latest_file": latest_file, "findings": [ { "code": f.code, "severity": f.severity, "file_path": f.file_path, "description": f.description, "is_passed": f.is_passed, "details": f.details } for f in report.findings ] } print(json.dumps(data, indent=2, ensure_ascii=False)) else: status_str = "THÀNH CÔNG (100% PASS)" if report.is_success else f"CÓ {report.failed_checks} LỖI" print(f"[DailyAudit 3:00 AM] Hoàn tất kiểm toán: {status_str} ({report.passed_checks}/{report.total_checks} checks passed in {report.execution_time_s:.2f}s)") print(f"[DailyAudit 3:00 AM] Báo cáo đã lưu tại: {report_file}") sys.exit(0 if report.is_success else 1) if __name__ == "__main__": main()