dag.py
python
sha256:1a5c0bab2cb0d4c3eed597349ccaa433a4a6add8726890f2d6f577c17dc54972
docs: file follow-up issue for symlog filename-too-long (muse#61)
Sonnet 4.6
22 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:1a5c0bab2cb0d4c3eed597349ccaa433a4a6add8726890f2d6f577c17dc54972
docs: file follow-up issue for symlog filename-too-long (muse#61)
Sonnet 4.6
22 days ago