AirPlan-V2/lib/air_runtime/adr_watcher.py

"""
ADR 文件变更监控 — P1-21 ADR 变更自动检测。
基于 SHA256 hash 对比检测 ADR 文件变更，自动触发级联失效。
"""

from __future__ import annotations

import hashlib
import re
from dataclasses import dataclass, field
from pathlib import Path


@dataclass
class ADRChange:
    """ADR 文件变更记录。"""
    adr_id: str
    kind: str  # "new" | "superseded" | "modified" | "deleted"
    path: str = ""
    old_hash: str = ""
    new_hash: str = ""
    old_content: str | None = None  # 3.2.17: 变更前内容，供 ChangeClassifier 分类


class ADRWatcher:
    """监控 ADR 文件变更，自动触发级联失效。

    AirEng 在每轮轮询时调用 detect_changes()，
    发现 superseded 或 modified 变更时自动触发 invalidate_by_adr()。
    """

    def __init__(self, adr_dir: Path):
        self._adr_dir = adr_dir
        self._known_hashes: dict[str, str] = {}
        self._content_snapshots: dict[str, str] = {}  # 3.2.17: 内容快照供 ChangeClassifier 使用

    def snapshot(self) -> None:
        """启动时记录所有 ADR 的内容 hash 和内容快照。"""
        if not self._adr_dir.exists():
            return
        for adr_file in sorted(self._adr_dir.glob("ADR-*.md")):
            adr_id = self._extract_adr_id(adr_file)
            raw = adr_file.read_bytes()
            self._known_hashes[adr_id] = hashlib.sha256(raw).hexdigest()
            self._content_snapshots[adr_id] = adr_file.read_text(encoding="utf-8")

    def detect_changes(self) -> list[ADRChange]:
        """对比当前 ADR hash 与已知 hash，返回变更列表。"""
        if not self._adr_dir.exists():
            return []

        changes: list[ADRChange] = []
        seen_ids: set[str] = set()

        for adr_file in sorted(self._adr_dir.glob("ADR-*.md")):
            adr_id = self._extract_adr_id(adr_file)
            seen_ids.add(adr_id)
            current_hash = hashlib.sha256(adr_file.read_bytes()).hexdigest()
            old_hash = self._known_hashes.get(adr_id)

            if old_hash is None:
                content = adr_file.read_text(encoding="utf-8")
                changes.append(ADRChange(
                    adr_id=adr_id, kind="new",
                    path=str(adr_file), old_hash="", new_hash=current_hash,
                ))
                self._content_snapshots[adr_id] = content
            elif current_hash != old_hash:
                old_content = self._content_snapshots.get(adr_id)
                new_content = adr_file.read_text(encoding="utf-8")
                status = self._parse_status(adr_file)
                kind = "superseded" if status == "superseded" else "modified"
                changes.append(ADRChange(
                    adr_id=adr_id, kind=kind,
                    path=str(adr_file), old_hash=old_hash, new_hash=current_hash,
                    old_content=old_content,
                ))
                self._content_snapshots[adr_id] = new_content
            self._known_hashes[adr_id] = current_hash

        # 检查被删除的 ADR
        for adr_id in list(self._known_hashes.keys()):
            if adr_id not in seen_ids:
                old_content = self._content_snapshots.get(adr_id)
                changes.append(ADRChange(
                    adr_id=adr_id, kind="deleted",
                    path="", old_hash=self._known_hashes[adr_id], new_hash="",
                    old_content=old_content,
                ))
                del self._known_hashes[adr_id]

        return changes

    def get_content_for_classification(self, change: ADRChange) -> tuple[str | None, str | None]:
        """3.2.17: 获取 ADR 变更的旧/新内容，供 ChangeClassifier 爆炸半径分类。

        返回 (old_content, new_content):
        - modified/superseded/deleted: old=change.old_content, new=从磁盘读取
        - new: old=None, new=从磁盘读取
        """
        if change.kind == "deleted":
            return (change.old_content, None)
        if change.kind == "new":
            new_path = Path(change.path) if change.path else None
            new_content = new_path.read_text(encoding="utf-8") if new_path and new_path.exists() else None
            return (None, new_content)
        # modified / superseded: old_content already captured in change object
        new_path = Path(change.path) if change.path else None
        new_content = new_path.read_text(encoding="utf-8") if new_path and new_path.exists() else None
        return (change.old_content, new_content)

    @staticmethod
    def _extract_adr_id(adr_file: Path) -> str:
        """从文件名提取 ADR ID，如 'ADR-0005-ffmpeg-decode.md' → 'ADR-0005'。"""
        match = re.match(r"(ADR-\d+)", adr_file.stem)
        if match:
            return match.group(1)
        return adr_file.stem

    @staticmethod
    def _parse_status(adr_file: Path) -> str:
        """解析 ADR 文件中的 Status 字段。"""
        try:
            text = adr_file.read_text(encoding="utf-8")
        except (OSError, UnicodeDecodeError):
            return "unknown"
        for line in text.splitlines():
            lower = line.lower().strip()
            if lower.startswith("status:") or lower.startswith("status :"):
                status = line.split(":", 1)[1].strip().lower()
                return status
        return "unknown"