""" 动态任务依赖图(DAG)— V2 P1-14 修复。 替代 V1 静态 todo.md 表格,支持 Arc 增量重规划,Eng 增量吸收。 """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any @dataclass class TaskNode: id: str status: str = "TODO" # TODO | DISPATCHED | DONE | BLOCKED task: str = "" files_dirs: str = "" done_when: str = "" in_degree: int = 0 out_edges: list[str] = field(default_factory=list) write_set: list[str] = field(default_factory=list) meta: dict[str, Any] = field(default_factory=dict) @dataclass class Edge: source: str target: str kind: str = "dependency" # dependency | conflict | sync @dataclass class EdgeChange: added: list[Edge] = field(default_factory=list) removed: list[Edge] = field(default_factory=list) @dataclass class PlanDelta: """Arc 重规划产出的增量差异,替代全量覆盖 todo.md。""" removed_tasks: list[str] = field(default_factory=list) added_tasks: list[TaskNode] = field(default_factory=list) modified_tasks: list[TaskNode] = field(default_factory=list) edge_changes: EdgeChange = field(default_factory=EdgeChange) class TaskGraph: """动态任务依赖图,支持增量更新和全量替换。""" def __init__(self): self.nodes: dict[str, TaskNode] = {} self.edges: list[Edge] = [] def add_node(self, node: TaskNode) -> None: self.nodes[node.id] = node def add_edge(self, edge: Edge) -> None: self.edges.append(edge) if edge.target in self.nodes: self.nodes[edge.target].in_degree += 1 if edge.source in self.nodes: self.nodes[edge.source].out_edges.append(edge.target) def apply_delta(self, delta: PlanDelta) -> None: """增量吸收 Arc 的重规划结果,保留已调度任务不受影响。""" for task_id in delta.removed_tasks: self._remove_node(task_id) for node in delta.added_tasks: self._add_node(node) for node in delta.modified_tasks: self._update_node(node) for edge in delta.edge_changes.removed: self._remove_edge(edge) for edge in delta.edge_changes.added: self._add_edge(edge) def apply_full_replace(self, nodes: list[TaskNode], edges: list[Edge]) -> None: """全量替换模式:Arc 产出完整 DAG,保留已完成任务状态。""" done_status = {tid: n.status for tid, n in self.nodes.items() if n.status == "DONE"} self.nodes = {n.id: n for n in nodes} self.edges = list(edges) for tid, status in done_status.items(): if tid in self.nodes: self.nodes[tid].status = status for edge in self.edges: if edge.target in self.nodes: self.nodes[edge.target].in_degree += 1 if edge.source in self.nodes: self.nodes[edge.source].out_edges.append(edge.target) def ready_tasks(self) -> list[str]: """返回当前入度为 0 且状态为 TODO 的任务。""" return [nid for nid, n in self.nodes.items() if n.in_degree == 0 and n.status == "TODO"] def diff(self, other: TaskGraph) -> PlanDelta: """对比自身与 other,产出 PlanDelta(add/remove/modify node + edge changes)。 self = 新图, other = 旧图(before replan)。 """ delta = PlanDelta() old_ids = set(other.nodes.keys()) new_ids = set(self.nodes.keys()) # 移除 delta.removed_tasks = list(old_ids - new_ids) # 新增 delta.added_tasks = [self.nodes[tid] for tid in (new_ids - old_ids)] # 修改 for tid in old_ids & new_ids: old_n = other.nodes[tid] new_n = self.nodes[tid] if (old_n.task != new_n.task or old_n.files_dirs != new_n.files_dirs or old_n.done_when != new_n.done_when or old_n.write_set != new_n.write_set): delta.modified_tasks.append(new_n) # Edge 差异 old_edges = {(e.source, e.target, e.kind) for e in other.edges} new_edges = {(e.source, e.target, e.kind) for e in self.edges} for s, t, k in (new_edges - old_edges): delta.edge_changes.added.append(Edge(source=s, target=t, kind=k)) for s, t, k in (old_edges - new_edges): delta.edge_changes.removed.append(Edge(source=s, target=t, kind=k)) return delta @classmethod def load(cls, path) -> TaskGraph: """从 _export_task_graph_json 写的格式还原 TaskGraph。""" from pathlib import Path from air_runtime.io import safe_json_load p = Path(path) data = safe_json_load(p) graph = cls() if not data or not isinstance(data, dict): return graph for nid, nd in data.get("nodes", {}).items(): graph.nodes[nid] = TaskNode( id=nd.get("id", nid), status=nd.get("status", "TODO"), task=nd.get("task", ""), files_dirs=nd.get("filesDirs", ""), done_when=nd.get("doneWhen", ""), in_degree=nd.get("inDegree", 0), out_edges=list(nd.get("outEdges", [])), write_set=list(nd.get("writeSet", [])), ) for ed in data.get("edges", []): graph.edges.append(Edge( source=ed["source"], target=ed["target"], kind=ed.get("kind", "dependency"), )) return graph def task_ids_by_status(self, status: str) -> list[str]: return [nid for nid, n in self.nodes.items() if n.status == status] def find_cycles(self) -> list[list[str]]: """检测依赖环(DFS)。""" visited: set[str] = set() rec_stack: set[str] = set() cycles: list[list[str]] = [] def dfs(node_id: str, path: list[str]) -> None: visited.add(node_id) rec_stack.add(node_id) path.append(node_id) for target in self.nodes.get(node_id, TaskNode(id=node_id)).out_edges: if target not in visited: dfs(target, path.copy()) elif target in rec_stack: cycle_start = path.index(target) cycles.append(path[cycle_start:]) rec_stack.discard(node_id) for nid in self.nodes: if nid not in visited: dfs(nid, []) return cycles def export_todo_md(self) -> str: """导出为人可读的 todo.md 表格,保留 V1 的可见性优势。""" lines = ["| Task | Status | Files/Dirs | Done When | Validation | ADR |", "|------|--------|------------|-----------|------------|-----|"] for nid, node in self.nodes.items(): lines.append(f"| {node.task} | {node.status} | {node.files_dirs} | " f"{node.done_when} | | |") return "\n".join(lines) + "\n" def _remove_node(self, task_id: str) -> None: if task_id in self.nodes: del self.nodes[task_id] self.edges = [e for e in self.edges if e.source != task_id and e.target != task_id] def _add_node(self, node: TaskNode) -> None: self.nodes[node.id] = node def _update_node(self, node: TaskNode) -> None: if node.id in self.nodes: existing_status = self.nodes[node.id].status self.nodes[node.id] = node if existing_status in ("DISPATCHED", "DONE"): self.nodes[node.id].status = existing_status def _remove_edge(self, edge: Edge) -> None: self.edges = [e for e in self.edges if not (e.source == edge.source and e.target == edge.target)] if edge.target in self.nodes: self.nodes[edge.target].in_degree = max(0, self.nodes[edge.target].in_degree - 1) def _add_edge(self, edge: Edge) -> None: self.edges.append(edge) if edge.target in self.nodes: self.nodes[edge.target].in_degree += 1 if edge.source in self.nodes: self.nodes[edge.source].out_edges.append(edge.target)