AirPlan-V2/lib/air_runtime/review_runtime.py

"""
AirRvr 需求审查运行时 — V2 新增组件。
基于原始需求文档对已完成任务进行独立审查，验证交付物与需求的一致性。
"""

from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path

from air_runtime.io import atomic_json_write
from air_runtime.paths import rvr_state_path
from air_runtime.utils import session_stamp


@dataclass
class RequirementCoverage:
    requirement: str
    status: str  # covered | partial | missing
    evidence: str = ""


@dataclass
class CodeToDesignItem:
    design_item: str
    implementation_status: str  # aligned | divergent | missing
    code_location: str = ""
    design_location: str = ""
    divergence_detail: str = ""


@dataclass
class ReviewReport:
    task_id: str
    verdict: str  # pass | conditional-pass | fail
    coverage: list[RequirementCoverage] = field(default_factory=list)
    intent_alignment: str = "aligned"  # aligned | divergent
    divergence_notes: str = ""
    regression_risk: str = "none"  # none | low | medium | high
    code_quality: dict = field(default_factory=lambda: {
        "complexity": "low", "readability": "good", "duplication": "none", "error_handling": "complete",
    })
    lifecycle_health: dict = field(default_factory=lambda: {
        "resource_leak": "none", "connection_management": "proper",
        "timeout_strategy": "present", "retry_strategy": "present",
    })
    runtime_stability: dict = field(default_factory=lambda: {
        "crash_risk": "none", "race_condition": "none",
        "memory_leak": "none", "user_impact": "none",
    })
    code_to_design_table: list[CodeToDesignItem] = field(default_factory=list)
    logging_checks: dict = field(default_factory=lambda: {
        "spdlog_integrated": False,
        "non_standard_logging": [],
        "debug_release_switch": False,
        "critical_path_logging": False,
        "unified_format": False,
    })
    recommendations: list[str] = field(default_factory=list)


class ReviewRuntime:
    """AirRvr 审查运行时 — 管理审查会话和报告持久化。"""

    REVIEW_MODES = ["per-task", "per-wave", "per-milestone"]

    def __init__(self, project_root: Path):
        self._project_root = project_root
        self._state_dir = rvr_state_path(project_root).parent
        self._reviews_dir = self._state_dir / "reviews"
        self._reviews_dir.mkdir(parents=True, exist_ok=True)

    def save_report(self, report: ReviewReport) -> Path:
        report_path = self._reviews_dir / f"{report.task_id}-{session_stamp()}.json"
        atomic_json_write(report_path, self._report_to_dict(report))
        return report_path

    def load_report(self, task_id: str, timestamp: str) -> ReviewReport | None:
        from air_runtime.io import safe_json_load
        report_path = self._reviews_dir / f"{task_id}-{timestamp}.json"
        data = safe_json_load(report_path)
        if data:
            return self._dict_to_report(data)
        return None

    def get_integration_verdict(self, report: ReviewReport) -> str:
        """与 AirEng 集成：pass → 允许合并，conditional-pass → 合并但记录遗留项，fail → 阻止合并。"""
        return report.verdict

    def get_verdict_for_task(self, task_id: str) -> dict:
        """从持久化的 review report 读 verdict，返回 dict 含 verdict/residual/reportPath。
        没有 report 时返回 {"verdict": "pass", "reportPath": ""}（默认放行）。"""
        from air_runtime.io import safe_json_load
        # reports/ 是 AirEng 约定的存放路径（验证脚本和 eng_mode 期望的位置）
        reports_dir = self._state_dir / "reports"
        report_path = reports_dir / f"{task_id}.json"
        if not report_path.exists():
            # 兼容旧路径 reviews/ 下的 {task_id}-{ts}.json，找最新一份
            alt = self._reviews_dir
            if alt.exists():
                candidates = sorted(alt.glob(f"{task_id}-*.json"), key=lambda p: p.stat().st_mtime, reverse=True)
                if candidates:
                    report_path = candidates[0]
            if not report_path.exists():
                return {"verdict": "pass", "reportPath": "", "residual": []}
        report = safe_json_load(report_path)
        if not report or not isinstance(report, dict):
            return {"verdict": "pass", "reportPath": str(report_path), "residual": []}
        return {
            "verdict": report.get("verdict", "pass"),
            "residual": report.get("residual", []),
            "reportPath": str(report_path),
            "summary": report.get("summary", ""),
        }

    @staticmethod
    def _report_to_dict(report: ReviewReport) -> dict:
        return {
            "taskId": report.task_id,
            "verdict": report.verdict,
            "coverage": [c.__dict__ for c in report.coverage],
            "intentAlignment": report.intent_alignment,
            "divergenceNotes": report.divergence_notes,
            "regressionRisk": report.regression_risk,
            "codeQuality": report.code_quality,
            "lifecycleHealth": report.lifecycle_health,
            "runtimeStability": report.runtime_stability,
            "codeToDesignTable": [c.__dict__ for c in report.code_to_design_table],
            "loggingChecks": report.logging_checks,
            "recommendations": report.recommendations,
        }

    @staticmethod
    def _dict_to_report(data: dict) -> ReviewReport:
        return ReviewReport(
            task_id=data.get("taskId", ""),
            verdict=data.get("verdict", "fail"),
            coverage=[RequirementCoverage(**c) for c in data.get("coverage", [])],
            intent_alignment=data.get("intentAlignment", "aligned"),
            divergence_notes=data.get("divergenceNotes", ""),
            regression_risk=data.get("regressionRisk", "none"),
            code_quality=data.get("codeQuality", {}),
            lifecycle_health=data.get("lifecycleHealth", {}),
            runtime_stability=data.get("runtimeStability", {}),
            code_to_design_table=[CodeToDesignItem(**c) for c in data.get("codeToDesignTable", [])],
            logging_checks=data.get("loggingChecks", {}),
            recommendations=data.get("recommendations", []),
        )