Complete REUSE-WP-0005: registry federation and relation graphs

Add federation manifest and schema, federation compose and graph CLI commands,
relation cycle/reference checks, federated index and Mermaid graph artifacts,
RegistryFederation guide, and CI validation updates.

reuse_surface/graph.py

@@ -0,0 +1,153 @@
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from reuse_surface.federation import FEDERATED_INDEX_PATH, compose_federated_index
+from reuse_surface.registry import ROOT, load_index, parse_front_matter
+
+GRAPH_PATH = ROOT / "docs" / "graph" / "capability-graph.mmd"
+RELATION_TYPES = [
+    "depends_on",
+    "supports",
+    "used_by",
+    "related_to",
+    "specializes",
+    "generalizes",
+    "replaces",
+    "wraps",
+]
+
+
+@dataclass
+class RelationEdge:
+    source_id: str
+    target_id: str
+    relation_type: str
+
+
+def _node_id(capability_id: str) -> str:
+    return re.sub(r"[^a-zA-Z0-9_]", "_", capability_id)
+
+
+def _load_local_relations() -> dict[str, dict[str, list[str]]]:
+    index = load_index()
+    relations_by_id: dict[str, dict[str, list[str]]] = {}
+    for item in index.get("capabilities", []):
+        path = ROOT / item["path"]
+        if not path.exists():
+            continue
+        entry = parse_front_matter(path)
+        relations = entry.get("relations") or {}
+        relations_by_id[entry["id"]] = {
+            relation_type: list(targets)
+            for relation_type, targets in relations.items()
+            if isinstance(targets, list)
+        }
+    return relations_by_id
+
+
+def _known_ids() -> set[str]:
+    if FEDERATED_INDEX_PATH.exists():
+        import yaml
+
+        data = yaml.safe_load(FEDERATED_INDEX_PATH.read_text(encoding="utf-8"))
+    else:
+        data, _ = compose_federated_index()
+    return {item["id"] for item in data.get("capabilities", [])}
+
+
+def collect_edges() -> list[RelationEdge]:
+    relations_by_id = _load_local_relations()
+    edges: list[RelationEdge] = []
+    for source_id, relation_map in relations_by_id.items():
+        for relation_type, targets in relation_map.items():
+            for target_id in targets:
+                edges.append(
+                    RelationEdge(
+                        source_id=source_id,
+                        target_id=target_id,
+                        relation_type=relation_type,
+                    )
+                )
+    return edges
+
+
+def find_depends_on_cycles() -> list[list[str]]:
+    relations_by_id = _load_local_relations()
+    graph: dict[str, list[str]] = {
+        cap_id: list(relation_map.get("depends_on", []))
+        for cap_id, relation_map in relations_by_id.items()
+    }
+    cycles: list[list[str]] = []
+    visited: set[str] = set()
+    stack: set[str] = set()
+    path: list[str] = []
+
+    def dfs(node: str) -> None:
+        visited.add(node)
+        stack.add(node)
+        path.append(node)
+        for neighbor in graph.get(node, []):
+            if neighbor not in visited:
+                dfs(neighbor)
+            elif neighbor in stack:
+                start = path.index(neighbor)
+                cycles.append(path[start:] + [neighbor])
+        path.pop()
+        stack.remove(node)
+
+    for node in graph:
+        if node not in visited:
+            dfs(node)
+    return cycles
+
+
+def find_broken_references(known: set[str] | None = None) -> list[str]:
+    known = known or _known_ids()
+    warnings: list[str] = []
+    for edge in collect_edges():
+        if edge.target_id not in known:
+            warnings.append(
+                f"broken relation: {edge.source_id} "
+                f"{edge.relation_type} -> {edge.target_id}"
+            )
+    return warnings
+
+
+def check_relations() -> list[str]:
+    warnings: list[str] = []
+    for cycle in find_depends_on_cycles():
+        warnings.append(f"depends_on cycle: {' -> '.join(cycle)}")
+    warnings.extend(find_broken_references())
+    return warnings
+
+
+def _node_labels() -> dict[str, str]:
+    index = load_index()
+    labels: dict[str, str] = {}
+    for item in index.get("capabilities", []):
+        labels[item["id"]] = f"{item['id']}<br/>{item['vector']}"
+    return labels
+
+
+def render_mermaid() -> str:
+    labels = _node_labels()
+    edges = collect_edges()
+    lines = ["graph LR"]
+    for cap_id, label in sorted(labels.items()):
+        lines.append(f'  {_node_id(cap_id)}["{label}"]')
+    for edge in edges:
+        lines.append(
+            f"  {_node_id(edge.source_id)} -->|{edge.relation_type}| {_node_id(edge.target_id)}"
+        )
+    return "\n".join(lines) + "\n"
+
+
+def write_graph(path: Path | None = None) -> Path:
+    target = path or GRAPH_PATH
+    target.parent.mkdir(parents=True, exist_ok=True)
+    target.write_text(render_mermaid(), encoding="utf-8")
+    return target