gabriel / muse public
dag.py python
37 lines 1.3 KB
Raw
sha256:133f9bcf57a62ec7ebf0cd71138b54a200b15989aea2fb2a6912497e2926aa8a feat(dev-safety): Phase 1 of #185 — audit tool for ambiguou… Sonnet 5 patch 2 days ago
1 """Lightweight DAG cycle detector for the identity domain plugin.
2
3 Self-contained — no external deps, no imports from musehub.
4 Enforces I1 (acyclicity) during merge by running a DFS from each proposed
5 target node and rejecting edges that would reach the source.
6 """
7
8 from collections import defaultdict
9
10 class _DAG:
11 def __init__(self) -> None:
12 self._adj: dict[str, list[str]] = defaultdict(list)
13
14 def add_edge(self, frm: str, to: str) -> None:
15 self._adj[frm].append(to)
16
17 def would_cycle(self, frm: str, to: str) -> bool:
18 """Return True if adding frm→to would create a cycle (or is a self-loop)."""
19 if frm == to:
20 return True
21 from muse.core.graph import walk_dag
22 for node in walk_dag(to, lambda n: self._adj.get(n, []), order="dfs"):
23 if node == frm:
24 return True
25 return False
26
27 def build_dag_from_paths(relationship_paths: set[str]) -> _DAG:
28 """Build a DAG from a set of canonical relationship file paths."""
29 from .records import parse_relationship_path
30
31 dag = _DAG()
32 for path in relationship_paths:
33 parsed = parse_relationship_path(path)
34 if parsed is not None:
35 frm, _edge, to = parsed
36 dag.add_edge(frm, to)
37 return dag
File History 1 commit
sha256:133f9bcf57a62ec7ebf0cd71138b54a200b15989aea2fb2a6912497e2926aa8a feat(dev-safety): Phase 1 of #185 — audit tool for ambiguou… Sonnet 5 patch 2 days ago