"""Exact reviewed model-import cycles; preserve the complete artifact graph.""" from pathlib import Path import yaml def reviewed_import_cycles(artifacts, root: Path) -> list[dict] | None: """Return reviews only when every cyclic component matches its exact edge set. None means the ordinary zero-cycle threshold still applies. Reviews cannot excuse new nodes, edges, relationship types, or a missing boundary rationale. """ path = root / "validation/model-import-reviews.yaml" if not path.is_file(): return None try: payload = yaml.safe_load(path.read_text()) reviews = payload["reviews"] if not isinstance(reviews, list): return None by_id = {a.id: a for a in artifacts} edges = {(a.id, rel.get("type"), rel.get("target")) for a in artifacts for rel in a.relationships if rel.get("target") in by_id} graph = {key: {target for source, _, target in edges if source == key} for key in by_id} reachable = {} for node in graph: seen, pending = set(), list(graph[node]) while pending: target = pending.pop() if target not in seen: seen.add(target) pending.extend(graph[target] - seen) reachable[node] = seen components = {frozenset(other for other in graph if other in reachable[node] and node in reachable[other]) for node in graph if node in reachable[node]} matched = [] for nodes in sorted(components, key=lambda group: sorted(group)): actual = {edge for edge in edges if edge[0] in nodes and edge[2] in nodes} for review in reviews: expected = {tuple(edge) for edge in review["edges"]} rationale = (root / review["boundary_review"]).resolve() if (actual == expected and all(by_id[node].kind == "model" for node in nodes) and rationale.is_relative_to(root.resolve()) and rationale.is_file() and all(kind == "uses" for _, kind, _ in actual)): matched.append({"nodes": sorted(nodes), "edges": sorted(actual), "boundary_review": review["boundary_review"]}) break else: return None return matched except (KeyError, TypeError, ValueError, OSError, yaml.YAMLError): return None