From 90a0db005fc7c937c29137742e7adbc7ed2c70bf Mon Sep 17 00:00:00 2001 From: AirLongDian Date: Thu, 11 Jun 2026 11:30:31 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20P1-21=20=E8=A1=A5=E5=85=85=20=E2=80=94?= =?UTF-8?q?=20ADRWatcher=20=E8=87=AA=E5=8A=A8=E6=A3=80=E6=B5=8B=20+=20Part?= =?UTF-8?q?ialReplanner=20=E5=B1=80=E9=83=A8=E9=87=8D=E8=A7=84=E5=88=92?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 adr_watcher.py: SHA256 hash 监控 ADR 文件变更 - snapshot() 初始快照, detect_changes() 增量检测 - 识别 new/modified/superseded/deleted 四种变更类型 - _parse_status() 解析 ADR Status 字段 - 新增 partial_replanner.py: 局部重规划器 - replan() 仅生成受影响任务的替代计划 - _extract_stable_interfaces() 提取未受影响 DONE 任务接口约束 - 产出 replan-request.json 供 Arc 读取 - Eng monitor_engine 集成 ADR 检测: 每轮轮询调用 _detect_adr_changes() - Eng handle_adr_invalidation 集成 PartialReplanner 输出 - events.py 新增 ADR 变更相关事件常量 - task_graph.py apply_full_replace 保留 DISPATCHED, 不保留 INVALIDATED - 10 项补充测试全通过，含完整 ADR 流程场景 Co-Authored-By: Claude Sonnet 4.6 --- lib/air_runtime/adr_watcher.py | 109 ++++++++ lib/air_runtime/events.py | 3 + lib/air_runtime/modes/eng_mode.py | 49 ++++ lib/air_runtime/partial_replanner.py | 106 ++++++++ test_p1_21_supplement.py | 358 +++++++++++++++++++++++++++ 5 files changed, 625 insertions(+) create mode 100644 lib/air_runtime/adr_watcher.py create mode 100644 lib/air_runtime/partial_replanner.py create mode 100644 test_p1_21_supplement.py diff --git a/lib/air_runtime/adr_watcher.py b/lib/air_runtime/adr_watcher.py new file mode 100644 index 0000000..9b6061e --- /dev/null +++ b/lib/air_runtime/adr_watcher.py @@ -0,0 +1,109 @@ +""" +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" + path: str = "" + old_hash: str = "" + new_hash: str = "" + + +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] = {} + + 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) + self._known_hashes[adr_id] = hashlib.sha256( + adr_file.read_bytes() + ).hexdigest() + + 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: + changes.append(ADRChange( + adr_id=adr_id, kind="new", + path=str(adr_file), old_hash="", new_hash=current_hash, + )) + elif current_hash != old_hash: + status = self._parse_status(adr_file) + if status == "superseded": + changes.append(ADRChange( + adr_id=adr_id, kind="superseded", + path=str(adr_file), old_hash=old_hash, new_hash=current_hash, + )) + else: + changes.append(ADRChange( + adr_id=adr_id, kind="modified", + path=str(adr_file), old_hash=old_hash, new_hash=current_hash, + )) + self._known_hashes[adr_id] = current_hash + + # 检查被删除的 ADR + for adr_id in list(self._known_hashes.keys()): + if adr_id not in seen_ids: + changes.append(ADRChange( + adr_id=adr_id, kind="deleted", + path="", old_hash=self._known_hashes[adr_id], new_hash="", + )) + del self._known_hashes[adr_id] + + return changes + + @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" diff --git a/lib/air_runtime/events.py b/lib/air_runtime/events.py index 1cc1543..0932527 100644 --- a/lib/air_runtime/events.py +++ b/lib/air_runtime/events.py @@ -33,6 +33,9 @@ SEC_SCAN = "sec.scan" REVIEW_SESSION = "review.session" ENGINE_CYCLE = "engine.cycle" WORKER_TIMEOUT = "worker.timeout" +ADR_CHANGE_DETECTED = "adr.change.detected" +ADR_INVALIDATION = "adr.invalidation" +ADR_UNFREEZED = "adr.unfreezed" LOCK_ACQUIRED = "lock.acquired" LOCK_RELEASED = "lock.released" STALE_LOCK_CLEANED = "stale_lock.cleaned" diff --git a/lib/air_runtime/modes/eng_mode.py b/lib/air_runtime/modes/eng_mode.py index f40d495..506e42d 100644 --- a/lib/air_runtime/modes/eng_mode.py +++ b/lib/air_runtime/modes/eng_mode.py @@ -308,6 +308,32 @@ def dispatch_worker_group(project_root: Path, group_name: str = "") -> dict: return result +def _detect_adr_changes(project_root: Path, state: dict) -> list: + """P1-21: 检查 ADR 文件变更，返回需要级联失效的变更列表。""" + from air_runtime.adr_watcher import ADRWatcher, ADRChange + adr_dir = project_root / "AirPlan" / "docs" / "architecture" / "adr" + if not adr_dir.exists(): + return [] + + watcher = ADRWatcher(adr_dir) + # 从引擎状态恢复已知 hash + known = state.get("adrWatcherHashes", {}) + watcher._known_hashes = known + + # 首次无 snapshot → 先初始化 + if not known: + watcher.snapshot() + state["adrWatcherHashes"] = dict(watcher._known_hashes) + return [] + + changes = watcher.detect_changes() + # 持久化更新后的 hash + state["adrWatcherHashes"] = dict(watcher._known_hashes) + + # 只返回需要级联失效的变更 + return [c for c in changes if c.kind in ("superseded", "modified")] + + def monitor_engine(project_root: Path) -> dict: """L1 代码级轮询：硬编码循环检测 Worker 状态，不依赖 LLM 自觉。""" paths = _paths(project_root) @@ -350,6 +376,21 @@ def monitor_engine(project_root: Path) -> dict: except OSError: resource_pressure = False + # P1-21: ADR 变更自动检测 + adr_changes = _detect_adr_changes(project_root, state) + if adr_changes: + for change in adr_changes: + if change.kind in ("superseded", "modified"): + interventions.append({ + "adrId": change.adr_id, + "reason": f"adr-{change.kind}", + "action": "invalidate-by-adr", + }) + log = EventLog(event_log_path(project_root)) + log.emit("adr.change.detected", { + "adrId": change.adr_id, "kind": change.kind, + }) + # 新增：检查 pending worktree merges — merge 失败自动升级到 AirDbg wt_root = project_root / ".git" / "worktrees" if wt_root.exists(): @@ -616,6 +657,13 @@ def handle_adr_invalidation(project_root: Path, adr_id: str) -> dict: # 6: git revert 已合并的旧代码（按 task_id 查找对应 commit） revert_results = _git_revert_invalidated(project_root, report.invalidated_task_ids) + # 6.5: 生成局部重规划请求（PartialReplanner） + from air_runtime.partial_replanner import PartialReplanner + replanner = PartialReplanner() + partial_delta = replanner.replan(graph, report.invalidated_task_ids) + replan_request_path = paths["plan_dir"] / f"replan-request-{session_stamp()}.json" + atomic_json_write(replan_request_path, partial_delta.replan_request) + # 7: 写回更新后的 task-graph.json from air_runtime.modes.arc_mode import _export_task_graph_json _export_task_graph_json(graph, tg_json) @@ -641,6 +689,7 @@ def handle_adr_invalidation(project_root: Path, adr_id: str) -> dict: }, "terminatedWorkers": terminated_workers, "revertResults": revert_results, + "replanRequestPath": str(replan_request_path), "nextStep": "arc-replan-then-unfreeze", } diff --git a/lib/air_runtime/partial_replanner.py b/lib/air_runtime/partial_replanner.py new file mode 100644 index 0000000..7f8b01f --- /dev/null +++ b/lib/air_runtime/partial_replanner.py @@ -0,0 +1,106 @@ +""" +局部重规划 — P1-21 仅重新生成受 ADR 变更影响的任务子集。 +替代全量重规划，保留未受影响任务的接口约束。 +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from air_runtime.task_graph import TaskGraph, TaskNode, PlanDelta + + +@dataclass +class Interface: + """未受影响任务暴露的公共接口约束。""" + task_id: str + write_set: list[str] = field(default_factory=list) + adr_refs: list[str] = field(default_factory=list) + + +@dataclass +class ReplanContext: + """受影响任务的上下文信息，供 Arc 局部重规划使用。""" + task_id: str + task: str + files_dirs: str + done_when: str + write_set: list[str] = field(default_factory=list) + adr_refs: list[str] = field(default_factory=list) + status: str = "" + + +class PartialReplanner: + """仅重新生成受 ADR 变更影响的任务子集。 + + 与 incremental_replan_mode 的区别： + - incremental_replan_mode: 全量重建 DAG 再 diff + - PartialReplanner: 只对受影响部分重新规划，保留稳定接口约束 + """ + + def replan(self, graph: TaskGraph, invalidated_ids: list[str], + new_adr_path: Path | None = None) -> PlanDelta: + """局部重规划：仅生成受影响任务的替代任务。 + + Args: + graph: 当前任务图（已包含 INVALIDATED 标记） + invalidated_ids: 被 ADR 变更级联失效的任务 ID 列表 + new_adr_path: 新 ADR 文件路径（可选，供 Arc 参考） + """ + delta = PlanDelta() + + # 1. 收集受影响任务的上下文 + affected_context = self._collect_affected_context(graph, invalidated_ids) + + # 2. 提取未受影响任务的稳定接口 + stable_interfaces = self._extract_stable_interfaces(graph, set(invalidated_ids)) + + # 3. 生成局部重规划指令文件（供 Arc 读取） + replan_request = { + "type": "partial-replan", + "invalidatedTaskIds": invalidated_ids, + "affectedContext": [ctx.__dict__ for ctx in affected_context], + "stableInterfaces": [iface.__dict__ for iface in stable_interfaces], + "newAdrPath": str(new_adr_path) if new_adr_path else None, + } + + # 4. 构建增量 delta + # removed_tasks 已在 invalidate_by_adr 中填充 + # added_tasks 留空——由 Arc 读取 replan-request.json 后生成新任务 + delta.replan_request = replan_request + + return delta + + def _collect_affected_context(self, graph: TaskGraph, + invalidated_ids: list[str]) -> list[ReplanContext]: + """收集受影响任务的上下文。""" + contexts = [] + for tid in invalidated_ids: + node = graph.nodes.get(tid) + if node: + contexts.append(ReplanContext( + task_id=node.id, + task=node.task, + files_dirs=node.files_dirs, + done_when=node.done_when, + write_set=list(node.write_set), + adr_refs=list(node.adr_refs), + status=node.status, + )) + return contexts + + def _extract_stable_interfaces(self, graph: TaskGraph, + invalidated_ids: set) -> list[Interface]: + """提取未受影响 DONE 任务的接口约束，确保重规划不破坏依赖。""" + interfaces = [] + for nid, node in graph.nodes.items(): + if nid not in invalidated_ids and node.status == "DONE": + if node.write_set or node.adr_refs: + interfaces.append(Interface( + task_id=node.id, + write_set=list(node.write_set), + adr_refs=list(node.adr_refs), + )) + return interfaces diff --git a/test_p1_21_supplement.py b/test_p1_21_supplement.py new file mode 100644 index 0000000..d5acfa9 --- /dev/null +++ b/test_p1_21_supplement.py @@ -0,0 +1,358 @@ +#!/usr/bin/env python3 +"""P1-21 补充功能测试：ADRWatcher + PartialReplanner""" + +import sys +import tempfile +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent / "lib")) + + +def test_adr_watcher_snapshot(): + """测试 ADRWatcher 初始快照""" + from air_runtime.adr_watcher import ADRWatcher + + with tempfile.TemporaryDirectory() as tmpdir: + adr_dir = Path(tmpdir) + # 创建 ADR 文件 + (adr_dir / "ADR-0005-ffmpeg-decode.md").write_text( + "# ADR-0005: ffmpeg 解码\n\nStatus: accepted\n\n使用 ffmpeg 实现视频解码。" + ) + (adr_dir / "ADR-0012-auth.md").write_text( + "# ADR-0012: 认证方案\n\nStatus: accepted\n\n使用 JWT 认证。" + ) + + watcher = ADRWatcher(adr_dir) + watcher.snapshot() + + assert len(watcher._known_hashes) == 2 + assert "ADR-0005" in watcher._known_hashes + assert "ADR-0012" in watcher._known_hashes + + # 初始快照后无变更 + changes = watcher.detect_changes() + assert len(changes) == 0 + + print("✓ ADRWatcher 初始快照测试通过") + return True + + +def test_adr_watcher_detect_modified(): + """测试 ADRWatcher 检测 ADR 修改""" + from air_runtime.adr_watcher import ADRWatcher + + with tempfile.TemporaryDirectory() as tmpdir: + adr_dir = Path(tmpdir) + adr_file = adr_dir / "ADR-0005-ffmpeg-decode.md" + adr_file.write_text("# ADR-0005\n\nStatus: accepted\n\n用 ffmpeg。") + + watcher = ADRWatcher(adr_dir) + watcher.snapshot() + + # 修改 ADR 内容 + adr_file.write_text("# ADR-0005\n\nStatus: accepted\n\n改用 gstreamer。") + changes = watcher.detect_changes() + + assert len(changes) == 1 + assert changes[0].adr_id == "ADR-0005" + assert changes[0].kind == "modified" + assert changes[0].old_hash != changes[0].new_hash + + print("✓ ADRWatcher 检测修改测试通过") + return True + + +def test_adr_watcher_detect_superseded(): + """测试 ADRWatcher 检测 ADR 被替代""" + from air_runtime.adr_watcher import ADRWatcher + + with tempfile.TemporaryDirectory() as tmpdir: + adr_dir = Path(tmpdir) + adr_file = adr_dir / "ADR-0005-ffmpeg-decode.md" + adr_file.write_text("# ADR-0005\n\nStatus: accepted\n\n用 ffmpeg。") + + watcher = ADRWatcher(adr_dir) + watcher.snapshot() + + # 标记为 superseded + adr_file.write_text("# ADR-0005\n\nStatus: superseded\n\n被 ADR-0020 替代。") + changes = watcher.detect_changes() + + assert len(changes) == 1 + assert changes[0].kind == "superseded" + + print("✓ ADRWatcher 检测 superseded 测试通过") + return True + + +def test_adr_watcher_detect_new(): + """测试 ADRWatcher 检测新增 ADR""" + from air_runtime.adr_watcher import ADRWatcher + + with tempfile.TemporaryDirectory() as tmpdir: + adr_dir = Path(tmpdir) + (adr_dir / "ADR-0005-old.md").write_text("# ADR-0005\n\nStatus: accepted\n") + + watcher = ADRWatcher(adr_dir) + watcher.snapshot() + + # 新增 ADR + (adr_dir / "ADR-0020-gstreamer.md").write_text("# ADR-0020\n\nStatus: accepted\n") + changes = watcher.detect_changes() + + assert len(changes) == 1 + assert changes[0].adr_id == "ADR-0020" + assert changes[0].kind == "new" + + print("✓ ADRWatcher 检测新增 ADR 测试通过") + return True + + +def test_adr_watcher_extract_id(): + """测试 ADR ID 提取""" + from air_runtime.adr_watcher import ADRWatcher + + assert ADRWatcher._extract_adr_id(Path("ADR-0005-ffmpeg-decode.md")) == "ADR-0005" + assert ADRWatcher._extract_adr_id(Path("ADR-0012.md")) == "ADR-0012" + assert ADRWatcher._extract_adr_id(Path("ADR-9999-super-long-name.md")) == "ADR-9999" + + print("✓ ADR ID 提取测试通过") + return True + + +def test_adr_watcher_parse_status(): + """测试 ADR Status 字段解析""" + from air_runtime.adr_watcher import ADRWatcher + + with tempfile.TemporaryDirectory() as tmpdir: + f1 = Path(tmpdir) / "ADR-0001.md" + f1.write_text("# ADR-0001\n\nStatus: accepted\n") + assert ADRWatcher._parse_status(f1) == "accepted" + + f2 = Path(tmpdir) / "ADR-0002.md" + f2.write_text("# ADR-0002\n\nStatus: superseded\n") + assert ADRWatcher._parse_status(f2) == "superseded" + + f3 = Path(tmpdir) / "ADR-0003.md" + f3.write_text("# ADR-0003\n\n无 Status 字段\n") + assert ADRWatcher._parse_status(f3) == "unknown" + + print("✓ ADR Status 解析测试通过") + return True + + +def test_partial_replanner(): + """测试局部重规划器""" + from air_runtime.task_graph import TaskGraph, TaskNode, Edge + from air_runtime.partial_replanner import PartialReplanner + + graph = TaskGraph() + # 失效的任务 + graph.add_node(TaskNode(id="T-001", status="INVALIDATED", task="ffmpeg 解码器", + files_dirs="src/decoder.cpp", write_set=["src/decoder.cpp"], + adr_refs=["ADR-0005"])) + graph.add_node(TaskNode(id="T-002", status="INVALIDATED", task="ffmpeg 编码器", + files_dirs="src/encoder.cpp", write_set=["src/encoder.cpp"], + adr_refs=["ADR-0005"])) + # 稳定的已完成任务 + graph.add_node(TaskNode(id="T-003", status="DONE", task="UI 界面", + write_set=["src/ui/main.cpp"], adr_refs=["ADR-0008"])) + graph.add_node(TaskNode(id="T-004", status="DONE", task="日志模块", + write_set=["src/logger.cpp"], adr_refs=[])) + + replanner = PartialReplanner() + delta = replanner.replan(graph, ["T-001", "T-002"]) + + # 检查 replan_request 中的受影响上下文 + request = delta.replan_request + assert request["type"] == "partial-replan" + assert set(request["invalidatedTaskIds"]) == {"T-001", "T-002"} + assert len(request["affectedContext"]) == 2 + assert request["affectedContext"][0]["task_id"] == "T-001" + + # 检查稳定接口 + stable = request["stableInterfaces"] + assert len(stable) == 2 # T-003 和 T-004 + stable_ids = {s["task_id"] for s in stable} + assert "T-003" in stable_ids + assert "T-004" in stable_ids + + print("✓ 局部重规划器测试通过") + return True + + +def test_partial_replanner_stable_interfaces(): + """测试稳定接口提取""" + from air_runtime.task_graph import TaskGraph, TaskNode + from air_runtime.partial_replanner import PartialReplanner + + graph = TaskGraph() + graph.add_node(TaskNode(id="T-001", status="INVALIDATED", task="失效任务", adr_refs=["ADR-0005"])) + graph.add_node(TaskNode(id="T-002", status="DONE", task="完成有写集", + write_set=["src/a.cpp", "src/b.cpp"], adr_refs=["ADR-0008"])) + graph.add_node(TaskNode(id="T-003", status="DONE", task="完成无写集")) + graph.add_node(TaskNode(id="T-004", status="TODO", task="未开始")) + + replanner = PartialReplanner() + interfaces = replanner._extract_stable_interfaces(graph, {"T-001"}) + + # T-002 有写集，应该包含 + # T-003 无写集且无 adr_refs，不包含 + # T-004 不是 DONE，不包含 + assert len(interfaces) == 1 + assert interfaces[0].task_id == "T-002" + + print("✓ 稳定接口提取测试通过") + return True + + +def test_eng_monitor_detects_adr_change(): + """测试 Eng 监控循环检测 ADR 变更""" + from air_runtime.modes.eng_mode import enter_engine, monitor_engine, _detect_adr_changes + + with tempfile.TemporaryDirectory() as tmpdir: + project_root = Path(tmpdir) + + # 初始化引擎 + enter_engine(project_root) + + # 创建 ADR 目录（无文件） + adr_dir = project_root / "AirPlan" / "docs" / "architecture" / "adr" + adr_dir.mkdir(parents=True, exist_ok=True) + + # 创建初始 ADR + (adr_dir / "ADR-0005-ffmpeg.md").write_text("# ADR-0005\n\nStatus: accepted\n\n用 ffmpeg。") + + # 第一次监控：初始化 hash 快照 + from air_runtime.io import safe_json_load + from air_runtime.modes.eng_mode import _paths + paths = _paths(project_root) + state = safe_json_load(paths["state"]) or {} + changes = _detect_adr_changes(project_root, state) + assert len(changes) == 0 # 首次只初始化，无变更 + + # 保存 hash 到 state + from air_runtime.io import atomic_json_write + atomic_json_write(paths["state"], state) + + # 修改 ADR 为 superseded + (adr_dir / "ADR-0005-ffmpeg.md").write_text("# ADR-0005\n\nStatus: superseded\n\n被替代。") + + # 第二次检测 + state = safe_json_load(paths["state"]) or {} + changes = _detect_adr_changes(project_root, state) + assert len(changes) == 1 + assert changes[0].adr_id == "ADR-0005" + assert changes[0].kind == "superseded" + + print("✓ Eng 监控检测 ADR 变更测试通过") + return True + + +def test_full_adr_change_to_invalidation_flow(): + """完整场景：ADR 变更 → 自动检测 → 级联失效 → 局部重规划""" + from air_runtime.adr_watcher import ADRWatcher + from air_runtime.task_graph import TaskGraph, TaskNode, Edge, PlanDelta + from air_runtime.partial_replanner import PartialReplanner + + with tempfile.TemporaryDirectory() as tmpdir: + project_root = Path(tmpdir) + adr_dir = project_root / "docs" / "architecture" / "adr" + adr_dir.mkdir(parents=True) + + # 初始 ADR + adr_file = adr_dir / "ADR-0005-ffmpeg-decode.md" + adr_file.write_text("# ADR-0005\n\nStatus: accepted\n\n使用 ffmpeg 实现视频解码。") + + # 初始化 watcher + watcher = ADRWatcher(adr_dir) + watcher.snapshot() + + # 初始化 task graph + graph = TaskGraph() + graph.add_node(TaskNode(id="T-001", status="DONE", task="ffmpeg 解码器", + files_dirs="src/decoder.cpp", write_set=["src/decoder.cpp"], + adr_refs=["ADR-0005"])) + graph.add_node(TaskNode(id="T-002", status="TODO", task="解码器集成测试")) + graph.add_node(TaskNode(id="T-003", status="DONE", task="UI 界面", + write_set=["src/ui.cpp"], adr_refs=["ADR-0008"])) + graph.add_edge(Edge(source="T-001", target="T-002")) + + # 用户修改 ADR（ffmpeg → gstreamer） + adr_file.write_text("# ADR-0005\n\nStatus: superseded\n\n改用 gstreamer。") + + # Watcher 检测变更 + changes = watcher.detect_changes() + assert len(changes) == 1 + assert changes[0].kind == "superseded" + assert changes[0].adr_id == "ADR-0005" + + # 触发级联失效 + delta = PlanDelta() + report = graph.invalidate_by_adr("ADR-0005", delta) + assert report.invalidated_completed == 1 + assert report.cascaded_downstream == 1 + assert graph.dispatch_frozen == True + + # 局部重规划 + replanner = PartialReplanner() + partial_delta = replanner.replan(graph, report.invalidated_task_ids) + request = partial_delta.replan_request + assert request["type"] == "partial-replan" + assert "T-001" in request["invalidatedTaskIds"] + assert "T-002" in request["invalidatedTaskIds"] + # T-003 是稳定的 + assert len(request["stableInterfaces"]) == 1 + assert request["stableInterfaces"][0]["task_id"] == "T-003" + + # 解冻 + graph.add_node(TaskNode(id="T-101", status="TODO", task="gstreamer 解码器", + files_dirs="src/decoder.cpp", adr_refs=["ADR-0005"])) + graph.unfreeze_dispatch() + assert graph.dispatch_frozen == False + + print("✓ 完整 ADR 变更流程测试通过（检测→失效→重规划→解冻）") + return True + + +def main(): + print("=" * 50) + print("P1-21 补充功能测试：ADRWatcher + PartialReplanner") + print("=" * 50) + + tests = [ + ("ADRWatcher 快照", test_adr_watcher_snapshot), + ("ADRWatcher 检测修改", test_adr_watcher_detect_modified), + ("ADRWatcher 检测 superseded", test_adr_watcher_detect_superseded), + ("ADRWatcher 检测新增", test_adr_watcher_detect_new), + ("ADR ID 提取", test_adr_watcher_extract_id), + ("ADR Status 解析", test_adr_watcher_parse_status), + ("局部重规划器", test_partial_replanner), + ("稳定接口提取", test_partial_replanner_stable_interfaces), + ("Eng 监控检测 ADR", test_eng_monitor_detects_adr_change), + ("完整 ADR 流程", test_full_adr_change_to_invalidation_flow), + ] + + passed = 0 + failed = 0 + + for name, test_fn in tests: + try: + test_fn() + passed += 1 + except Exception as e: + print(f"✗ {name} 失败: {e}") + import traceback + traceback.print_exc() + failed += 1 + + print("=" * 50) + print(f"测试结果: {passed} 通过, {failed} 失败") + print("=" * 50) + + return failed == 0 + + +if __name__ == "__main__": + success = main() + sys.exit(0 if success else 1)