feat: add --directed flag to preserve edge direction (source→target)

README.md

@@ -171,6 +171,7 @@ When the user types `/graphify`, invoke the Skill tool with `skill: "graphify"`
 /graphify ./raw                    # run on a specific folder
 /graphify ./raw --mode deep        # more aggressive INFERRED edge extraction
 /graphify ./raw --update           # re-extract only changed files, merge into existing graph
+/graphify ./raw --directed          # build directed graph (preserves edge direction: source→target)
 /graphify ./raw --cluster-only     # rerun clustering on existing graph, no re-extraction
 /graphify ./raw --no-viz           # skip HTML, just produce report + JSON
 /graphify ./raw --obsidian                          # also generate Obsidian vault (opt-in)

graphify/analyze.py

@@ -468,7 +468,9 @@ def graph_diff(G_old: nx.Graph, G_new: nx.Graph) -> dict:
         for n in removed_node_ids
     ]
 
-    def edge_key(G: nx.Graph, u: str, v: str, data: dict) -> tuple:
+    def edge_key(G: nx.Graph | nx.DiGraph, u: str, v: str, data: dict) -> tuple:
+        if G.is_directed():
+            return (u, v, data.get("relation", ""))
         return (min(u, v), max(u, v), data.get("relation", ""))
 
     old_edge_keys = {

graphify/build.py

@@ -26,13 +26,18 @@
 from .validate import validate_extraction
 
 
-def build_from_json(extraction: dict) -> nx.Graph:
+def build_from_json(extraction: dict, *, directed: bool = False) -> nx.Graph | nx.DiGraph:
+    """Build a NetworkX graph from an extraction dict.
+
+    directed=True produces a DiGraph that preserves edge direction (source→target).
+    directed=False (default) produces an undirected Graph for backward compatibility.
+    """
     errors = validate_extraction(extraction)
     # Dangling edges (stdlib/external imports) are expected - only warn about real schema errors.
     real_errors = [e for e in errors if "does not match any node id" not in e]
     if real_errors:
         print(f"[graphify] Extraction warning ({len(real_errors)} issues): {real_errors[0]}", file=sys.stderr)
-    G = nx.Graph()
+    G = nx.DiGraph() if directed else nx.Graph()
     for node in extraction.get("nodes", []):
         G.add_node(node["id"], **{k: v for k, v in node.items() if k != "id"})
     node_set = set(G.nodes())
@@ -41,8 +46,7 @@ def build_from_json(extraction: dict) -> nx.Graph:
         if src not in node_set or tgt not in node_set:
             continue  # skip edges to external/stdlib nodes - expected, not an error
         attrs = {k: v for k, v in edge.items() if k not in ("source", "target")}
-        # Preserve original edge direction - undirected graphs lose it otherwise,
-        # causing display functions to show edges backwards.
+        # _src/_tgt kept for backward compatibility with existing consumers.
         attrs["_src"] = src
         attrs["_tgt"] = tgt
         G.add_edge(src, tgt, **attrs)
@@ -52,9 +56,12 @@ def build_from_json(extraction: dict) -> nx.Graph:
     return G
 
 
-def build(extractions: list[dict]) -> nx.Graph:
+def build(extractions: list[dict], *, directed: bool = False) -> nx.Graph | nx.DiGraph:
     """Merge multiple extraction results into one graph.
 
+    directed=True produces a DiGraph that preserves edge direction (source→target).
+    directed=False (default) produces an undirected Graph for backward compatibility.
+
     Extractions are merged in order. For nodes with the same ID, the last
     extraction's attributes win (NetworkX add_node overwrites). Pass AST
     results before semantic results so semantic labels take precedence, or
@@ -67,4 +74,4 @@ def build(extractions: list[dict]) -> nx.Graph:
         combined["hyperedges"].extend(ext.get("hyperedges", []))
         combined["input_tokens"] += ext.get("input_tokens", 0)
         combined["output_tokens"] += ext.get("output_tokens", 0)
-    return build_from_json(combined)
+    return build_from_json(combined, directed=directed)

graphify/cluster.py

@@ -56,22 +56,28 @@ def _partition(G: nx.Graph) -> dict[str, int]:
 _MIN_SPLIT_SIZE = 10             # only split if community has at least this many nodes
 
 
-def cluster(G: nx.Graph) -> dict[int, list[str]]:
+def cluster(G: nx.Graph | nx.DiGraph) -> dict[int, list[str]]:
     """Run Leiden community detection. Returns {community_id: [node_ids]}.
 
     Community IDs are stable across runs: 0 = largest community after splitting.
     Oversized communities (> 25% of graph nodes, min 10) are split by running
     a second Leiden pass on the subgraph.
+
+    Accepts directed or undirected graphs. Directed graphs are converted to
+    undirected internally since Louvain/Leiden require undirected input.
     """
     if G.number_of_nodes() == 0:
         return {}
     if G.number_of_edges() == 0:
         return {i: [n] for i, n in enumerate(sorted(G.nodes))}
 
+    # Louvain/Leiden require undirected graphs
+    G_undirected = G.to_undirected() if G.is_directed() else G
+
     # Leiden warns and drops isolates - handle them separately
-    isolates = [n for n in G.nodes() if G.degree(n) == 0]
-    connected_nodes = [n for n in G.nodes() if G.degree(n) > 0]
-    connected = G.subgraph(connected_nodes)
+    isolates = [n for n in G_undirected.nodes() if G_undirected.degree(n) == 0]
+    connected_nodes = [n for n in G_undirected.nodes() if G_undirected.degree(n) > 0]
+    connected = G_undirected.subgraph(connected_nodes)
 
     raw: dict[int, list[str]] = {}
     if connected.number_of_nodes() > 0:
@@ -85,12 +91,12 @@ def cluster(G: nx.Graph) -> dict[int, list[str]]:
         raw[next_cid] = [node]
         next_cid += 1
 
-    # Split oversized communities
-    max_size = max(_MIN_SPLIT_SIZE, int(G.number_of_nodes() * _MAX_COMMUNITY_FRACTION))
+    # Split oversized communities (use undirected for Leiden compatibility)
+    max_size = max(_MIN_SPLIT_SIZE, int(G_undirected.number_of_nodes() * _MAX_COMMUNITY_FRACTION))
     final_communities: list[list[str]] = []
     for nodes in raw.values():
         if len(nodes) > max_size:
-            final_communities.extend(_split_community(G, nodes))
+            final_communities.extend(_split_community(G_undirected, nodes))
         else:
             final_communities.append(nodes)
 
@@ -117,14 +123,15 @@ def _split_community(G: nx.Graph, nodes: list[str]) -> list[list[str]]:
         return [sorted(nodes)]
 
 
-def cohesion_score(G: nx.Graph, community_nodes: list[str]) -> float:
+def cohesion_score(G: nx.Graph | nx.DiGraph, community_nodes: list[str]) -> float:
     """Ratio of actual intra-community edges to maximum possible."""
     n = len(community_nodes)
     if n <= 1:
         return 1.0
     subgraph = G.subgraph(community_nodes)
     actual = subgraph.number_of_edges()
-    possible = n * (n - 1) / 2
+    # Directed: max edges = n*(n-1); undirected: n*(n-1)/2
+    possible = n * (n - 1) if G.is_directed() else n * (n - 1) / 2
     return round(actual / possible, 2) if possible > 0 else 0.0
 
 

graphify/export.py

@@ -104,8 +104,9 @@ def _hyperedge_script(hyperedges_json: str) -> str:
 </script>"""
 
 
-def _html_script(nodes_json: str, edges_json: str, legend_json: str) -> str:
+def _html_script(nodes_json: str, edges_json: str, legend_json: str, *, directed: bool = False) -> str:
     return f"""<script>
+const GRAPH_IS_DIRECTED = {'true' if directed else 'false'};
 const RAW_NODES = {nodes_json};
 const RAW_EDGES = {edges_json};
 const LEGEND = {legend_json};
@@ -130,7 +131,7 @@ def _html_script(nodes_json: str, edges_json: str, legend_json: str) -> str:
   dashes: e.dashes,
   width: e.width,
   color: e.color,
-  arrows: {{ to: {{ enabled: true, scaleFactor: 0.5 }} }},
+  arrows: {{ to: {{ enabled: GRAPH_IS_DIRECTED, scaleFactor: 0.5 }} }},
 }})));
 
 const container = document.getElementById('graph');
@@ -425,7 +426,7 @@ def to_html(
   </div>
   <div id="stats">{stats}</div>
 </div>
-{_html_script(nodes_json, edges_json, legend_json)}
+{_html_script(nodes_json, edges_json, legend_json, directed=G.is_directed())}
 {_hyperedge_script(hyperedges_json)}
 </body>
 </html>"""

graphify/skill-claw.md

@@ -15,6 +15,7 @@ Turn any folder of files into a navigable knowledge graph with community detecti
 /graphify <path>                                      # full pipeline on specific path
 /graphify <path> --mode deep                          # thorough extraction, richer INFERRED edges
 /graphify <path> --update                             # incremental - re-extract only new/changed files
+/graphify <path> --directed                            # build directed graph (preserves edge direction: source→target)
 /graphify <path> --cluster-only                       # rerun clustering on existing graph
 /graphify <path> --no-viz                             # skip visualization, just report + JSON
 /graphify <path> --html                               # (HTML is generated by default - this flag is a no-op)
@@ -110,6 +111,8 @@ Then act on it:
 
 **Before starting:** note whether `--mode deep` was given. You must pass `DEEP_MODE=true` to every subagent in Step B2 if it was. Track this from the original invocation - do not lose it.
 
+Also note whether `--directed` was given. If it was, set `DIRECTED=True` and pass `directed=DIRECTED` to every `build_from_json()` call. Directed graphs preserve edge direction (source→target), producing arrows in the HTML visualization and `"directed": true` in graph.json.
+
 This step has two parts: **structural extraction** (deterministic, free) and **semantic extraction** (Claude, costs tokens).
 
 **Run Part A (AST) and Part B (semantic) in parallel. Dispatch all semantic subagents AND start AST extraction in the same message. Both can run simultaneously since they operate on different file types. Merge results in Part C as before.**
@@ -302,7 +305,7 @@ from pathlib import Path
 extraction = json.loads(Path('.graphify_extract.json').read_text())
 detection  = json.loads(Path('.graphify_detect.json').read_text())
 
-G = build_from_json(extraction)
+G = build_from_json(extraction, directed=DIRECTED)
 communities = cluster(G)
 cohesion = score_all(G, communities)
 tokens = {'input': extraction.get('input_tokens', 0), 'output': extraction.get('output_tokens', 0)}
@@ -355,7 +358,7 @@ extraction = json.loads(Path('.graphify_extract.json').read_text())
 detection  = json.loads(Path('.graphify_detect.json').read_text())
 analysis   = json.loads(Path('.graphify_analysis.json').read_text())
 
-G = build_from_json(extraction)
+G = build_from_json(extraction, directed=DIRECTED)
 communities = {int(k): v for k, v in analysis['communities'].items()}
 cohesion = {int(k): v for k, v in analysis['cohesion'].items()}
 tokens = {'input': extraction.get('input_tokens', 0), 'output': extraction.get('output_tokens', 0)}
@@ -393,7 +396,7 @@ extraction = json.loads(Path('.graphify_extract.json').read_text())
 analysis   = json.loads(Path('.graphify_analysis.json').read_text())
 labels_raw = json.loads(Path('.graphify_labels.json').read_text()) if Path('.graphify_labels.json').exists() else {}
 
-G = build_from_json(extraction)
+G = build_from_json(extraction, directed=DIRECTED)
 communities = {int(k): v for k, v in analysis['communities'].items()}
 cohesion = {int(k): v for k, v in analysis['cohesion'].items()}
 labels = {int(k): v for k, v in labels_raw.items()}
@@ -424,7 +427,7 @@ extraction = json.loads(Path('.graphify_extract.json').read_text())
 analysis   = json.loads(Path('.graphify_analysis.json').read_text())
 labels_raw = json.loads(Path('.graphify_labels.json').read_text()) if Path('.graphify_labels.json').exists() else {}
 
-G = build_from_json(extraction)
+G = build_from_json(extraction, directed=DIRECTED)
 communities = {int(k): v for k, v in analysis['communities'].items()}
 labels = {int(k): v for k, v in labels_raw.items()}
 
@@ -447,7 +450,7 @@ from graphify.build import build_from_json
 from graphify.export import to_cypher
 from pathlib import Path
 
-G = build_from_json(json.loads(Path('.graphify_extract.json').read_text()))
+G = build_from_json(json.loads(Path('.graphify_extract.json').read_text()), directed=DIRECTED)
 to_cypher(G, 'graphify-out/cypher.txt')
 print('cypher.txt written - import with: cypher-shell < graphify-out/cypher.txt')
 "
@@ -465,7 +468,7 @@ from pathlib import Path
 
 extraction = json.loads(Path('.graphify_extract.json').read_text())
 analysis   = json.loads(Path('.graphify_analysis.json').read_text())
-G = build_from_json(extraction)
+G = build_from_json(extraction, directed=DIRECTED)
 communities = {int(k): v for k, v in analysis['communities'].items()}
 
 result = push_to_neo4j(G, uri='NEO4J_URI', user='NEO4J_USER', password='NEO4J_PASSWORD', communities=communities)
@@ -488,7 +491,7 @@ extraction = json.loads(Path('.graphify_extract.json').read_text())
 analysis   = json.loads(Path('.graphify_analysis.json').read_text())
 labels_raw = json.loads(Path('.graphify_labels.json').read_text()) if Path('.graphify_labels.json').exists() else {}
 
-G = build_from_json(extraction)
+G = build_from_json(extraction, directed=DIRECTED)
 communities = {int(k): v for k, v in analysis['communities'].items()}
 labels = {int(k): v for k, v in labels_raw.items()}
 
@@ -509,7 +512,7 @@ from pathlib import Path
 extraction = json.loads(Path('.graphify_extract.json').read_text())
 analysis   = json.loads(Path('.graphify_analysis.json').read_text())
 
-G = build_from_json(extraction)
+G = build_from_json(extraction, directed=DIRECTED)
 communities = {int(k): v for k, v in analysis['communities'].items()}
 
 to_graphml(G, communities, 'graphify-out/graph.graphml')
@@ -685,7 +688,7 @@ G_existing = json_graph.node_link_graph(existing_data, edges='links')
 
 # Load new extraction
 new_extraction = json.loads(Path('.graphify_extract.json').read_text())
-G_new = build_from_json(new_extraction)
+G_new = build_from_json(new_extraction, directed=DIRECTED)
 
 # Merge: new nodes/edges into existing graph
 G_existing.update(G_new)
@@ -709,7 +712,7 @@ from pathlib import Path
 # Load old graph (before update) from backup written before merge
 old_data = json.loads(Path('.graphify_old.json').read_text()) if Path('.graphify_old.json').exists() else None
 new_extract = json.loads(Path('.graphify_extract.json').read_text())
-G_new = build_from_json(new_extract)
+G_new = build_from_json(new_extract, directed=DIRECTED)
 
 if old_data:
     G_old = json_graph.node_link_graph(old_data, edges='links')