admin/AirPlan-V2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: P1-21 ADR 变更级联失效 — 架构变更自动失效已完成任务

Browse Source 
- TaskNode 新增 adr_refs 字段（ADR→任务溯源链）
- 新增 INVALIDATED 状态，允许覆盖 DONE/DISPATCHED
- TaskGraph.invalidate_by_adr() 级联失效 + BFS 下游传播
- PlanDelta 新增 rollback_ref（回滚快照引用）
- CascadeReport 数据结构（失效统计 + 任务ID列表）
- Eng dispatch_frozen 冻结调度，ready_tasks() 返回空
- handle_adr_invalidation() 10步处理流程（含 git revert）
- unfreeze_after_replan() Arc 重新规划后解冻
- AirRvr check_invalidated_cleanup() 检查旧代码残留
- INV-15 + L1 保障项 15 写入 SKILL.md
- 11 项功能测试全通过，含 ffmpeg→gstreamer 完整场景

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
AirLongDian
2026-06-11 11:13:53 +08:00
parent 19a4550830
commit f2f49d0094
6 changed files with 709 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
lib/air_runtime/modes/arc_mode.py
View File

@@ -78,10 +78,12 @@ def _ensure_dirs(paths: dict[str, Path]) -> None:
def _export_task_graph_json(graph: TaskGraph, path: Path) -> None:
data = {
"dispatchFrozen": graph.dispatch_frozen, # P1-21
"nodes": {nid: {"id": n.id, "status": n.status, "task": n.task,
"filesDirs": n.files_dirs, "doneWhen": n.done_when,
"inDegree": n.in_degree, "outEdges": n.out_edges,
"writeSet": n.write_set, "testRequired": n.test_required}
"writeSet": n.write_set, "testRequired": n.test_required,
"adrRefs": n.adr_refs} # P1-21
for nid, n in graph.nodes.items()},
"edges": [{"source": e.source, "target": e.target, "kind": e.kind} for e in graph.edges],
}
@@ -96,9 +98,14 @@ def _build_graph_from_todo(todo_path: Path) -> tuple[TaskGraph, list[str]]:
violations = [] # P1-19.1: 记录 Done When 不含"测试通过"的任务
for t in tasks:
# P1-21: 从 todo.md ADR 列提取 adr_refs
adr_refs = []
if hasattr(t, "adr") and t.adr:
adr_refs = [a.strip() for a in t.adr.split(",") if a.strip()]
node = TaskNode(
id=t.task_id, status=t.status, task=t.task,
files_dirs=t.files_dirs, done_when=t.done_when,
adr_refs=adr_refs,
)
graph.add_node(node)

179
lib/air_runtime/modes/eng_mode.py
View File

@@ -26,7 +26,7 @@ from air_runtime.modes.merge_pipeline import (
sync_engine_managed_docs,
update_todo_after_merge,
)
from air_runtime.task_graph import TaskGraph
from air_runtime.task_graph import TaskGraph, CascadeReport, PlanDelta
from air_runtime.todo_parser import parse_tasks
from air_runtime.utils import now_iso, session_stamp, truncate_history
@@ -192,6 +192,23 @@ def dispatch_worker_group(project_root: Path, group_name: str = "") -> dict:
atomic_json_write(paths["state"], state)
state = safe_json_load(paths["state"]) or _init_state(project_root)
# P1-21: 检查调度冻结（ADR 级联失效期间）
tg_json = airplan_root(project_root) / "state" / "airarc" / "reviews" / "task-graph.json"
if tg_json.exists():
try:
graph = TaskGraph.load(tg_json)
if graph.dispatch_frozen:
log = EventLog(event_log_path(project_root))
log.emit("eng.blocked", {"reason": "dispatch frozen — ADR cascade invalidation in progress"})
return {
"blocked": True,
"reason": "dispatch frozen — ADR cascade invalidation in progress",
"waveId": "",
"taskIds": [],
}
except Exception:
pass
# P1-19.3: 检查是否有 block-release verdict，阻止所有后续派发
from air_runtime.review_runtime import ReviewRuntime
rvr = ReviewRuntime(project_root)
@@ -546,6 +563,166 @@ def _select_ready_tasks(project_root: Path, max_count: int) -> list[str]:
return [t.task_id for t in tasks if t.status == "TODO"][:max_count]
def handle_adr_invalidation(project_root: Path, adr_id: str) -> dict:
"""P1-21: 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. 解冻调度
"""
tg_json = airplan_root(project_root) / "state" / "airarc" / "reviews" / "task-graph.json"
if not tg_json.exists():
return {"error": "task-graph.json not found", "adrId": adr_id}
graph = TaskGraph.load(tg_json)
delta = PlanDelta()
# 2-4: 级联失效
report = graph.invalidate_by_adr(adr_id, delta)
log = EventLog(event_log_path(project_root))
log.emit("adr.invalidation", {
"adrId": adr_id,
"invalidatedCompleted": report.invalidated_completed,
"terminatedInProgress": report.terminated_in_progress,
"cascadedDownstream": report.cascaded_downstream,
})
# 4: 中止进行中的相关 Worker
paths = _paths(project_root)
_ensure_dirs(paths)
state = safe_json_load(paths["state"]) or _init_state(project_root)
terminated_workers = []
for worker in list(state.get("activeWorkers", [])):
if worker.get("taskId") in report.invalidated_task_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
]
# 5: 创建回滚快照
rollback_ref = _create_rollback_snapshot(project_root, report.invalidated_task_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)
# 7: 写回更新后的 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,
"rollbackRef": rollback_ref,
}
atomic_json_write(paths["state"], state)
return {
"adrId": adr_id,
"cascadeReport": {
"invalidatedCompleted": report.invalidated_completed,
"terminatedInProgress": report.terminated_in_progress,
"cascadedDownstream": report.cascaded_downstream,
"rollbackRef": rollback_ref,
"invalidatedTaskIds": report.invalidated_task_ids,
},
"terminatedWorkers": terminated_workers,
"revertResults": revert_results,
"nextStep": "arc-replan-then-unfreeze",
}
def _create_rollback_snapshot(project_root: Path, invalidated_task_ids: list[str]) -> str:
"""P1-21: 为失效任务创建 git tag 回滚点。"""
import subprocess
ref = f"airplan/adr-invalidate-{session_stamp()}"
try:
subprocess.run(
["git", "tag", ref],
cwd=project_root, capture_output=True, timeout=30,
)
except Exception:
pass
return ref
def _git_revert_invalidated(project_root: Path, invalidated_task_ids: list[str]) -> list[dict]:
"""P1-21: 尝试 git revert 已合并的失效任务对应的 commit。"""
import subprocess
results = []
for tid in invalidated_task_ids:
try:
# 查找包含 task_id 的 commit
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]
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,
)
else:
results.append({"taskId": tid, "commit": None, "reverted": False, "reason": "no commit found"})
except Exception as e:
results.append({"taskId": tid, "commit": None, "reverted": False, "reason": str(e)})
return results
def unfreeze_after_replan(project_root: Path, new_task_graph_path: Path | None = None) -> dict:
"""P1-21: Arc 重新生成受影响部分后，apply_delta + 解冻调度。"""
paths = _paths(project_root)
_ensure_dirs(paths)
tg_json = airplan_root(project_root) / "state" / "airarc" / "reviews" / "task-graph.json"
if not tg_json.exists():
return {"error": "task-graph.json not found"}
graph = TaskGraph.load(tg_json)
# 如果 Arc 生成了新的任务图，增量合并
if new_task_graph_path and new_task_graph_path.exists():
new_graph = TaskGraph.load(new_task_graph_path)
delta = new_graph.diff(graph)
graph.apply_delta(delta)
# 解冻
graph.unfreeze_dispatch()
from air_runtime.modes.arc_mode import _export_task_graph_json
_export_task_graph_json(graph, tg_json)
# 更新引擎状态
state = safe_json_load(paths["state"]) or _init_state(project_root)
state["dispatchFrozen"] = False
adr_info = state.pop("adrInvalidationInProgress", {})
atomic_json_write(paths["state"], state)
log = EventLog(event_log_path(project_root))
log.emit("adr.unfreezed", {"previousAdrInvalidation": adr_info})
return {"frozen": False, "readyTasks": graph.ready_tasks()}
def maybe_replan(project_root: Path, todo_path: Path | None = None) -> dict | None:
"""检查 todo.md mtime vs task_graph.json mtime，若 todo 更新则触发 replan。"""
from air_runtime.modes.arc_mode import incremental_replan_mode

19
lib/air_runtime/review_runtime.py
View File

@@ -136,6 +136,25 @@ class ReviewRuntime:
"deliveryVerdict": report.get("highRiskAudit", {}).get("deliveryVerdict", "safe-to-ship"), # P1-19.2
}
def check_invalidated_cleanup(self, invalidated_task_ids: list[str]) -> dict:
"""P1-21: 检查 INVALIDATED 任务的代码是否已清理（无残留）。"""
from air_runtime.io import safe_json_load
residual = []
for tid in invalidated_task_ids:
# 检查是否有残留的 result 文件（说明旧代码未被 revert）
result_dir = self._state_dir.parent / "airdo" / "tasks" / tid
if result_dir.exists():
result_file = result_dir / "result.json"
if result_file.exists():
data = safe_json_load(result_file)
if data and data.get("status") == "done":
residual.append({"taskId": tid, "reason": "done result still exists — code may not be reverted"})
return {
"cleaned": len(residual) == 0,
"residualCount": len(residual),
"residualDetails": residual,
}
@staticmethod
def _report_to_dict(report: ReviewReport) -> dict:
# P1-19.2: highRiskAudit 序列化

91
lib/air_runtime/task_graph.py
View File

@@ -12,7 +12,7 @@ from typing import Any
@dataclass
class TaskNode:
id: str
status: str = "TODO" # TODO | DISPATCHED | DONE | BLOCKED
status: str = "TODO" # TODO | DISPATCHED | DONE | BLOCKED | INVALIDATED
task: str = ""
files_dirs: str = ""
done_when: str = ""
@@ -21,6 +21,7 @@ class TaskNode:
write_set: list[str] = field(default_factory=list)
meta: dict[str, Any] = field(default_factory=dict)
test_required: bool = False # P1-19.1: 边界测试强制标记
adr_refs: list[str] = field(default_factory=list) # P1-21: ADR→任务溯源链
@dataclass
@@ -36,6 +37,16 @@ class EdgeChange:
removed: list[Edge] = field(default_factory=list)
@dataclass
class CascadeReport:
"""P1-21: ADR 变更级联失效报告。"""
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)
@dataclass
class PlanDelta:
"""Arc 重规划产出的增量差异，替代全量覆盖 todo.md。"""
@@ -43,6 +54,7 @@ class PlanDelta:
added_tasks: list[TaskNode] = field(default_factory=list)
modified_tasks: list[TaskNode] = field(default_factory=list)
edge_changes: EdgeChange = field(default_factory=EdgeChange)
rollback_ref: str = "" # P1-21: 回滚快照引用
class TaskGraph:
@@ -51,6 +63,7 @@ class TaskGraph:
def __init__(self):
self.nodes: dict[str, TaskNode] = {}
self.edges: list[Edge] = []
self.dispatch_frozen: bool = False # P1-21: 调度冻结
def add_node(self, node: TaskNode) -> None:
self.nodes[node.id] = node
@@ -90,7 +103,9 @@ class TaskGraph:
self.nodes[edge.source].out_edges.append(edge.target)
def ready_tasks(self) -> list[str]:
"""返回当前入度为 0 且状态为 TODO 的任务。"""
"""返回当前入度为 0 且状态为 TODO 的任务。调度冻结时返回空。"""
if self.dispatch_frozen:
return []
return [nid for nid, n in self.nodes.items() if n.in_degree == 0 and n.status == "TODO"]
def diff(self, other: TaskGraph) -> PlanDelta:
@@ -138,6 +153,7 @@ class TaskGraph:
graph = cls()
if not data or not isinstance(data, dict):
return graph
graph.dispatch_frozen = data.get("dispatchFrozen", False)
for nid, nd in data.get("nodes", {}).items():
graph.nodes[nid] = TaskNode(
id=nd.get("id", nid),
@@ -149,6 +165,7 @@ class TaskGraph:
out_edges=list(nd.get("outEdges", [])),
write_set=list(nd.get("writeSet", [])),
test_required=nd.get("testRequired", False),
adr_refs=list(nd.get("adrRefs", [])),
)
for ed in data.get("edges", []):
graph.edges.append(Edge(
@@ -205,7 +222,8 @@ class TaskGraph:
if node.id in self.nodes:
existing_status = self.nodes[node.id].status
self.nodes[node.id] = node
if existing_status in ("DISPATCHED", "DONE"):
# INVALIDATED 可覆盖 DONE/DISPATCHED（P1-21: ADR 级联失效）
if existing_status in ("DISPATCHED", "DONE") and node.status != "INVALIDATED":
self.nodes[node.id].status = existing_status
def _remove_edge(self, edge: Edge) -> None:
@@ -220,3 +238,70 @@ class TaskGraph:
self.nodes[edge.target].in_degree += 1
if edge.source in self.nodes:
self.nodes[edge.source].out_edges.append(edge.target)
# P1-21: ADR 级联失效
def tasks_by_adr(self, adr_id: str) -> list[TaskNode]:
"""查找所有引用指定 ADR 的任务（含已完成）。"""
return [n for n in self.nodes.values() if adr_id in n.adr_refs]
def _find_downstream(self, task_ids: list[str]) -> list[str]:
"""BFS 遍历下游依赖任务。"""
visited: set[str] = set()
queue = list(task_ids)
while queue:
current = queue.pop(0)
if current in visited:
continue
visited.add(current)
node = self.nodes.get(current)
if node:
for target in node.out_edges:
if target not in visited:
queue.append(target)
# 排除起点自身
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 变更时级联失效所有相关任务。"""
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"]
# 1. 冻结调度
self.dispatch_frozen = True
# 2. 标记已完成任务为 INVALIDATED
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. 级联失效下游
downstream_ids = self._find_downstream([t.id for t in completed + in_progress])
cascaded = []
for tid in downstream_ids:
node = self.nodes.get(tid)
if node and node.status in ("TODO", "DISPATCHED"):
node.status = "INVALIDATED"
delta.removed_tasks.append(tid)
cascaded.append(tid)
# 5. 回滚快照引用（由调用方在 git revert 后填入）
all_invalidated = [t.id for t in completed + in_progress] + cascaded
return CascadeReport(
invalidated_completed=len(completed),
terminated_in_progress=len(in_progress),
cascaded_downstream=len(cascaded),
rollback_ref=delta.rollback_ref,
invalidated_task_ids=all_invalidated,
)
def unfreeze_dispatch(self) -> None:
"""P1-21: 解冻调度，在 Arc 重新生成受影响任务后调用。"""
self.dispatch_frozen = False