""" AirEng mode — V2 调度引擎。 L1 代码级保障：硬编码轮询循环、Worker 超时、资源压力检测、事务化合并。 """ from __future__ import annotations import json import os import time from datetime import datetime, timezone from pathlib import Path from air_runtime.io import atomic_json_write, safe_json_load from air_runtime.lock import FileLock from air_runtime.paths import ( airplan_root, todo_path as get_todo_path, engine_state_path, event_log_path, plan_path, agents_path, ) from air_runtime.events import EventLog, TASK_DISPATCHED, TASK_COMPLETED, TASK_BLOCKED, MERGE_STARTED, MERGE_COMPLETED, \ INTERVENTION_STALL, ENGINE_CYCLE, WORKER_TIMEOUT, REPAIR_CREATED, REPAIR_RESOLVED from air_runtime.evidence_gate import EvidenceGatePolicy, EvidenceClass from air_runtime.modes.merge_pipeline import ( apply_document_updates, enforce_doc_sync_requirements, sync_engine_managed_docs, update_todo_after_merge, ) 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 WORKER_MAX_WALL_TIME = 7200 # 2小时硬上限 DEFAULT_CONCURRENCY = 3 MONITOR_INTERVAL_SECONDS = 300 # 5分钟 AIRDBG_MAX_ATTEMPTS = 1 # AirDbg 升级最大尝试次数，超过则降级为串行重执行 def check_worktree_merge_status(project_root: Path, task_id: str) -> dict: """ 检查某 task 的 worktree 是否需要 merge 回主分支。 如果 merge 失败（conflicts），自动升级到 AirDbg。 再失败则降级为串行重执行。 返回: {"status": "ok" | "upgraded_to_airdbg" | "downgraded_to_serial", ...} """ from air_runtime.worktree import WorktreeIsolation wt_path = project_root / ".git" / "worktrees" / f"air-{task_id}" if not wt_path.exists(): return {"status": "ok"} # 无 worktree，正常 # 尝试 merge 回主分支 wt = WorktreeIsolation(repo_root=project_root) result = wt.merge_back(task_id, wt_path) if result.successful: # merge 成功，清理 worktree wt.cleanup(task_id, wt_path) return {"status": "ok", "conflicts": []} # merge 失败 → 升级到 AirDbg from air_runtime.modes.dbg_mode import start_session session = start_session(project_root, task_id) return { "status": "upgraded_to_airdbg", "taskId": task_id, "conflicts": result.conflicts, "sessionId": session.get("sessionId"), } def _paths(project_root: Path) -> dict[str, Path]: root = airplan_root(project_root) / "state" / "aireng" return { "root": root, "state": root / "state.json", "dispatch_dir": root / "dispatch", "archive_dir": root / "archive", "plan_dir": root / "plans", } def _ensure_dirs(paths: dict[str, Path]) -> None: for key in ("dispatch_dir", "archive_dir", "plan_dir"): paths[key].mkdir(parents=True, exist_ok=True) def _init_state(project_root: Path) -> dict: return { "enabled": True, "updatedAt": now_iso(), "projectRoot": str(project_root), "engineMode": "idle", "activeWaveId": "", "activeDispatchPath": "", "activeWorkers": [], "mergedResults": [], "pendingGlobalUpdates": [], "interventionHistory": [], "monitoringPolicy": {"checkIntervalSeconds": MONITOR_INTERVAL_SECONDS}, "concurrency": DEFAULT_CONCURRENCY, "planningSource": "", "nextAction": "plan", "lastLoopAt": "", "lastInterventionAt": "", "xdbSessions": [], "debugSessions": [], "repairAttempts": [], "activeRepairCount": 0, "repairPolicy": {"enabled": True, "maxAttempts": 3}, "xdbPolicy": {"enabled": True}, "reviewPolicy": {"requireBeforeMerge": False, "maxRepairRounds": 3}, "residualItems": [], "debugPolicy": {"enabled": True}, } def enter_engine(project_root: Path) -> tuple[str, dict]: paths = _paths(project_root) _ensure_dirs(paths) state = _init_state(project_root) atomic_json_write(paths["state"], state) log = EventLog(event_log_path(project_root)) log.emit("engine.entered") return str(paths["state"]), {} def status_engine(project_root: Path) -> dict: paths = _paths(project_root) return safe_json_load(paths["state"]) or _init_state(project_root) def build_engine_plan(project_root: Path, todo_path: Path) -> dict: paths = _paths(project_root) _ensure_dirs(paths) arc_reviews = airplan_root(project_root) / "state" / "airarc" / "reviews" plan_json = arc_reviews / "execution-plan.json" task_graph_json = arc_reviews / "task-graph.json" planning_source = "engine-fallback" plan_data: dict = {} if plan_json.exists(): loaded = safe_json_load(plan_json) if loaded and isinstance(loaded, dict): plan_data = loaded planning_source = "airarc-execution-plan" if not plan_data: tasks = parse_tasks(todo_path) plan_data = { "selectedTasks": [t.task_id for t in tasks if t.status == "TODO"], "parallelGroups": [], } plan_path = paths["plan_dir"] / f"{session_stamp()}.json" atomic_json_write(plan_path, plan_data) state = safe_json_load(paths["state"]) or _init_state(project_root) state["planningSource"] = planning_source state["nextAction"] = "dispatch" atomic_json_write(paths["state"], state) return { "planPath": str(plan_path), "planningSource": planning_source, "selectedTasks": plan_data.get("selectedTasks", []), "parallelGroupCount": len(plan_data.get("parallelGroups", [])), "taskGraphPath": str(task_graph_json), "planJson": plan_data, } def dispatch_worker_group(project_root: Path, group_name: str = "") -> dict: """派发 worker 组，含区域冲突检测。""" paths = _paths(project_root) _ensure_dirs(paths) # P0 修复：每次派发前检查 todo.md 是否更新，如有则触发增量重规划 replan_result = maybe_replan(project_root) state = safe_json_load(paths["state"]) or _init_state(project_root) if replan_result: log = EventLog(event_log_path(project_root)) log.emit("eng.replan.triggered", { "added": replan_result.get("added_count", 0), "removed": replan_result.get("removed_count", 0), "modified": replan_result.get("modified_count", 0), }) state["lastReplanAt"] = now_iso() 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) # 扫描最新的审查报告，检查是否有 block-release rvr_state = rvr._state_dir / "reports" block_release_found = False latest_verdict = "safe-to-ship" if rvr_state.exists(): for report_file in sorted(rvr_state.glob("*.json"), key=lambda p: p.stat().st_mtime, reverse=True)[:10]: report_data = safe_json_load(report_file) if report_data: dv = report_data.get("highRiskAudit", {}).get("deliveryVerdict", "safe-to-ship") if dv == "block-release": block_release_found = True latest_verdict = dv break elif dv == "needs-fix": latest_verdict = dv if block_release_found: log = EventLog(event_log_path(project_root)) log.emit("eng.blocked", {"reason": "block-release verdict from review", "verdict": latest_verdict}) return { "blocked": True, "reason": "block-release verdict from AirRvr review - all dispatches halted", "deliveryVerdict": latest_verdict, "waveId": "", "taskIds": [], } wave_id = f"wave-{session_stamp()}" task_ids = _select_ready_tasks(project_root, state.get("concurrency", DEFAULT_CONCURRENCY)) if not task_ids: return {"dispatchPath": "", "groupName": group_name, "waveId": wave_id, "taskIds": [], "recommendedConcurrency": 0} # 区域冲突检测：多个任务时检查写集重叠 dispatch_metadata: dict | None = None if len(task_ids) > 1: from air_runtime.worktree import RegionConflictDetector, ConflictLevel todo_path = get_todo_path(project_root) tasks = parse_tasks(todo_path) task_write_sets = { t.task_id: [f.strip() for f in t.files_dirs.split(",") if f.strip()] for t in tasks if t.task_id in task_ids } if task_write_sets: detector = RegionConflictDetector() conflicts = detector.detect_batch(task_write_sets) hard_blocked = [c for c in conflicts if c.level == ConflictLevel.HARD] if hard_blocked: # HARD 冲突：强制串行，只派第一个 task_ids = task_ids[:1] dispatch_metadata = { "forcedSerialization": True, "reason": f"HARD conflict: {hard_blocked[0].task_a} <-> {hard_blocked[0].task_b}", } else: soft_conflicts = [c for c in conflicts if c.level == ConflictLevel.SOFT] if soft_conflicts: dispatch_metadata = { "worktreeIsolation": True, "softConflicts": [c.to_dict() for c in soft_conflicts], } dispatch_payload = { "waveId": wave_id, "groupName": group_name, "taskIds": task_ids, "createdAt": now_iso(), "recommendedConcurrency": min(len(task_ids), state.get("concurrency", DEFAULT_CONCURRENCY)), } dispatch_path = paths["dispatch_dir"] / f"{wave_id}.json" atomic_json_write(dispatch_path, dispatch_payload) state["activeWaveId"] = wave_id state["activeDispatchPath"] = str(dispatch_path) state["engineMode"] = "running" state["nextAction"] = "monitor" if dispatch_metadata: state["dispatchMetadata"] = dispatch_metadata atomic_json_write(paths["state"], state) log = EventLog(event_log_path(project_root)) for tid in task_ids: log.emit(TASK_DISPATCHED, {"taskId": tid, "waveId": wave_id}) result = { "dispatchPath": str(dispatch_path), "groupName": group_name, "waveId": wave_id, "taskIds": task_ids, "recommendedConcurrency": dispatch_payload["recommendedConcurrency"], } if dispatch_metadata: result["dispatchMetadata"] = dispatch_metadata 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) state = safe_json_load(paths["state"]) or _init_state(project_root) active_workers = state.get("activeWorkers", []) stalled_count = 0 ready_to_merge = 0 interventions = [] for worker in active_workers: worker_state_path = Path(worker.get("workerStatePath", "")) age = (datetime.now(timezone.utc) - datetime.fromisoformat(worker.get("spawnedAt", now_iso()))).total_seconds() # 超时检测 if age > WORKER_MAX_WALL_TIME: interventions.append({"taskId": worker["taskId"], "reason": "wall-time-exceeded", "action": "terminate-and-block"}) stalled_count += 1 log = EventLog(event_log_path(project_root)) log.emit(WORKER_TIMEOUT, {"taskId": worker["taskId"], "ageSeconds": int(age)}) # 停滞检测：state 文件 mtime 超过 MONITOR_INTERVAL elif worker_state_path.exists(): mtime = worker_state_path.stat().st_mtime if time.time() - mtime > MONITOR_INTERVAL_SECONDS: interventions.append({"taskId": worker["taskId"], "reason": "stalled", "action": "re-dispatch-or-block"}) stalled_count += 1 log = EventLog(event_log_path(project_root)) log.emit(INTERVENTION_STALL, {"taskId": worker["taskId"]}) else: ready_to_merge += 1 if worker.get("status") == "done" else 0 # 资源压力检测 try: load = os.getloadavg()[0] cpu_count = os.cpu_count() or 4 resource_pressure = load > cpu_count * 2 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(): for wt_dir in wt_root.iterdir(): if wt_dir.is_dir() and wt_dir.name.startswith("air-"): task_id = wt_dir.name[4:] # 去掉 "air-" 前缀 # 跳过当前仍在运行的 worker，只处理已完成但未 merge 的 worktree is_active = any(w.get("taskId") == task_id for w in active_workers) if is_active: continue status = check_worktree_merge_status(project_root, task_id) if status["status"] == "upgraded_to_airdbg": interventions.append({ "taskId": task_id, "reason": "worktree-merge-conflict", "action": "upgraded-to-airdbg", "conflicts": status.get("conflicts", []), "sessionId": status.get("sessionId"), }) log = EventLog(event_log_path(project_root)) log.emit("worktree.merge.conflict", { "taskId": task_id, "action": "upgraded-to-airdbg", "sessionId": status.get("sessionId"), }) elif status["status"] == "downgraded_to_serial": interventions.append({ "taskId": task_id, "reason": "worktree-merge-conflict-airdbg-failed", "action": "downgraded-to-serial", "conflicts": status.get("conflicts", []), }) log = EventLog(event_log_path(project_root)) log.emit("worktree.merge.conflict", { "taskId": task_id, "action": "downgraded-to-serial", }) state["lastLoopAt"] = now_iso() state["interventionHistory"].extend(interventions) # 将升级到 AirDbg 的 session 记入 state.debugSessions for iv in interventions: if iv.get("action") == "upgraded-to-airdbg" and iv.get("sessionId"): state.setdefault("debugSessions", []).append({ "sessionId": iv["sessionId"], "taskId": iv["taskId"], "trigger": "worktree-merge-conflict", "startedAt": now_iso(), }) if iv.get("action") == "downgraded-to-serial": state.setdefault("repairAttempts", []).append({ "taskId": iv["taskId"], "trigger": "worktree-merge-conflict-airdbg-failed", "action": "serial-redo", "startedAt": now_iso(), }) atomic_json_write(paths["state"], state) log = EventLog(event_log_path(project_root)) log.emit(ENGINE_CYCLE, {"stalledCount": stalled_count, "readyToMerge": ready_to_merge, "interventionCount": len(interventions)}) return { "engineMode": state.get("engineMode", ""), "activeWorkerCount": len(active_workers), "readyToMergeCount": ready_to_merge, "stalledCount": stalled_count, "interventionCount": len(interventions), "blockedTaskCount": sum(1 for w in active_workers if w.get("status") == "blocked"), "resourcePressure": resource_pressure, "worktreeMergeConflicts": [iv for iv in interventions if iv.get("reason", "").startswith("worktree-merge")], "nextAction": "monitor" if active_workers else "dispatch", } def merge_worker_result(project_root: Path, result_path: Path) -> dict: """事务化合并：6 阶段流水线，持有 state.json 锁。""" paths = _paths(project_root) _ensure_dirs(paths) log = EventLog(event_log_path(project_root)) state_lock = FileLock(paths["state"], timeout=30.0) todo_lock = FileLock(get_todo_path(project_root), timeout=10.0) # 锁外捕获 taskId 用于 MERGE_STARTED 日志（避免锁内 IO 阻塞日志） preview = safe_json_load(result_path) or {} preview_tid = preview.get("taskId", "") if isinstance(preview, dict) else "" log.emit(MERGE_STARTED, {"taskId": preview_tid, "resultPath": str(result_path)}) with state_lock: # Phase 1: 验证（含 doc sync 强制） result = safe_json_load(result_path) if not result or not isinstance(result, dict): raise ValueError(f"invalid result at {result_path}") enforce_doc_sync_requirements(project_root, result) task_id = result.get("taskId", "") status = result.get("status", "") # Phase 1.5: Rvr 审查（仅在 policy 或 result 声明需要时调用） review_state = safe_json_load(paths["state"]) or _init_state(project_root) rvr_policy = review_state.get("reviewPolicy", {"requireBeforeMerge": False}) if rvr_policy.get("requireBeforeMerge") or result.get("requireReview"): from air_runtime.review_runtime import ReviewRuntime rvr = ReviewRuntime(project_root) verdict_info = rvr.get_verdict_for_task(task_id) verdict = verdict_info.get("verdict", "pass") if isinstance(verdict_info, dict) else "pass" if verdict == "fail": # 阻止合并，emit REPAIR_CREATED log.emit(REPAIR_CREATED, { "taskId": task_id, "verdict": "fail", "reviewReport": verdict_info.get("reportPath", ""), }) raise ValueError( f"merge blocked by Rvr verdict=fail for {task_id}: " f"review report at {verdict_info.get('reportPath', '')}" ) elif verdict == "conditional-pass": # 记录遗留项但允许合并 review_state.setdefault("residualItems", []).append({ "taskId": task_id, "verdict": "conditional-pass", "residual": verdict_info.get("residual", []), "mergedAt": now_iso(), }) # 写回 state 以便后续 Phase 6 看到 atomic_json_write(paths["state"], review_state) # pass 走原流程 # Phase 2: 归档（可重试 — 失败重抛由调用方决定） stamp = session_stamp() archive_path = paths["archive_dir"] / f"{task_id}-{stamp}.json" atomic_json_write(archive_path, result) # Phase 3: 应用文档更新（原子写入） applied = apply_document_updates(project_root, result) # Phase 4: 同步引擎管理文档（原子写入） sync_paths = sync_engine_managed_docs(project_root, result, applied) # Phase 5: 更新 todo（嵌套 FileLock） with todo_lock: update_todo_after_merge(project_root, result, applied, sync_paths) # Phase 6: 更新引擎状态（原子写入） state = safe_json_load(paths["state"]) or _init_state(project_root) state["mergedResults"].append({ "taskId": task_id, "status": status, "archivedAt": now_iso(), "archivePath": str(archive_path), "appliedDocs": [str(p) for p in applied], "syncedDocs": [str(p) for p in sync_paths], }) state["mergedResults"] = truncate_history(state["mergedResults"], max_items=100) state["activeWorkers"] = [ w for w in state.get("activeWorkers", []) if w.get("taskId") != task_id ] state["lastMergeAt"] = now_iso() atomic_json_write(paths["state"], state) # Phase 6.5: 同步 task-graph.json 节点状态（P1-24） tg_json = airplan_root(project_root) / "state" / "airarc" / "reviews" / "task-graph.json" if tg_json.exists(): from air_runtime.modes.arc_mode import _export_task_graph_json graph = TaskGraph.load(tg_json) if task_id in graph.nodes: new_status = "DONE" if status == "done" else status.upper() graph.nodes[task_id].status = new_status _export_task_graph_json(graph, tg_json) log.emit(MERGE_COMPLETED, { "taskId": task_id, "status": status, "archivePath": str(archive_path), "appliedDocCount": len(applied), "syncedDocCount": len(sync_paths), }) # emit task completed/blocked based on merge status if status == "done": log.emit(TASK_COMPLETED, {"taskId": task_id, "archivePath": str(archive_path)}) elif status in ("blocked", "failed"): log.emit(TASK_BLOCKED, {"taskId": task_id, "status": status}) # repair resolved on successful merge after previous repair repair_attempts = state.get("repairAttempts", []) if repair_attempts and any(r.get("taskId") == task_id for r in repair_attempts): log.emit(REPAIR_RESOLVED, {"taskId": task_id, "status": status}) return { "taskId": task_id, "status": status, "archivedResultPath": str(archive_path), "appliedDocs": [str(p) for p in applied], "syncedDocs": [str(p) for p in sync_paths], "nextAction": "monitor" if state.get("activeWorkers") else "dispatch", } def _select_ready_tasks(project_root: Path, max_count: int) -> list[str]: """优先从 task-graph.json 的 DAG 计算 in-degree 为 0 的 TODO task。 DAG 中 ready 为空意味着无任务可派发（全部完成或全部被依赖阻塞），不应 fallback 到 todo.md。""" task_graph_json = airplan_root(project_root) / "state" / "airarc" / "reviews" / "task-graph.json" if task_graph_json.exists(): try: graph = TaskGraph.load(task_graph_json) ready = graph.ready_tasks() return ready[:max_count] # 空列表也是正确答案，不 fallback except Exception: pass # fallback：仅在 task-graph.json 不存在时使用 todo.md todo = get_todo_path(project_root) if not todo.exists(): return [] tasks = parse_tasks(todo) return [t.task_id for t in tasks if t.status == "TODO"][:max_count] def spawn_workers(project_root: Path, task_ids: list[str]) -> list[dict]: """T-1.21: 为每个 ready 任务准备 spawn 指令，返回 Agent 可直接消费的 Skill 调用参数列表。 不实际启动子进程——启动由 Agent 框架的 Skill 工具完成。 """ tg_json = airplan_root(project_root) / "state" / "airarc" / "reviews" / "task-graph.json" graph = TaskGraph.load(tg_json) if tg_json.exists() else TaskGraph() instructions = [] for tid in task_ids: node = graph.nodes.get(tid) task_text = node.task if node else "" instructions.append({ "skill": "airplan-v2:do", "args": f"--sub enter --task-id {tid} --task-text '{task_text}' --project .", "taskId": tid, "taskText": task_text, }) return instructions 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) # 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) # 更新引擎状态 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, "replanRequestPath": str(replan_request_path), "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 todo = todo_path or get_todo_path(project_root) tg = airplan_root(project_root) / "state" / "airarc" / "reviews" / "task-graph.json" if not tg.exists(): return None if not todo.exists(): return None if todo.stat().st_mtime <= tg.stat().st_mtime: return None return incremental_replan_mode(project_root, todo, tg) def main(args) -> None: project_root = Path(args.project).expanduser().resolve() sub = args.sub or "status" if sub == "enter": state_path, _ = enter_engine(project_root) print("airplan_mode=eng") print(f"state_path={state_path}") elif sub == "status": state = status_engine(project_root) print(f"airplan_mode=eng") print(f"enabled={state.get('enabled', False)}") print(f"engine_mode={state.get('engineMode', '')}") print(f"active_workers={len(state.get('activeWorkers', []))}") print(f"merged_results={len(state.get('mergedResults', []))}") print(f"next_action={state.get('nextAction', '')}") elif sub == "plan": tpath = Path(args.todo).expanduser().resolve() if args.todo else get_todo_path(project_root) result = build_engine_plan(project_root, tpath) print("airplan_mode=eng") print(f"planning_source={result['planningSource']}") print(f"selected_tasks={','.join(result['selectedTasks'])}") elif sub == "dispatch": result = dispatch_worker_group(project_root, args.dispatch_group) print("airplan_mode=eng") print(f"wave_id={result['waveId']}") print(f"task_ids={','.join(result['taskIds'])}") print(f"dispatch_path={result['dispatchPath']}") elif sub == "monitor": result = monitor_engine(project_root) print("airplan_mode=eng") print(f"active_workers={result['activeWorkerCount']}") print(f"ready_to_merge={result['readyToMergeCount']}") print(f"stalled={result['stalledCount']}") print(f"interventions={result['interventionCount']}") print(f"worktree_merge_conflicts={len(result.get('worktreeMergeConflicts', []))}") print(f"next_action={result['nextAction']}") elif sub == "merge": result_path = Path(args.result).expanduser().resolve() merged = merge_worker_result(project_root, result_path) print("airplan_mode=eng") print(f"task_id={merged['taskId']}") print(f"status={merged['status']}") print(f"next_action={merged['nextAction']}")