From 9702f1b186fe253083649e65dad1d84bb4d7baf2 Mon Sep 17 00:00:00 2001 From: AirPlan Date: Mon, 15 Jun 2026 14:45:41 +0800 Subject: [PATCH] =?UTF-8?q?feat:=203.2.17=E4=B8=AD=E9=80=94=E5=8F=98?= =?UTF-8?q?=E6=9B=B4=E5=A4=84=E7=90=86(git=E9=A9=B1=E5=8A=A8)=20+=203.2.9a?= =?UTF-8?q?=20Worker=20git=E5=BC=BA=E5=88=B6=20+=203.2.5=20Phase7=20squash?= =?UTF-8?q?=20merge=20+=20Arc=20git=E5=88=9D=E5=A7=8B=E5=8C=96=E6=A3=80?= =?UTF-8?q?=E6=B5=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 ChangeClassifier（爆炸半径分类：IMPLEMENTATION/INTERFACE/GLOBAL_CONSTRAINT） - 新增 ImpactPropagator（BFS影响传播：IMPACTED/BOUNDARY/SAFE差异化标记） - task_graph.py：invalidate_by_adr()差异化失效 + CascadeReport扩展字段（向后兼容） - eng_mode.py：三阶段差异化流程（分类→传播→失效→git操作→验证任务→重规划） - eng_mode.py：_git_squash_merge_and_tag() + Phase 7 集成 - do_mode.py：_ensure_all_committed() Worker git操作强制 - adr_watcher.py：内容快照 + get_content_for_classification() - events.py：ADR_CLASSIFIED/IMPACT_PROPAGATED/BOUNDARY_VERIFICATION_GENERATED - partial_replanner.py：replan_with_constraints() + generate_verification_tasks() - project_bootstrap.py：ensure_git_initialized() Arc规划前检测 - test_p1_21_phase2.py：21个新测试，95个全量测试0失败 Co-Authored-By: Claude Opus 4.7 --- lib/air_runtime/adr_watcher.py | 53 ++- lib/air_runtime/change_classifier.py | 101 +++++ lib/air_runtime/events.py | 3 + lib/air_runtime/impact_propagator.py | 104 +++++ lib/air_runtime/modes/do_mode.py | 27 ++ lib/air_runtime/modes/eng_mode.py | 340 +++++++++++++--- lib/air_runtime/partial_replanner.py | 56 +++ lib/air_runtime/project_bootstrap.py | 65 ++++ lib/air_runtime/task_graph.py | 121 +++++- test_p1_21_phase2.py | 559 +++++++++++++++++++++++++++ test_p1_21_supplement.py | 4 +- 11 files changed, 1351 insertions(+), 82 deletions(-) create mode 100644 lib/air_runtime/change_classifier.py create mode 100644 lib/air_runtime/impact_propagator.py create mode 100644 test_p1_21_phase2.py diff --git a/lib/air_runtime/adr_watcher.py b/lib/air_runtime/adr_watcher.py index 9b6061e..ce25885 100644 --- a/lib/air_runtime/adr_watcher.py +++ b/lib/air_runtime/adr_watcher.py @@ -15,10 +15,11 @@ from pathlib import Path class ADRChange: """ADR 文件变更记录。""" adr_id: str - kind: str # "new" | "superseded" | "modified" + 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: @@ -31,16 +32,17 @@ class ADRWatcher: 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。""" + """启动时记录所有 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() + 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，返回变更列表。""" @@ -57,35 +59,56 @@ class ADRWatcher: 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) - 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, - )) + 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'。""" diff --git a/lib/air_runtime/change_classifier.py b/lib/air_runtime/change_classifier.py new file mode 100644 index 0000000..4ff38b6 --- /dev/null +++ b/lib/air_runtime/change_classifier.py @@ -0,0 +1,101 @@ +""" +ADR 变更爆炸半径分类 — P1-21 / 3.2.17 Phase 1 组件。 +将 ADR 内容变更按影响范围分为 IMPLEMENTATION / INTERFACE / GLOBAL_CONSTRAINT 三级。 +""" + +from __future__ import annotations + +import enum +from dataclasses import dataclass, field +from pathlib import Path + + +class BlastRadius(enum.Enum): + IMPLEMENTATION = "implementation" + INTERFACE = "interface" + GLOBAL_CONSTRAINT = "global_constraint" + + +@dataclass +class ChangeClassification: + adr_id: str + blast_radius: BlastRadius + reason: str = "" + changed_sections: list[str] = field(default_factory=list) + + +class ChangeClassifier: + """基于关键词启发式的 ADR 变更爆炸半径分类。 + + 分类规则： + - IMPLEMENTATION: 仅实现细节变更，不影响公开接口。关键词：implementation detail, can use, for example, internal, private + - INTERFACE: 公开 API/合约变更。关键词：interface, api, contract, signature, must implement, public header, exported, caller + - GLOBAL_CONSTRAINT: 跨模块约束变更。关键词：must, shall, all modules, coding standard, every module, entire system, global policy + + 回退策略：内容不足无法确定时回退到 INTERFACE（保守：宁可多失效不少失效）。 + """ + + _INTERFACE_KEYWORDS = [ + "interface", "api", "contract", "signature", "must implement", + "public header", "exported", "caller", + ] + _GLOBAL_KEYWORDS = [ + "must", "shall", "all modules", "coding standard", + "every module", "entire system", "global policy", + ] + _IMPLEMENTATION_KEYWORDS = [ + "implementation detail", "can use", "for example", + "internally", "private", "optional", + ] + + def classify(self, adr_id: str, old_content: str | None, + new_content: str | None) -> ChangeClassification: + if new_content is None: + return ChangeClassification( + adr_id=adr_id, blast_radius=BlastRadius.GLOBAL_CONSTRAINT, + reason="ADR deleted — all dependent tasks must be invalidated", + ) + if old_content is None: + return ChangeClassification( + adr_id=adr_id, blast_radius=BlastRadius.IMPLEMENTATION, + reason="new ADR — no existing tasks to invalidate", + ) + + text = new_content.lower() + + global_hits = sum(1 for kw in self._GLOBAL_KEYWORDS if kw in text) + interface_hits = sum(1 for kw in self._INTERFACE_KEYWORDS if kw in text) + impl_hits = sum(1 for kw in self._IMPLEMENTATION_KEYWORDS if kw in text) + + if global_hits >= 2: + return ChangeClassification( + adr_id=adr_id, blast_radius=BlastRadius.GLOBAL_CONSTRAINT, + reason=f"global constraint keywords matched ({global_hits} hits)", + ) + if interface_hits >= 2: + return ChangeClassification( + adr_id=adr_id, blast_radius=BlastRadius.INTERFACE, + reason=f"interface keywords matched ({interface_hits} hits)", + ) + if interface_hits >= 1: + return ChangeClassification( + adr_id=adr_id, blast_radius=BlastRadius.INTERFACE, + reason="interface keyword matched — conservative classification", + ) + if impl_hits >= interface_hits and impl_hits > 0: + return ChangeClassification( + adr_id=adr_id, blast_radius=BlastRadius.IMPLEMENTATION, + reason=f"implementation keywords matched ({impl_hits} hits)", + ) + + # 无法确定 → 保守回退 + return ChangeClassification( + adr_id=adr_id, blast_radius=BlastRadius.INTERFACE, + reason="inconclusive — defaulting to INTERFACE (conservative)", + ) + + def classify_from_files(self, adr_id: str, old_path: Path | None, + new_path: Path | None) -> ChangeClassification: + old_content = old_path.read_text(encoding="utf-8") if old_path and old_path.exists() else None + new_content = new_path.read_text(encoding="utf-8") if new_path and new_path.exists() else None + return self.classify(adr_id, old_content, new_content) diff --git a/lib/air_runtime/events.py b/lib/air_runtime/events.py index b47baf7..652c0ba 100755 --- a/lib/air_runtime/events.py +++ b/lib/air_runtime/events.py @@ -41,7 +41,10 @@ WORKER_TIMEOUT = "worker.timeout" WORKTREE_MERGE_CONFLICT = "worktree.merge.conflict" ARC_REPLANNED = "arc.replanned" ADR_CHANGE_DETECTED = "adr.change.detected" +ADR_CLASSIFIED = "adr.classified" ADR_INVALIDATION = "adr.invalidation" +IMPACT_PROPAGATED = "impact.propagated" +BOUNDARY_VERIFICATION_GENERATED = "boundary.verification.generated" ADR_UNFREEZED = "adr.unfreezed" LOCK_ACQUIRED = "lock.acquired" LOCK_RELEASED = "lock.released" diff --git a/lib/air_runtime/impact_propagator.py b/lib/air_runtime/impact_propagator.py new file mode 100644 index 0000000..1e1debb --- /dev/null +++ b/lib/air_runtime/impact_propagator.py @@ -0,0 +1,104 @@ +""" +BFS 影响传播 — P1-21 / 3.2.17 Phase 1 组件。 +根据 BlastRadius 在任务 DAG 中传播影响，标记每个任务为 IMPACTED / BOUNDARY / SAFE。 +""" + +from __future__ import annotations + +import enum +from collections import deque +from dataclasses import dataclass, field + +from air_runtime.change_classifier import BlastRadius +from air_runtime.task_graph import TaskGraph + + +class ImpactLabel(enum.Enum): + IMPACTED = "impacted" + BOUNDARY = "boundary" + SAFE = "safe" + + +@dataclass +class PropagationResult: + impacted: list[str] = field(default_factory=list) + boundary: list[str] = field(default_factory=list) + safe: list[str] = field(default_factory=list) + + +class ImpactPropagator: + """BFS 影响传播，基于 BlastRadius 差异化标记。 + + 传播规则： + - IMPLEMENTATION: 仅直接引用 ADR 的任务 → IMPACTED，其他 → SAFE（不下游传播） + - INTERFACE: 直接引用 → IMPACTED，下游 1 跳邻居 → BOUNDARY，其他 → SAFE + - GLOBAL_CONSTRAINT: 全图 → IMPACTED + + 结果存储在 TaskNode.meta["impact_label"]，不修改 status 字段（正交）。 + """ + + _META_KEY = "impact_label" + + def propagate(self, graph: TaskGraph, adr_id: str, + blast_radius: BlastRadius) -> PropagationResult: + self.clear_labels(graph) + + tasks_ref_adr = [n.id for n in graph.nodes.values() if adr_id in n.adr_refs] + all_ids = set(graph.nodes.keys()) + + if blast_radius == BlastRadius.GLOBAL_CONSTRAINT: + for nid in all_ids: + graph.nodes[nid].meta[self._META_KEY] = ImpactLabel.IMPACTED.value + return PropagationResult( + impacted=list(all_ids), boundary=[], safe=[], + ) + + if blast_radius == BlastRadius.IMPLEMENTATION: + impacted = list(tasks_ref_adr) + safe = [nid for nid in all_ids if nid not in impacted] + for nid in impacted: + graph.nodes[nid].meta[self._META_KEY] = ImpactLabel.IMPACTED.value + for nid in safe: + graph.nodes[nid].meta[self._META_KEY] = ImpactLabel.SAFE.value + return PropagationResult(impacted=impacted, boundary=[], safe=safe) + + # INTERFACE: direct refs → IMPACTED, 1-hop downstream → BOUNDARY, rest → SAFE + impacted = set(tasks_ref_adr) + + # BFS 收集下游 1 跳邻居（仅从 IMPACTED 出发，排除自身） + boundary = set() + for nid in list(impacted): + node = graph.nodes.get(nid) + if node: + for target in node.out_edges: + if target not in impacted: + boundary.add(target) + + safe = all_ids - impacted - boundary + + for nid in impacted: + graph.nodes[nid].meta[self._META_KEY] = ImpactLabel.IMPACTED.value + for nid in boundary: + graph.nodes[nid].meta[self._META_KEY] = ImpactLabel.BOUNDARY.value + for nid in safe: + graph.nodes[nid].meta[self._META_KEY] = ImpactLabel.SAFE.value + + return PropagationResult( + impacted=list(impacted), boundary=list(boundary), safe=list(safe), + ) + + def impacted_tasks(self, graph: TaskGraph) -> list[str]: + return [nid for nid, n in graph.nodes.items() + if n.meta.get(self._META_KEY) == ImpactLabel.IMPACTED.value] + + def boundary_tasks(self, graph: TaskGraph) -> list[str]: + return [nid for nid, n in graph.nodes.items() + if n.meta.get(self._META_KEY) == ImpactLabel.BOUNDARY.value] + + def safe_tasks(self, graph: TaskGraph) -> list[str]: + return [nid for nid, n in graph.nodes.items() + if n.meta.get(self._META_KEY) == ImpactLabel.SAFE.value] + + def clear_labels(self, graph: TaskGraph) -> None: + for node in graph.nodes.values(): + node.meta.pop(self._META_KEY, None) diff --git a/lib/air_runtime/modes/do_mode.py b/lib/air_runtime/modes/do_mode.py index f6a60aa..c7d388f 100755 --- a/lib/air_runtime/modes/do_mode.py +++ b/lib/air_runtime/modes/do_mode.py @@ -17,6 +17,30 @@ from air_runtime.contracts import WorkerResult, now_iso from air_runtime.utils import sanitize_task_id, session_stamp +def _ensure_all_committed(project_root: Path, task_id: str) -> None: + """3.2.9a: Worker git 操作强制 — 检查所有变更已提交。 + + 在 Worker finish 前强制检查 git status，未提交的变更阻止完成。 + 确保 Worktree 分支上所有修改都已 git add + git commit。 + """ + import subprocess + try: + r = subprocess.run( + ["git", "status", "--porcelain"], + cwd=project_root, capture_output=True, text=True, timeout=10, + ) + if r.returncode != 0: + return # git not available, skip check + if r.stdout.strip(): + raise RuntimeError( + f"Worker {task_id}: 存在未提交的变更，禁止 finish。\n" + f"未提交文件:\n{r.stdout.strip()}\n" + f"请先 git add && git commit 后再 finish。" + ) + except FileNotFoundError: + pass # git not available, skip check + + # GUI 任务检测关键词 GUI_INDICATORS = { "gui", "ui", "render", "layout", "dialog", "osd", @@ -204,6 +228,9 @@ def finish_worker(project_root: Path, task_id: str, result_path: Path | None = N tid = sanitize_task_id(task_id) paths = _paths(project_root, tid) + # 3.2.9a: Worker git 操作强制 — 所有变更必须已提交 + _ensure_all_committed(project_root, tid) + # 加载 result if result_path and result_path.exists(): result_data = safe_json_load(result_path) diff --git a/lib/air_runtime/modes/eng_mode.py b/lib/air_runtime/modes/eng_mode.py index ef22d4d..54ea4b6 100755 --- a/lib/air_runtime/modes/eng_mode.py +++ b/lib/air_runtime/modes/eng_mode.py @@ -20,7 +20,8 @@ from air_runtime.paths import ( from air_runtime.events import EventLog, TASK_DISPATCHED, TASK_COMPLETED, TASK_BLOCKED, MERGE_STARTED, MERGE_COMPLETED, \ INTERVENTION_STALL, ENGINE_CYCLE, ENGINE_ENTERED, ENG_REPLAN_TRIGGERED, ENG_BLOCKED, \ WORKER_TIMEOUT, REPAIR_CREATED, REPAIR_RESOLVED, \ - ADR_CHANGE_DETECTED, ADR_INVALIDATION, ADR_UNFREEZED, WORKTREE_MERGE_CONFLICT + ADR_CHANGE_DETECTED, ADR_CLASSIFIED, ADR_INVALIDATION, IMPACT_PROPAGATED, \ + BOUNDARY_VERIFICATION_GENERATED, ADR_UNFREEZED, WORKTREE_MERGE_CONFLICT from air_runtime.evidence_gate import EvidenceGatePolicy, EvidenceClass from air_runtime.modes.merge_pipeline import ( apply_document_updates, @@ -310,30 +311,37 @@ 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 文件变更，返回需要级联失效的变更列表。""" +def _detect_adr_changes(project_root: Path, state: dict) -> tuple[list, object | None]: + """P1-21: 检查 ADR 文件变更，返回 (需要级联失效的变更列表, ADRWatcher 实例)。 + + 3.2.17: 同时持久化内容快照供 ChangeClassifier 爆炸半径分类使用。 + """ from air_runtime.adr_watcher import ADRWatcher, ADRChange adr_dir = project_root / "AirPlan" / "docs" / "architecture" / "adr" if not adr_dir.exists(): - return [] + return [], None watcher = ADRWatcher(adr_dir) - # 从引擎状态恢复已知 hash + # 从引擎状态恢复已知 hash 和内容快照 known = state.get("adrWatcherHashes", {}) - watcher._known_hashes = known + watcher._known_hashes = dict(known) + watcher._content_snapshots = dict(state.get("adrWatcherContentSnapshots", {})) # 首次无 snapshot → 先初始化 if not known: watcher.snapshot() state["adrWatcherHashes"] = dict(watcher._known_hashes) - return [] + state["adrWatcherContentSnapshots"] = dict(watcher._content_snapshots) + return [], watcher changes = watcher.detect_changes() - # 持久化更新后的 hash + # 持久化更新后的 hash 和内容快照 state["adrWatcherHashes"] = dict(watcher._known_hashes) + state["adrWatcherContentSnapshots"] = dict(watcher._content_snapshots) # 只返回需要级联失效的变更 - return [c for c in changes if c.kind in ("superseded", "modified")] + actionable = [c for c in changes if c.kind in ("superseded", "modified", "deleted")] + return actionable, watcher def monitor_engine(project_root: Path) -> dict: @@ -380,19 +388,28 @@ def monitor_engine(project_root: Path) -> dict: except OSError: pass - # P1-21: ADR 变更自动检测 - adr_changes = _detect_adr_changes(project_root, state) + # P1-21: ADR 变更自动检测（3.2.17 差异化流程） + adr_changes, adr_watcher = _detect_adr_changes(project_root, state) + adr_invalidation_results = [] if adr_changes: for change in adr_changes: - if change.kind in ("superseded", "modified"): + if change.kind in ("superseded", "modified", "deleted"): + log = EventLog(event_log_path(project_root)) + log.emit(ADR_CHANGE_DETECTED, { + "adrId": change.adr_id, "kind": change.kind, + }) + # 直接调用差异化失效处理 + inv_result = handle_adr_invalidation( + project_root, change.adr_id, + adr_change=change, watcher=adr_watcher, + ) + adr_invalidation_results.append(inv_result) 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, + "blastRadius": inv_result.get("blastRadius", ""), + "invalidatedCount": len(inv_result.get("cascadeReport", {}).get("invalidatedTaskIds", [])), }) # 新增：检查 pending worktree merges — merge 失败自动升级到 AirDbg @@ -467,6 +484,7 @@ def monitor_engine(project_root: Path) -> dict: "resourcePressure": resource_pressure, "worktreeMergeConflicts": [iv for iv in interventions if iv.get("reason", "").startswith("worktree-merge")], + "adrInvalidationResults": adr_invalidation_results, "nextAction": "monitor" if active_workers else "dispatch", } @@ -486,6 +504,8 @@ def merge_worker_result(project_root: Path, result_path: Path) -> dict: log.emit(MERGE_STARTED, {"taskId": preview_tid, "resultPath": str(result_path)}) + squash_result = {} + with state_lock: # Phase 1: 验证（含 doc sync 强制） result = safe_json_load(result_path) @@ -570,6 +590,15 @@ def merge_worker_result(project_root: Path, result_path: Path) -> dict: graph.nodes[task_id].status = new_status _export_task_graph_json(graph, tg_json) + # Phase 7: git squash merge + tag（3.2.5） + squash_result = {} + if status == "done": + wt_path = project_root / ".git" / "worktrees" / f"air-{task_id}" + squash_result = _git_squash_merge_and_tag( + project_root, task_id, + worktree_path=wt_path if wt_path.exists() else None, + ) + log.emit(MERGE_COMPLETED, { "taskId": task_id, "status": status, @@ -595,6 +624,7 @@ def merge_worker_result(project_root: Path, result_path: Path) -> dict: "archivedResultPath": str(archive_path), "appliedDocs": [str(p) for p in applied], "syncedDocs": [str(p) for p in sync_paths], + "squashResult": squash_result, "nextAction": "monitor" if state.get("activeWorkers") else "dispatch", } @@ -672,20 +702,15 @@ def spawn_workers(project_root: Path, task_ids: list[str]) -> list[dict]: return instructions -def handle_adr_invalidation(project_root: Path, adr_id: str) -> dict: - """P1-21: ADR 变更级联失效处理。 +def handle_adr_invalidation(project_root: Path, adr_id: str, + adr_change=None, watcher=None) -> dict: + """P1-21 / 3.2.17: ADR 变更级联失效处理（三阶段差异化流程）。 - 10步流程： - 1. 加载 task-graph.json - 2. 调用 invalidate_by_adr() 级联失效 - 3. 冻结调度 - 4. 中止进行中的相关 Worker - 5. 创建回滚快照（git tag） - 6. git revert 已合并的旧代码 - 7. 写回更新后的 task-graph.json - 8. 等待 Arc 重新生成受影响部分的任务 - 9. apply_delta() 吸收新任务 - 10. 解冻调度 + Phase 1: ChangeClassifier 爆炸半径分类 → ImpactPropagator BFS 传播 → 差异化失效 + Phase 2: Git 操作（revert IMPACTED+DONE, 清理 worktree, BOUNDARY 验证任务生成） + Phase 3: 提取 SAFE 接口约束 → 局部重规划 → 解冻 + + 当 adr_change/watcher 为 None 时回退到旧统一失效行为（向后兼容）。 """ tg_json = airplan_root(project_root) / "state" / "airarc" / "reviews" / "task-graph.json" if not tg_json.exists(): @@ -693,75 +718,203 @@ def handle_adr_invalidation(project_root: Path, adr_id: str) -> dict: graph = TaskGraph.load(tg_json) delta = PlanDelta() - - # 2-4: 级联失效 - report = graph.invalidate_by_adr(adr_id, delta) log = EventLog(event_log_path(project_root)) + + # ── Phase 1: 分类 → 传播 → 失效 ── + blast_radius = "" + boundary_task_ids: list[str] = [] + impact_labels: dict[str, str] | None = None + + if adr_change is not None and watcher is not None: + # 3.2.17 差异化流程 + from air_runtime.change_classifier import ChangeClassifier + from air_runtime.impact_propagator import ImpactPropagator, ImpactLabel + + old_content, new_content = watcher.get_content_for_classification(adr_change) + classifier = ChangeClassifier() + classification = classifier.classify(adr_change.adr_id, old_content, new_content) + blast_radius = classification.blast_radius.value + log.emit(ADR_CLASSIFIED, { + "adrId": adr_change.adr_id, + "blastRadius": blast_radius, + "reason": classification.reason, + }) + + propagator = ImpactPropagator() + propagation_result = propagator.propagate(graph, adr_change.adr_id, classification.blast_radius) + log.emit(IMPACT_PROPAGATED, { + "adrId": adr_change.adr_id, + "blastRadius": blast_radius, + "impactedCount": len(propagation_result.impacted), + "boundaryCount": len(propagation_result.boundary), + "safeCount": len(propagation_result.safe), + }) + + impact_labels = {} + for nid in propagation_result.impacted: + impact_labels[nid] = ImpactLabel.IMPACTED.value + for nid in propagation_result.boundary: + impact_labels[nid] = ImpactLabel.BOUNDARY.value + for nid in propagation_result.safe: + impact_labels[nid] = ImpactLabel.SAFE.value + + # 级联失效 + report = graph.invalidate_by_adr(adr_id, delta, impact_labels=impact_labels) + report.blast_radius = blast_radius log.emit(ADR_INVALIDATION, { "adrId": adr_id, + "blastRadius": blast_radius or "uniform", "invalidatedCompleted": report.invalidated_completed, "terminatedInProgress": report.terminated_in_progress, "cascadedDownstream": report.cascaded_downstream, + "boundaryTaskCount": len(report.boundary_task_ids), }) - # 4: 中止进行中的相关 Worker + # ── Phase 1.5: 终止 IMPACTED Worker ── paths = _paths(project_root) _ensure_dirs(paths) state = safe_json_load(paths["state"]) or _init_state(project_root) terminated_workers = [] + impacted_ids = set(report.invalidated_task_ids) for worker in list(state.get("activeWorkers", [])): - if worker.get("taskId") in report.invalidated_task_ids: + if worker.get("taskId") in impacted_ids: terminated_workers.append(worker["taskId"]) state["activeWorkers"] = [ w for w in state.get("activeWorkers", []) - if w.get("taskId") not in report.invalidated_task_ids + if w.get("taskId") not in impacted_ids ] - # 5: 创建回滚快照 - rollback_ref = _create_rollback_snapshot(project_root, report.invalidated_task_ids) + # ── Phase 2: Git 操作 ── + # 2a: 创建回滚快照 + rollback_ref = _create_rollback_snapshot(project_root, list(impacted_ids)) report.rollback_ref = rollback_ref delta.rollback_ref = rollback_ref - # 6: git revert 已合并的旧代码（按 task_id 查找对应 commit） - revert_results = _git_revert_invalidated(project_root, report.invalidated_task_ids) + # 2b: git revert IMPACTED+DONE 任务的 commit + revert_results = _git_revert_invalidated(project_root, list(impacted_ids)) - # 6.5: 生成局部重规划请求（PartialReplanner） + # 2c: 清理 IMPACTED+DISPATCHED 的 worktree + worktree_cleanup_results = _cleanup_impacted_worktrees(project_root, list(impacted_ids)) + + # 2d: 为 BOUNDARY+DONE 任务生成验证任务 + verification_delta = PlanDelta() + if report.boundary_task_ids: + from air_runtime.partial_replanner import PartialReplanner + replanner = PartialReplanner() + verification_delta = replanner.generate_verification_tasks(graph, report.boundary_task_ids) + if verification_delta.added_tasks: + log.emit(BOUNDARY_VERIFICATION_GENERATED, { + "adrId": adr_id, + "boundaryTaskIds": report.boundary_task_ids, + "verificationTaskIds": [n.id for n in verification_delta.added_tasks], + }) + + # ── Phase 3: 局部重规划 ── from air_runtime.partial_replanner import PartialReplanner replanner = PartialReplanner() - partial_delta = replanner.replan(graph, report.invalidated_task_ids) + + # 3a: 提取 SAFE+DONE 任务的接口约束 + safe_interfaces = _extract_safe_interfaces(graph) + + # 3b: 带冻结接口约束的局部重规划 + invalidated_ids = report.invalidated_task_ids + if safe_interfaces: + partial_delta = replanner.replan_with_constraints( + graph, invalidated_ids, safe_interfaces, + ) + else: + partial_delta = replanner.replan(graph, invalidated_ids) + + # 3c: 合并验证任务到重规划结果 + for vn in verification_delta.added_tasks: + partial_delta.added_tasks.append(vn) + for ve in verification_delta.edge_changes.added: + partial_delta.edge_changes.added.append(ve) + + # 3d: 应用 delta + graph.apply_delta(partial_delta) + + # 保存局部重规划请求 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 + # ── 写回 task-graph.json ── from air_runtime.modes.arc_mode import _export_task_graph_json _export_task_graph_json(graph, tg_json) - # 更新引擎状态 + # ── 更新引擎状态 ── state["dispatchFrozen"] = True state["adrInvalidationInProgress"] = { "adrId": adr_id, "startedAt": now_iso(), - "invalidatedTaskIds": report.invalidated_task_ids, + "blastRadius": blast_radius, + "invalidatedTaskIds": invalidated_ids, + "boundaryTaskIds": report.boundary_task_ids, "rollbackRef": rollback_ref, } atomic_json_write(paths["state"], state) return { "adrId": adr_id, + "blastRadius": blast_radius or "uniform", "cascadeReport": { "invalidatedCompleted": report.invalidated_completed, "terminatedInProgress": report.terminated_in_progress, "cascadedDownstream": report.cascaded_downstream, "rollbackRef": rollback_ref, - "invalidatedTaskIds": report.invalidated_task_ids, + "invalidatedTaskIds": invalidated_ids, + "boundaryTaskIds": report.boundary_task_ids, }, "terminatedWorkers": terminated_workers, "revertResults": revert_results, + "worktreeCleanupResults": worktree_cleanup_results, + "verificationTaskCount": len(verification_delta.added_tasks), + "safeInterfaceCount": len(safe_interfaces), "replanRequestPath": str(replan_request_path), "nextStep": "arc-replan-then-unfreeze", } +def _extract_safe_interfaces(graph: TaskGraph) -> list: + """3.2.17 Phase 3: 提取 SAFE+DONE 任务的接口约束，供重规划使用。""" + from air_runtime.partial_replanner import Interface + from air_runtime.impact_propagator import ImpactLabel + + interfaces = [] + for nid, node in graph.nodes.items(): + label = node.meta.get("impact_label", "") + if label == ImpactLabel.SAFE.value 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 + + +def _cleanup_impacted_worktrees(project_root: Path, impacted_ids: list[str]) -> list[dict]: + """3.2.17 Phase 2c: 清理 IMPACTED+DISPATCHED 任务的 worktree。""" + import subprocess + results = [] + wt_root = project_root / ".git" / "worktrees" + if not wt_root.exists(): + return results + + for tid in impacted_ids: + wt_dir = wt_root / f"air-{tid}" + if wt_dir.exists(): + try: + subprocess.run( + ["git", "worktree", "remove", str(wt_dir), "--force"], + cwd=project_root, capture_output=True, text=True, timeout=30, + ) + results.append({"taskId": tid, "worktree": str(wt_dir), "removed": True}) + except Exception as e: + results.append({"taskId": tid, "worktree": str(wt_dir), "removed": False, "error": str(e)}) + return results + + def _create_rollback_snapshot(project_root: Path, invalidated_task_ids: list[str]) -> str: """P1-21: 为失效任务创建 git tag 回滚点。""" import subprocess @@ -777,28 +930,43 @@ def _create_rollback_snapshot(project_root: Path, invalidated_task_ids: list[str def _git_revert_invalidated(project_root: Path, invalidated_task_ids: list[str]) -> list[dict]: - """P1-21: 尝试 git revert 已合并的失效任务对应的 commit。""" + """P1-21: 尝试 git revert 已合并的失效任务对应的 commit。 + + 优先使用 air/done/{task_id} tag 定位 commit，回退到 commit message 搜索。 + """ import subprocess results = [] for tid in invalidated_task_ids: try: - # 查找包含 task_id 的 commit - r = subprocess.run( - ["git", "log", "--oneline", "--all", "--grep", tid, "-1"], + commit_hash = None + # 优先尝试 air/done/{task_id} tag + tag_name = f"air/done/{tid}" + tag_check = subprocess.run( + ["git", "rev-parse", "--verify", f"refs/tags/{tag_name}"], cwd=project_root, capture_output=True, text=True, timeout=10, ) - if r.returncode == 0 and r.stdout.strip(): - commit_hash = r.stdout.strip().split()[0] + if tag_check.returncode == 0: + commit_hash = tag_check.stdout.strip() + else: + # 回退到 commit message 搜索 + r = subprocess.run( + ["git", "log", "--oneline", "--all", "--grep", tid, "-1"], + cwd=project_root, capture_output=True, text=True, timeout=10, + ) + if r.returncode == 0 and r.stdout.strip(): + commit_hash = r.stdout.strip().split()[0] + + if commit_hash: rv = subprocess.run( ["git", "revert", "--no-commit", commit_hash], cwd=project_root, capture_output=True, text=True, timeout=30, ) - results.append({"taskId": tid, "commit": commit_hash, "reverted": rv.returncode == 0}) if rv.returncode == 0: subprocess.run( ["git", "commit", "-m", f"AirPlan: revert invalidated task {tid}"], cwd=project_root, capture_output=True, timeout=10, ) + results.append({"taskId": tid, "commit": commit_hash, "reverted": rv.returncode == 0}) else: results.append({"taskId": tid, "commit": None, "reverted": False, "reason": "no commit found"}) except Exception as e: @@ -806,6 +974,74 @@ def _git_revert_invalidated(project_root: Path, invalidated_task_ids: list[str]) return results +def _git_squash_merge_and_tag(project_root: Path, task_id: str, + worktree_path: Path | None = None) -> dict: + """3.2.5 Phase 7: git squash merge + tag。 + + 步骤： + 1. git merge --squash + 2. git commit -m "AirPlan: done {task_id}" + 3. git tag air/done/{task_id} + 4. git worktree remove + 5. git rev-parse --verify refs/tags/air/done/{task_id} 验证 + """ + import subprocess + tag_name = f"air/done/{task_id}" + result = {"taskId": task_id, "squashed": False, "tagged": False, "worktreeCleaned": False} + + try: + # 1. 尝试 squash merge worktree 分支到当前分支 + branch_name = f"air/do/{task_id}" + # 检查 worktree 分支是否存在 + branch_check = subprocess.run( + ["git", "rev-parse", "--verify", branch_name], + cwd=project_root, capture_output=True, text=True, timeout=10, + ) + if branch_check.returncode == 0: + # 分支存在，执行 squash merge + mr = subprocess.run( + ["git", "merge", "--squash", branch_name], + cwd=project_root, capture_output=True, text=True, timeout=30, + ) + if mr.returncode == 0: + # 2. Commit + subprocess.run( + ["git", "commit", "-m", f"AirPlan: done {task_id}"], + cwd=project_root, capture_output=True, timeout=10, + ) + result["squashed"] = True + else: + # 无独立分支，尝试直接基于现有 HEAD commit 打 tag + result["squashed"] = True # 假设已在主分支上提交 + + # 3. Tag + subprocess.run( + ["git", "tag", "-f", tag_name, "HEAD"], + cwd=project_root, capture_output=True, timeout=10, + ) + result["tagged"] = True + + # 4. 清理 worktree + if worktree_path and worktree_path.exists(): + subprocess.run( + ["git", "worktree", "remove", str(worktree_path), "--force"], + cwd=project_root, capture_output=True, text=True, timeout=30, + ) + result["worktreeCleaned"] = True + + # 5. 验证 tag + verify = subprocess.run( + ["git", "rev-parse", "--verify", f"refs/tags/{tag_name}"], + cwd=project_root, capture_output=True, text=True, timeout=10, + ) + result["tagVerified"] = verify.returncode == 0 + + except Exception as e: + result["error"] = str(e) + + return result + + def unfreeze_after_replan(project_root: Path, new_task_graph_path: Path | None = None) -> dict: """P1-21: Arc 重新生成受影响部分后，apply_delta + 解冻调度。""" paths = _paths(project_root) diff --git a/lib/air_runtime/partial_replanner.py b/lib/air_runtime/partial_replanner.py index 7f8b01f..ef920f2 100644 --- a/lib/air_runtime/partial_replanner.py +++ b/lib/air_runtime/partial_replanner.py @@ -73,6 +73,62 @@ class PartialReplanner: return delta + def replan_with_constraints(self, graph: TaskGraph, + invalidated_ids: list[str], + frozen_interfaces: list[Interface], + new_adr_path: Path | None = None) -> PlanDelta: + """带冻结接口约束的局部重规划（3.2.17 Phase 3）。 + + frozen_interfaces 来自 SAFE 已完成任务，重规划不能破坏这些接口。 + """ + delta = PlanDelta() + + affected_context = self._collect_affected_context(graph, invalidated_ids) + stable_interfaces = self._extract_stable_interfaces(graph, set(invalidated_ids)) + + all_frozen = list(stable_interfaces) + for fi in frozen_interfaces: + if not any(s.task_id == fi.task_id for s in all_frozen): + all_frozen.append(fi) + + replan_request = { + "type": "partial-replan", + "invalidatedTaskIds": invalidated_ids, + "affectedContext": [ctx.__dict__ for ctx in affected_context], + "stableInterfaces": [iface.__dict__ for iface in all_frozen], + "newAdrPath": str(new_adr_path) if new_adr_path else None, + } + delta.replan_request = replan_request + return delta + + def generate_verification_tasks(self, graph: TaskGraph, + boundary_task_ids: list[str]) -> PlanDelta: + """为 BOUNDARY 已完成任务生成验证任务（3.2.17 Phase 3.4）。 + + 验证任务检查 BOUNDARY 任务的兼容性，确认是否与新 ADR 一致。 + """ + from air_runtime.task_graph import TaskNode, Edge + delta = PlanDelta() + for idx, tid in enumerate(boundary_task_ids): + node = graph.nodes.get(tid) + if not node or node.status != "DONE": + continue + verify_id = f"{tid}-VFY" + verify_node = TaskNode( + id=verify_id, + status="TODO", + task=f"[边界验证] 验证 {tid} ({node.task}) 与新 ADR 的兼容性", + files_dirs=node.files_dirs, + done_when="兼容性确认通过", + write_set=list(node.write_set), + test_required=True, + ) + delta.added_tasks.append(verify_node) + delta.edge_changes.added.append(Edge( + source=tid, target=verify_id, kind="verification", + )) + return delta + def _collect_affected_context(self, graph: TaskGraph, invalidated_ids: list[str]) -> list[ReplanContext]: """收集受影响任务的上下文。""" diff --git a/lib/air_runtime/project_bootstrap.py b/lib/air_runtime/project_bootstrap.py index 1e31d45..d264ae0 100755 --- a/lib/air_runtime/project_bootstrap.py +++ b/lib/air_runtime/project_bootstrap.py @@ -5,9 +5,74 @@ V2 保持与 V1 相同的不变量：制品驱动通信、上下文隔离。 from __future__ import annotations +import subprocess from pathlib import Path +def ensure_git_initialized(project_root: Path) -> dict: + """3.2.5: Arc 规划前 git 初始化检测。 + + 检测项目是否为 git 仓库，至少有 1 个 commit。 + 如果没有则自动初始化并创建空提交，确保 worktree 操作可行。 + """ + git_dir = project_root / ".git" + result = {"git": False, "commits": False, "initialized": False} + + if git_dir.exists(): + result["git"] = True + try: + r = subprocess.run( + ["git", "rev-parse", "HEAD"], + cwd=project_root, capture_output=True, text=True, timeout=10, + ) + if r.returncode == 0: + result["commits"] = True + return result + except Exception: + pass + + # 初始化 git + if not result["git"]: + try: + subprocess.run( + ["git", "init"], + cwd=project_root, capture_output=True, timeout=10, + ) + result["git"] = True + except Exception as e: + result["error"] = f"git init failed: {e}" + return result + + # 创建初始提交 + try: + subprocess.run( + ["git", "config", "user.email", "airplan@local"], + cwd=project_root, capture_output=True, timeout=5, + ) + subprocess.run( + ["git", "config", "user.name", "AirPlan"], + cwd=project_root, capture_output=True, timeout=5, + ) + # 创建 .gitkeep 确保有文件可提交 + gitkeep = project_root / ".gitkeep" + if not gitkeep.exists(): + gitkeep.touch() + subprocess.run( + ["git", "add", ".gitkeep"], + cwd=project_root, capture_output=True, timeout=10, + ) + subprocess.run( + ["git", "commit", "-m", "AirPlan: initial empty commit"], + cwd=project_root, capture_output=True, timeout=10, + ) + result["commits"] = True + result["initialized"] = True + except Exception as e: + result["error"] = f"initial commit failed: {e}" + + return result + + def ensure_project_bootstrap(project_root: Path) -> dict[str, bool]: """创建 AirPlan 必需目录结构。""" root = project_root / "AirPlan" diff --git a/lib/air_runtime/task_graph.py b/lib/air_runtime/task_graph.py index 3ae4537..e6bc303 100755 --- a/lib/air_runtime/task_graph.py +++ b/lib/air_runtime/task_graph.py @@ -40,12 +40,14 @@ class EdgeChange: @dataclass class CascadeReport: - """P1-21: ADR 变更级联失效报告。""" + """P1-21: ADR 变更级联失效报告。3.2.17 差异化处理后新增 boundary/impact 字段。""" invalidated_completed: int = 0 terminated_in_progress: int = 0 cascaded_downstream: int = 0 rollback_ref: str = "" invalidated_task_ids: list[str] = field(default_factory=list) + boundary_task_ids: list[str] = field(default_factory=list) + blast_radius: str = "" # "implementation" | "interface" | "global_constraint" @dataclass @@ -266,32 +268,40 @@ class TaskGraph: # 排除起点自身 return [tid for tid in visited if tid not in set(task_ids)] - def invalidate_by_adr(self, adr_id: str, delta: PlanDelta) -> CascadeReport: - """P1-21: ADR 变更时级联失效所有相关任务。""" + def invalidate_by_adr(self, adr_id: str, delta: PlanDelta, + impact_labels: dict[str, str] | None = None) -> CascadeReport: + """P1-21: ADR 变更时级联失效相关任务。 + + 当 impact_labels 为 None 时使用旧行为（所有引用+下游统一 INVALIDATED）。 + 当 impact_labels 提供时使用差异化行为（IMPACTED→INVALIDATED, BOUNDARY+DONE→验证, SAFE→保留）。 + """ affected = self.tasks_by_adr(adr_id) - completed = [t for t in affected if t.status == "DONE"] - in_progress = [t for t in affected if t.status == "DISPATCHED"] - pending = [t for t in affected if t.status == "TODO"] # 1. 冻结调度 self.dispatch_frozen = True - # 2. 标记已完成任务为 INVALIDATED + if impact_labels is None: + return self._invalidate_uniform(adr_id, delta, affected) + + return self._invalidate_differentiated(adr_id, delta, affected, impact_labels) + + def _invalidate_uniform(self, adr_id: str, delta: PlanDelta, + affected: list[TaskNode]) -> CascadeReport: + """旧行为：所有引用 ADR 的任务及其下游统一 INVALIDATED。""" + completed = [t for t in affected if t.status == "DONE"] + in_progress = [t for t in affected if t.status == "DISPATCHED"] + pending = [t for t in affected if t.status == "TODO"] + for t in completed: t.status = "INVALIDATED" delta.removed_tasks.append(t.id) - - # 3. 标记进行中任务为 INVALIDATED（调用方负责中止 Worker） for t in in_progress: t.status = "INVALIDATED" delta.removed_tasks.append(t.id) - - # 4. 标记 ADR 直接关联的 TODO 任务为 INVALIDATED for t in pending: t.status = "INVALIDATED" delta.removed_tasks.append(t.id) - # 5. 级联失效下游 downstream_ids = self._find_downstream([t.id for t in completed + in_progress + pending]) cascaded = [] for tid in downstream_ids: @@ -301,7 +311,6 @@ class TaskGraph: delta.removed_tasks.append(tid) cascaded.append(tid) - # 6. 回滚快照引用（由调用方在 git revert 后填入） all_invalidated = [t.id for t in completed + in_progress + pending] + cascaded return CascadeReport( @@ -312,6 +321,92 @@ class TaskGraph: invalidated_task_ids=all_invalidated, ) + def _invalidate_differentiated(self, adr_id: str, delta: PlanDelta, + affected: list[TaskNode], + impact_labels: dict[str, str]) -> CascadeReport: + """差异化行为：按 IMPACTED/BOUNDARY/SAFE 分别处理每个任务。""" + invalidated_completed = 0 + terminated_in_progress = 0 + invalidated_ids: list[str] = [] + boundary_ids: list[str] = [] + + from air_runtime.impact_propagator import ImpactLabel + + for t in affected: + label = impact_labels.get(t.id, ImpactLabel.SAFE.value) + if label == ImpactLabel.SAFE.value: + continue + + if label == ImpactLabel.BOUNDARY.value: + if t.status == "DONE": + t.meta["verification_needed"] = True + boundary_ids.append(t.id) + elif t.status in ("TODO", "BLOCKED"): + t.status = "INVALIDATED" + delta.removed_tasks.append(t.id) + invalidated_ids.append(t.id) + # DISPATCHED + BOUNDARY: wait for completion, no action + + elif label == ImpactLabel.IMPACTED.value: + if t.status == "DONE": + t.status = "INVALIDATED" + delta.removed_tasks.append(t.id) + invalidated_ids.append(t.id) + invalidated_completed += 1 + elif t.status == "DISPATCHED": + t.status = "INVALIDATED" + delta.removed_tasks.append(t.id) + invalidated_ids.append(t.id) + terminated_in_progress += 1 + elif t.status in ("TODO", "BLOCKED"): + t.status = "INVALIDATED" + delta.removed_tasks.append(t.id) + invalidated_ids.append(t.id) + + # 级联下游：仅 IMPACTED 任务的下游受影响 + impacted_affected = [t for t in affected + if impact_labels.get(t.id) == ImpactLabel.IMPACTED.value] + cascaded_ids: list[str] = [] + if impacted_affected: + downstream_ids = self._find_downstream([t.id for t in impacted_affected]) + for tid in downstream_ids: + node = self.nodes.get(tid) + if not node: + continue + dlabel = impact_labels.get(tid, ImpactLabel.SAFE.value) + if dlabel == ImpactLabel.SAFE.value: + continue + if dlabel == ImpactLabel.BOUNDARY.value: + if node.status == "DONE": + node.meta["verification_needed"] = True + boundary_ids.append(tid) + elif node.status in ("TODO", "BLOCKED"): + node.status = "INVALIDATED" + delta.removed_tasks.append(tid) + cascaded_ids.append(tid) + invalidated_ids.append(tid) + elif dlabel == ImpactLabel.IMPACTED.value: + if node.status in ("TODO", "DISPATCHED"): + node.status = "INVALIDATED" + delta.removed_tasks.append(tid) + cascaded_ids.append(tid) + invalidated_ids.append(tid) + elif node.status == "DONE": + node.status = "INVALIDATED" + delta.removed_tasks.append(tid) + cascaded_ids.append(tid) + invalidated_ids.append(tid) + invalidated_completed += 1 + + return CascadeReport( + invalidated_completed=invalidated_completed, + terminated_in_progress=terminated_in_progress, + cascaded_downstream=len(cascaded_ids), + rollback_ref=delta.rollback_ref, + invalidated_task_ids=invalidated_ids, + boundary_task_ids=boundary_ids, + ) + def unfreeze_dispatch(self) -> None: """P1-21: 解冻调度，在 Arc 重新生成受影响任务后调用。""" self.dispatch_frozen = False diff --git a/test_p1_21_phase2.py b/test_p1_21_phase2.py new file mode 100644 index 0000000..54d98da --- /dev/null +++ b/test_p1_21_phase2.py @@ -0,0 +1,559 @@ +#!/usr/bin/env python3 +"""3.2.17 中途变更处理（git驱动）Phase 2 功能测试 +测试：ChangeClassifier + ImpactPropagator + 差异化失效 + BOUNDARY验证 + SAFE接口 +""" + +import sys +import tempfile +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent / "lib")) + + +def test_change_classifier_implementation(): + """ChangeClassifier: 实现细节变更 → IMPLEMENTATION""" + from air_runtime.change_classifier import ChangeClassifier, BlastRadius + + classifier = ChangeClassifier() + old = "使用 ffmpeg 实现视频解码。\n\nimplementation detail: 可以用硬件加速 internally。" + new = "使用 ffmpeg 实现视频解码。\n\n改用 gstreamer 硬件加速，optional 可用 private 实现。" + + result = classifier.classify("ADR-0005", old, new) + assert result.blast_radius == BlastRadius.IMPLEMENTATION, \ + f"应为 IMPLEMENTATION，实际: {result.blast_radius}" + assert result.adr_id == "ADR-0005" + print("✓ IMPLEMENTATION 分类测试通过") + return True + + +def test_change_classifier_interface(): + """ChangeClassifier: 接口变更 → INTERFACE""" + from air_runtime.change_classifier import ChangeClassifier, BlastRadius + + classifier = ChangeClassifier() + old = "使用 ffmpeg API 进行解码。" + new = "使用 gstreamer API 进行解码。must implement 新的 contract 接口。caller 需要更新。" + + result = classifier.classify("ADR-0005", old, new) + assert result.blast_radius == BlastRadius.INTERFACE, \ + f"应为 INTERFACE，实际: {result.blast_radius}" + print("✓ INTERFACE 分类测试通过") + return True + + +def test_change_classifier_global(): + """ChangeClassifier: 全局约束变更 → GLOBAL_CONSTRAINT""" + from air_runtime.change_classifier import ChangeClassifier, BlastRadius + + classifier = ChangeClassifier() + old = "使用 ffmpeg。" + new = "coding standard: must 使用 RAII。all modules shall 遵循新规范。every module 需要更新。" + + result = classifier.classify("ADR-0005", old, new) + assert result.blast_radius == BlastRadius.GLOBAL_CONSTRAINT, \ + f"应为 GLOBAL_CONSTRAINT，实际: {result.blast_radius}" + print("✓ GLOBAL_CONSTRAINT 分类测试通过") + return True + + +def test_change_classifier_new_adr(): + """ChangeClassifier: 新增 ADR → IMPLEMENTATION""" + from air_runtime.change_classifier import ChangeClassifier, BlastRadius + + classifier = ChangeClassifier() + result = classifier.classify("ADR-0020", None, "新的 ADR 内容。") + assert result.blast_radius == BlastRadius.IMPLEMENTATION + assert "new ADR" in result.reason + print("✓ 新增 ADR 分类测试通过") + return True + + +def test_change_classifier_deleted_adr(): + """ChangeClassifier: 删除 ADR → GLOBAL_CONSTRAINT""" + from air_runtime.change_classifier import ChangeClassifier, BlastRadius + + classifier = ChangeClassifier() + result = classifier.classify("ADR-0005", "旧的 ADR 内容。", None) + assert result.blast_radius == BlastRadius.GLOBAL_CONSTRAINT + assert "deleted" in result.reason.lower() + print("✓ 删除 ADR 分类测试通过") + return True + + +def test_change_classifier_conservative_fallback(): + """ChangeClassifier: 无法确定 → INTERFACE（保守）""" + from air_runtime.change_classifier import ChangeClassifier, BlastRadius + + classifier = ChangeClassifier() + result = classifier.classify("ADR-0005", "旧内容", "新内容") + assert result.blast_radius == BlastRadius.INTERFACE + assert "conservative" in result.reason.lower() + print("✓ 保守回退分类测试通过") + return True + + +def test_impact_propagator_implementation(): + """ImpactPropagator: IMPLEMENTATION → 仅直接引用受影响""" + from air_runtime.task_graph import TaskGraph, TaskNode, Edge + from air_runtime.change_classifier import BlastRadius + from air_runtime.impact_propagator import ImpactPropagator, ImpactLabel + + graph = TaskGraph() + graph.add_node(TaskNode(id="T-001", task="ffmpeg 解码器", adr_refs=["ADR-0005"])) + graph.add_node(TaskNode(id="T-002", task="集成测试")) + graph.add_node(TaskNode(id="T-003", task="UI", adr_refs=["ADR-0008"])) + graph.add_edge(Edge(source="T-001", target="T-002")) + + propagator = ImpactPropagator() + result = propagator.propagate(graph, "ADR-0005", BlastRadius.IMPLEMENTATION) + + assert "T-001" in result.impacted + assert "T-002" not in result.impacted # 不传播到下游 + assert "T-003" in result.safe + # 验证 meta 标记 + assert graph.nodes["T-001"].meta["impact_label"] == ImpactLabel.IMPACTED.value + assert graph.nodes["T-002"].meta["impact_label"] == ImpactLabel.SAFE.value + print("✓ IMPLEMENTATION 传播测试通过") + return True + + +def test_impact_propagator_interface(): + """ImpactPropagator: INTERFACE → 直接引用 IMPACTED + 下游1跳 BOUNDARY""" + from air_runtime.task_graph import TaskGraph, TaskNode, Edge + from air_runtime.change_classifier import BlastRadius + from air_runtime.impact_propagator import ImpactPropagator, ImpactLabel + + graph = TaskGraph() + graph.add_node(TaskNode(id="T-001", task="ffmpeg 解码器", adr_refs=["ADR-0005"])) + graph.add_node(TaskNode(id="T-002", task="集成测试")) # 下游1跳 + graph.add_node(TaskNode(id="T-003", task="部署")) # 下游2跳 + graph.add_node(TaskNode(id="T-004", task="UI", adr_refs=["ADR-0008"])) # 无关 + graph.add_edge(Edge(source="T-001", target="T-002")) + graph.add_edge(Edge(source="T-002", target="T-003")) + + propagator = ImpactPropagator() + result = propagator.propagate(graph, "ADR-0005", BlastRadius.INTERFACE) + + assert "T-001" in result.impacted + assert "T-002" in result.boundary, f"T-002 应为 BOUNDARY，实际 impacted={result.impacted}, boundary={result.boundary}" + assert "T-003" in result.safe, f"T-003 应为 SAFE（下游2跳），实际: {result.safe}" + assert "T-004" in result.safe + assert graph.nodes["T-001"].meta["impact_label"] == ImpactLabel.IMPACTED.value + assert graph.nodes["T-002"].meta["impact_label"] == ImpactLabel.BOUNDARY.value + assert graph.nodes["T-003"].meta["impact_label"] == ImpactLabel.SAFE.value + print("✓ INTERFACE 传播(含 BOUNDARY)测试通过") + return True + + +def test_impact_propagator_global(): + """ImpactPropagator: GLOBAL_CONSTRAINT → 全图 IMPACTED""" + from air_runtime.task_graph import TaskGraph, TaskNode + from air_runtime.change_classifier import BlastRadius + from air_runtime.impact_propagator import ImpactPropagator, ImpactLabel + + graph = TaskGraph() + graph.add_node(TaskNode(id="T-001", task="任务1", adr_refs=["ADR-0005"])) + graph.add_node(TaskNode(id="T-002", task="任务2")) + graph.add_node(TaskNode(id="T-003", task="任务3", adr_refs=["ADR-0008"])) + + propagator = ImpactPropagator() + result = propagator.propagate(graph, "ADR-0005", BlastRadius.GLOBAL_CONSTRAINT) + + assert len(result.impacted) == 3 + assert len(result.boundary) == 0 + assert len(result.safe) == 0 + for nid in ["T-001", "T-002", "T-003"]: + assert graph.nodes[nid].meta["impact_label"] == ImpactLabel.IMPACTED.value + print("✓ GLOBAL_CONSTRAINT 全图传播测试通过") + return True + + +def test_impact_propagator_clear_labels(): + """ImpactPropagator: clear_labels 清除标记""" + from air_runtime.task_graph import TaskGraph, TaskNode + from air_runtime.change_classifier import BlastRadius + from air_runtime.impact_propagator import ImpactPropagator + + graph = TaskGraph() + graph.add_node(TaskNode(id="T-001", task="任务1", adr_refs=["ADR-0005"])) + graph.add_node(TaskNode(id="T-002", task="任务2")) + + propagator = ImpactPropagator() + propagator.propagate(graph, "ADR-0005", BlastRadius.GLOBAL_CONSTRAINT) + assert "impact_label" in graph.nodes["T-001"].meta + + propagator.clear_labels(graph) + assert "impact_label" not in graph.nodes["T-001"].meta + assert "impact_label" not in graph.nodes["T-002"].meta + print("✓ clear_labels 测试通过") + return True + + +def test_differentiated_invalidation_impacted_done(): + """差异化失效: IMPACTED + DONE → INVALIDATED""" + from air_runtime.task_graph import TaskGraph, TaskNode, PlanDelta + from air_runtime.impact_propagator import ImpactLabel + + graph = TaskGraph() + graph.add_node(TaskNode(id="T-001", status="DONE", task="ffmpeg 解码器", adr_refs=["ADR-0005"])) + + delta = PlanDelta() + labels = {"T-001": ImpactLabel.IMPACTED.value} + report = graph.invalidate_by_adr("ADR-0005", delta, impact_labels=labels) + + assert report.invalidated_completed == 1 + assert graph.nodes["T-001"].status == "INVALIDATED" + assert "T-001" in delta.removed_tasks + print("✓ IMPACTED+DONE→INVALIDATED 测试通过") + return True + + +def test_differentiated_invalidation_boundary_done(): + """差异化失效: BOUNDARY + DONE → 保持 DONE，标记 verification_needed""" + from air_runtime.task_graph import TaskGraph, TaskNode, PlanDelta + from air_runtime.impact_propagator import ImpactLabel + + graph = TaskGraph() + graph.add_node(TaskNode(id="T-001", status="DONE", task="集成测试", adr_refs=["ADR-0005"])) + + delta = PlanDelta() + labels = {"T-001": ImpactLabel.BOUNDARY.value} + report = graph.invalidate_by_adr("ADR-0005", delta, impact_labels=labels) + + assert graph.nodes["T-001"].status == "DONE" + assert graph.nodes["T-001"].meta.get("verification_needed") == True + assert "T-001" in report.boundary_task_ids + assert "T-001" not in delta.removed_tasks + print("✓ BOUNDARY+DONE→verification_needed 测试通过") + return True + + +def test_differentiated_invalidation_safe_untouched(): + """差异化失效: SAFE + DONE → 完全不处理""" + from air_runtime.task_graph import TaskGraph, TaskNode, PlanDelta + from air_runtime.impact_propagator import ImpactLabel + + graph = TaskGraph() + graph.add_node(TaskNode(id="T-001", status="DONE", task="UI 界面", adr_refs=["ADR-0008"])) + + delta = PlanDelta() + labels = {"T-001": ImpactLabel.SAFE.value} + report = graph.invalidate_by_adr("ADR-0005", delta, impact_labels=labels) + + assert graph.nodes["T-001"].status == "DONE" + assert graph.nodes["T-001"].meta.get("verification_needed") != True + assert "T-001" not in delta.removed_tasks + assert report.invalidated_completed == 0 + print("✓ SAFE→不处理 测试通过") + return True + + +def test_differentiated_invalidation_impacted_dispatched(): + """差异化失效: IMPACTED + DISPATCHED → INVALIDATED""" + from air_runtime.task_graph import TaskGraph, TaskNode, PlanDelta + from air_runtime.impact_propagator import ImpactLabel + + graph = TaskGraph() + graph.add_node(TaskNode(id="T-001", status="DISPATCHED", task="ffmpeg 流媒体", adr_refs=["ADR-0005"])) + + delta = PlanDelta() + labels = {"T-001": ImpactLabel.IMPACTED.value} + report = graph.invalidate_by_adr("ADR-0005", delta, impact_labels=labels) + + assert report.terminated_in_progress == 1 + assert graph.nodes["T-001"].status == "INVALIDATED" + print("✓ IMPACTED+DISPATCHED→INVALIDATED 测试通过") + return True + + +def test_backward_compat_uniform_invalidation(): + """向后兼容: impact_labels=None 时使用旧统一失效行为""" + from air_runtime.task_graph import TaskGraph, TaskNode, Edge, PlanDelta + + graph = TaskGraph() + graph.add_node(TaskNode(id="T-001", status="DONE", task="任务1", adr_refs=["ADR-0005"])) + graph.add_node(TaskNode(id="T-002", status="DONE", task="任务2", adr_refs=["ADR-0005"])) + graph.add_node(TaskNode(id="T-003", status="TODO", task="下游任务")) + graph.add_edge(Edge(source="T-001", target="T-003")) + + delta = PlanDelta() + report = graph.invalidate_by_adr("ADR-0005", delta) # 不传 impact_labels + + assert report.invalidated_completed == 2 + assert report.cascaded_downstream == 1 + assert graph.nodes["T-001"].status == "INVALIDATED" + assert graph.nodes["T-002"].status == "INVALIDATED" + assert graph.nodes["T-003"].status == "INVALIDATED" + print("✓ 向后兼容统一失效测试通过") + return True + + +def test_partial_replanner_with_constraints(): + """局部重规划: replan_with_constraints 包含冻结接口""" + from air_runtime.task_graph import TaskGraph, TaskNode + from air_runtime.partial_replanner import PartialReplanner, Interface + + 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/safe.cpp"], adr_refs=["ADR-0008"])) + + replanner = PartialReplanner() + frozen = [Interface(task_id="T-002", write_set=["src/safe.cpp"], adr_refs=["ADR-0008"])] + delta = replanner.replan_with_constraints(graph, ["T-001"], frozen) + + request = delta.replan_request + assert request["type"] == "partial-replan" + # 稳定接口应包含 T-002 + stable_ids = {s["task_id"] for s in request["stableInterfaces"]} + assert "T-002" in stable_ids + print("✓ replan_with_constraints 测试通过") + return True + + +def test_generate_verification_tasks(): + """局部重规划: generate_verification_tasks 为 BOUNDARY 生成验证任务""" + 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="DONE", task="集成测试", + files_dirs="tests/integration.cpp", write_set=["tests/integration.cpp"])) + graph.add_node(TaskNode(id="T-002", status="TODO", task="未完成任务")) + + replanner = PartialReplanner() + delta = replanner.generate_verification_tasks(graph, ["T-001", "T-002"]) + + # T-001 DONE → 生成验证任务；T-002 TODO → 不生成 + assert len(delta.added_tasks) == 1 + vfy = delta.added_tasks[0] + assert vfy.id == "T-001-VFY" + assert vfy.status == "TODO" + assert "边界验证" in vfy.task + assert vfy.test_required == True + + # 验证边 + assert len(delta.edge_changes.added) == 1 + assert delta.edge_changes.added[0].source == "T-001" + assert delta.edge_changes.added[0].target == "T-001-VFY" + print("✓ generate_verification_tasks 测试通过") + return True + + +def test_extract_safe_interfaces(): + """eng_mode: _extract_safe_interfaces 提取 SAFE+DONE 任务接口""" + from air_runtime.task_graph import TaskGraph, TaskNode + from air_runtime.impact_propagator import ImpactPropagator, ImpactLabel + from air_runtime.change_classifier import BlastRadius + from air_runtime.modes.eng_mode import _extract_safe_interfaces + + graph = TaskGraph() + graph.add_node(TaskNode(id="T-001", status="DONE", task="安全任务", + write_set=["src/safe.cpp"], adr_refs=["ADR-0008"])) + graph.add_node(TaskNode(id="T-002", status="DONE", task="受影响任务", + write_set=["src/broken.cpp"], adr_refs=["ADR-0005"])) + graph.add_node(TaskNode(id="T-003", status="TODO", task="安全但未开始")) + + # 先传播标记 + propagator = ImpactPropagator() + propagator.propagate(graph, "ADR-0005", BlastRadius.INTERFACE) + + interfaces = _extract_safe_interfaces(graph) + safe_ids = {i.task_id for i in interfaces} + assert "T-001" in safe_ids, f"T-001 是 SAFE+DONE 应有接口: {safe_ids}" + assert "T-002" not in safe_ids, "T-002 是 IMPACTED 不应有接口" + assert "T-003" not in safe_ids, "T-003 是 TODO 不应有接口" + print("✓ _extract_safe_interfaces 测试通过") + return True + + +def test_adr_watcher_content_for_classification(): + """ADRWatcher: get_content_for_classification 返回旧/新内容""" + from air_runtime.adr_watcher import ADRWatcher, ADRChange + + with tempfile.TemporaryDirectory() as tmpdir: + adr_dir = Path(tmpdir) + adr_file = adr_dir / "ADR-0005-ffmpeg.md" + adr_file.write_text("# ADR-0005\n\nStatus: accepted\n\n使用 ffmpeg。") + + watcher = ADRWatcher(adr_dir) + watcher.snapshot() + + # 修改 + adr_file.write_text("# ADR-0005\n\nStatus: superseded\n\n改用 gstreamer。") + changes = watcher.detect_changes() + assert len(changes) == 1 + + old_content, new_content = watcher.get_content_for_classification(changes[0]) + assert old_content is not None + assert "ffmpeg" in old_content + assert new_content is not None + assert "gstreamer" in new_content + + print("✓ get_content_for_classification 测试通过") + return True + + +def test_full_differentiated_flow(): + """完整差异化流程：分类→传播→失效→验证→安全接口""" + from air_runtime.adr_watcher import ADRWatcher + from air_runtime.task_graph import TaskGraph, TaskNode, Edge, PlanDelta + from air_runtime.change_classifier import ChangeClassifier + from air_runtime.impact_propagator import ImpactPropagator, ImpactLabel + from air_runtime.partial_replanner import PartialReplanner + from air_runtime.modes.eng_mode import _extract_safe_interfaces + + with tempfile.TemporaryDirectory() as tmpdir: + adr_dir = Path(tmpdir) / "adr" + adr_dir.mkdir(parents=True) + adr_file = adr_dir / "ADR-0005-ffmpeg.md" + adr_file.write_text("# ADR-0005\n\nStatus: accepted\n\n使用 ffmpeg API 实现视频解码。") + + watcher = ADRWatcher(adr_dir) + watcher.snapshot() + + # 修改 ADR: 接口变更 + adr_file.write_text( + "# ADR-0005\n\nStatus: modified\n\n" + "改用 gstreamer API。must implement 新接口。caller 需要更新签名。" + ) + changes = watcher.detect_changes() + change = changes[0] + + # Phase 1: 分类 + old_c, new_c = watcher.get_content_for_classification(change) + classifier = ChangeClassifier() + classification = classifier.classify(change.adr_id, old_c, new_c) + assert classification.blast_radius.value == "interface" + + # Phase 1: 传播 + graph = TaskGraph() + graph.add_node(TaskNode(id="T-001", status="DONE", task="ffmpeg 解码器", + write_set=["src/decoder.cpp"], adr_refs=["ADR-0005"])) + graph.add_node(TaskNode(id="T-002", status="DONE", task="集成测试")) + graph.add_node(TaskNode(id="T-003", status="DONE", task="UI 界面", + write_set=["src/ui.cpp"], adr_refs=["ADR-0008"])) + graph.add_node(TaskNode(id="T-004", status="TODO", task="部署")) + graph.add_edge(Edge(source="T-001", target="T-002")) + graph.add_edge(Edge(source="T-002", target="T-004")) + + propagator = ImpactPropagator() + prop_result = propagator.propagate(graph, "ADR-0005", classification.blast_radius) + assert "T-001" in prop_result.impacted + assert "T-002" in prop_result.boundary + assert "T-003" in prop_result.safe + assert "T-004" in prop_result.safe # 下游2跳 + + # Phase 1: 差异化失效 + impact_labels = {} + for nid in prop_result.impacted: + impact_labels[nid] = ImpactLabel.IMPACTED.value + for nid in prop_result.boundary: + impact_labels[nid] = ImpactLabel.BOUNDARY.value + for nid in prop_result.safe: + impact_labels[nid] = ImpactLabel.SAFE.value + + delta = PlanDelta() + report = graph.invalidate_by_adr("ADR-0005", delta, impact_labels=impact_labels) + + # T-001 IMPACTED+DONE → INVALIDATED + assert graph.nodes["T-001"].status == "INVALIDATED" + # T-002 BOUNDARY+DONE → verification_needed + assert graph.nodes["T-002"].status == "DONE" + assert graph.nodes["T-002"].meta.get("verification_needed") == True + # T-003 SAFE → untouched + assert graph.nodes["T-003"].status == "DONE" + # T-004 SAFE TODO → untouched + assert graph.nodes["T-004"].status == "TODO" + + # Phase 2: 生成验证任务 + replanner = PartialReplanner() + vfy_delta = replanner.generate_verification_tasks(graph, report.boundary_task_ids) + assert len(vfy_delta.added_tasks) == 1 + assert vfy_delta.added_tasks[0].id == "T-002-VFY" + + # Phase 3: 提取 SAFE 接口 + safe_ifaces = _extract_safe_interfaces(graph) + safe_ids = {i.task_id for i in safe_ifaces} + assert "T-003" in safe_ids + assert "T-001" not in safe_ids + + print("✓ 完整差异化流程测试通过") + return True + + +def test_cascade_report_extended_fields(): + """CascadeReport: 3.2.17 扩展字段 boundary_task_ids, blast_radius""" + from air_runtime.task_graph import CascadeReport + + report = CascadeReport( + invalidated_completed=1, + terminated_in_progress=0, + cascaded_downstream=0, + rollback_ref="ref-001", + invalidated_task_ids=["T-001"], + boundary_task_ids=["T-002"], + blast_radius="interface", + ) + assert report.boundary_task_ids == ["T-002"] + assert report.blast_radius == "interface" + + # 默认值 + report2 = CascadeReport() + assert report2.boundary_task_ids == [] + assert report2.blast_radius == "" + print("✓ CascadeReport 扩展字段测试通过") + return True + + +def main(): + print("=" * 50) + print("3.2.17 中途变更处理 Phase 2 功能测试") + print("=" * 50) + + tests = [ + ("IMPLEMENTATION 分类", test_change_classifier_implementation), + ("INTERFACE 分类", test_change_classifier_interface), + ("GLOBAL_CONSTRAINT 分类", test_change_classifier_global), + ("新增 ADR 分类", test_change_classifier_new_adr), + ("删除 ADR 分类", test_change_classifier_deleted_adr), + ("保守回退分类", test_change_classifier_conservative_fallback), + ("IMPLEMENTATION 传播", test_impact_propagator_implementation), + ("INTERFACE 传播+BOUNDARY", test_impact_propagator_interface), + ("GLOBAL 全图传播", test_impact_propagator_global), + ("clear_labels", test_impact_propagator_clear_labels), + ("IMPACTED+DONE→INVALIDATED", test_differentiated_invalidation_impacted_done), + ("BOUNDARY+DONE→verification", test_differentiated_invalidation_boundary_done), + ("SAFE→不处理", test_differentiated_invalidation_safe_untouched), + ("IMPACTED+DISPATCHED→INVALIDATED", test_differentiated_invalidation_impacted_dispatched), + ("向后兼容统一失效", test_backward_compat_uniform_invalidation), + ("replan_with_constraints", test_partial_replanner_with_constraints), + ("generate_verification_tasks", test_generate_verification_tasks), + ("_extract_safe_interfaces", test_extract_safe_interfaces), + ("get_content_for_classification", test_adr_watcher_content_for_classification), + ("完整差异化流程", test_full_differentiated_flow), + ("CascadeReport 扩展字段", test_cascade_report_extended_fields), + ] + + 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) diff --git a/test_p1_21_supplement.py b/test_p1_21_supplement.py index d5acfa9..6f11004 100644 --- a/test_p1_21_supplement.py +++ b/test_p1_21_supplement.py @@ -228,7 +228,7 @@ def test_eng_monitor_detects_adr_change(): 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) + changes, _watcher = _detect_adr_changes(project_root, state) assert len(changes) == 0 # 首次只初始化，无变更 # 保存 hash 到 state @@ -240,7 +240,7 @@ def test_eng_monitor_detects_adr_change(): # 第二次检测 state = safe_json_load(paths["state"]) or {} - changes = _detect_adr_changes(project_root, state) + changes, _watcher = _detect_adr_changes(project_root, state) assert len(changes) == 1 assert changes[0].adr_id == "ADR-0005" assert changes[0].kind == "superseded"