dag.py
python
sha256:ac29b8ba9021514a03ab2782d92bf67671f0efa5b3b70d46f7598c5d4e923378
docs: record muse#74 Phase 4 live verification (SCR_13, SCR_14)
Sonnet 5
13 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
2 commits
sha256:ac29b8ba9021514a03ab2782d92bf67671f0efa5b3b70d46f7598c5d4e923378
docs: record muse#74 Phase 4 live verification (SCR_13, SCR_14)
Sonnet 5
13 days ago
sha256:2562dffa0a0822ac1bdea854f9b267843c6ce95b497a9dc5c55837c80a3ebd0a
feat: domain_command_registry — Phase 1 of muse#74 (SCR_01-03)
Sonnet 4.6
patch
13 days ago