# generate GRAPH_REPORT.md - the human-readable audit trail from __future__ import annotations import re from datetime import date import networkx as nx def _safe_community_name(label: str) -> str: """Mirrors export.safe_name so community hub filenames and report wikilinks always agree.""" cleaned = re.sub(r'[\\/*?:"<>|#^[\]]', "", label.replace("\r\n", " ").replace("\r", " ").replace("\n", " ")).strip() cleaned = re.sub(r"\.(md|mdx|markdown)$", "", cleaned, flags=re.IGNORECASE) return cleaned or "unnamed" def load_learning_for_report(graph_path) -> dict | None: """Assemble the report's work-memory inputs from sibling artifacts. Reads the ``.graphify_learning.json`` overlay (preferred sources) next to ``graph_path`` and re-aggregates the memory docs for the query-scoped dead-ends. Best-effort: returns None if neither is available, so the report simply omits the section. Never raises. """ from pathlib import Path as _Path try: gp = _Path(graph_path) from graphify.reflect import load_learning_overlay, load_memory_docs, aggregate_lessons overlay = load_learning_overlay(gp) dead_ends: list[dict] = [] mem = gp.parent / "memory" if mem.is_dir(): agg = aggregate_lessons(load_memory_docs(mem)) dead_ends = agg.get("dead_ends", []) if not overlay and not dead_ends: return None return {"overlay": overlay, "dead_ends": dead_ends} except Exception: return None def _learning_section(lines: list, learning: dict | None, top_n: int = 10) -> None: """Append the ``## Work-memory lessons`` section, or nothing when empty.""" if not learning: return overlay = learning.get("overlay") or {} dead_ends = learning.get("dead_ends") or [] preferred = [ (nid, e) for nid, e in overlay.items() if isinstance(e, dict) and e.get("status") == "preferred" ] # Most-corroborated first (uses desc), then by score, then id for stability. preferred.sort(key=lambda kv: (-kv[1].get("uses", 0), -float(kv[1].get("score", 0) or 0), kv[0])) if not preferred and not dead_ends: return lines += ["", "## Work-memory lessons"] if preferred: lines += ["", "**Preferred sources** — corroborated by past sessions; start here."] for nid, e in preferred[:top_n]: label = e.get("label") or nid stale = " _(code changed — re-verify)_" if e.get("stale") else "" lines.append(f"- `{label}` ({e.get('uses', 0)}× useful, " f"score={e.get('score', 0)}){stale}") if dead_ends: lines += ["", "**Known dead ends** — questions that led nowhere; don't re-derive."] for d in dead_ends: nodes = ", ".join(f"`{n}`" for n in d.get("nodes", [])) lines.append(f"- \"{d.get('question', '')}\"" + (f" -> {nodes}" if nodes else "")) def generate( G: nx.Graph, communities: dict[int, list[str]], cohesion_scores: dict[int, float], community_labels: dict[int, str], god_node_list: list[dict], surprise_list: list[dict], detection_result: dict, token_cost: dict, root: str, suggested_questions: list[dict] | None = None, min_community_size: int = 3, built_at_commit: str | None = None, learning: dict | None = None, obsidian: bool = False, ) -> str: today = date.today().isoformat() # JSON deserialization produces string keys; normalize to int so .get(cid) works. if community_labels: community_labels = {int(k) if isinstance(k, str) else k: v for k, v in community_labels.items()} confidences = [d.get("confidence", "EXTRACTED") for _, _, d in G.edges(data=True)] total = len(confidences) or 1 ext_pct = round(confidences.count("EXTRACTED") / total * 100) inf_pct = round(confidences.count("INFERRED") / total * 100) amb_pct = round(confidences.count("AMBIGUOUS") / total * 100) inf_edges = [(u, v, d) for u, v, d in G.edges(data=True) if d.get("confidence") == "INFERRED"] inf_scores = [d.get("confidence_score", 0.5) for _, _, d in inf_edges] inf_avg = round(sum(inf_scores) / len(inf_scores), 2) if inf_scores else None lines = [ f"# Graph Report - {root} ({today})", "", "## Corpus Check", ] if detection_result.get("warning"): lines.append(f"- {detection_result['warning']}") else: lines += [ f"- {detection_result['total_files']} files · ~{detection_result['total_words']:,} words", "- Verdict: corpus is large enough that graph structure adds value.", ] from .analyze import _is_file_node as _ifn non_empty = {cid: nodes for cid, nodes in communities.items() if any(not _ifn(G, n) for n in nodes)} thin_count_summary = sum( 1 for nodes in communities.values() if 0 < sum(1 for n in nodes if not _ifn(G, n)) < min_community_size ) shown_count = len(communities) - thin_count_summary lines += [ "", "## Summary", f"- {G.number_of_nodes()} nodes · {G.number_of_edges()} edges · {len(communities)} communities" + (f" ({shown_count} shown, {thin_count_summary} thin omitted)" if thin_count_summary else ""), f"- Extraction: {ext_pct}% EXTRACTED · {inf_pct}% INFERRED · {amb_pct}% AMBIGUOUS" + (f" · INFERRED: {len(inf_edges)} edges (avg confidence: {inf_avg})" if inf_avg is not None else ""), f"- Token cost: {token_cost.get('input', 0):,} input · {token_cost.get('output', 0):,} output", ] if built_at_commit: lines += [ "", "## Graph Freshness", f"- Built from commit: `{built_at_commit[:8]}`", "- Run `git rev-parse HEAD` and compare to check if the graph is stale.", "- Run `graphify update .` after code changes (no API cost).", ] # Community hub navigation. The `_COMMUNITY_*.md` notes these wikilinks target # are only created by the opt-in `--obsidian` export, and the report is written # at build time (before any export runs), so emitting wikilinks by default left # every link dangling — polluting an Obsidian vault's graph view and rendering as # literal brackets everywhere else (#1712). Emit wikilinks only when the caller # signals Obsidian output; otherwise a plain list, which navigates nowhere-to-break. if non_empty: lines += ["", "## Community Hubs (Navigation)"] for cid in non_empty: label = community_labels.get(cid, f"Community {cid}") if obsidian: safe = _safe_community_name(label) lines.append(f"- [[_COMMUNITY_{safe}|{label}]]") else: lines.append(f"- {label}") lines += [ "", "## God Nodes (most connected - your core abstractions)", ] for i, node in enumerate(god_node_list, 1): lines.append(f"{i}. `{node['label']}` - {node['degree']} edges") lines += ["", "## Surprising Connections (you probably didn't know these)"] if surprise_list: for s in surprise_list: relation = s.get("relation", "related_to") note = s.get("note", "") files = s.get("source_files", ["", ""]) conf = s.get("confidence", "EXTRACTED") cscore = s.get("confidence_score") if conf == "INFERRED" and cscore is not None: conf_tag = f"INFERRED {cscore:.2f}" else: conf_tag = conf sem_tag = " [semantically similar]" if relation == "semantically_similar_to" else "" lines += [ f"- `{s['source']}` --{relation}--> `{s['target']}` [{conf_tag}]{sem_tag}", f" {files[0]} → {files[1]}" + (f" _{note}_" if note else ""), ] else: lines.append("- None detected - all connections are within the same source files.") # Circular imports surfaced from file-level dependency graph. Only meaningful # for code — a documents-only corpus has no imports, so the section is pure # noise there ("None detected" on every run). Emit it only when the graph # actually contains code (#1657). _has_code = any( d.get("file_type") == "code" for _, d in G.nodes(data=True) ) or any( d.get("relation") in ("imports", "imports_from") for *_e, d in G.edges(data=True) ) if _has_code: from .analyze import find_import_cycles cycles = find_import_cycles(G) lines += ["", "## Import Cycles"] if cycles: for c in cycles: cycle = c.get("cycle", []) length = c.get("length", len(cycle)) if not cycle: continue cycle_path = " -> ".join(cycle + [cycle[0]]) lines.append(f"- {length}-file cycle: `{cycle_path}`") else: lines.append("- None detected.") hyperedges = G.graph.get("hyperedges", []) if hyperedges: lines += ["", "## Hyperedges (group relationships)"] for h in hyperedges: node_labels = ", ".join(h.get("nodes", [])) conf = h.get("confidence", "INFERRED") cscore = h.get("confidence_score") conf_tag = f"{conf} {cscore:.2f}" if cscore is not None else conf lines.append(f"- **{h.get('label', h.get('id', ''))}** — {node_labels} [{conf_tag}]") lines += ["", f"## Communities ({len(communities)} total, {thin_count_summary} thin omitted)"] for cid, nodes in communities.items(): label = community_labels.get(cid, f"Community {cid}") score = cohesion_scores.get(cid, 0.0) # Filter method/function stubs from display - they're structural noise real_nodes = [n for n in nodes if not _ifn(G, n)] if not real_nodes: continue if len(real_nodes) < min_community_size: continue display = [G.nodes[n].get("label", n) for n in real_nodes[:8]] suffix = f" (+{len(real_nodes)-8} more)" if len(real_nodes) > 8 else "" lines += [ "", f"### Community {cid} - \"{label}\"", f"Cohesion: {score:.2f}", f"Nodes ({len(real_nodes)}): {', '.join(display)}{suffix}", ] ambiguous = [(u, v, d) for u, v, d in G.edges(data=True) if d.get("confidence") == "AMBIGUOUS"] if ambiguous: lines += ["", "## Ambiguous Edges - Review These"] for u, v, d in ambiguous: ul = G.nodes[u].get("label", u) vl = G.nodes[v].get("label", v) lines += [ f"- `{ul}` → `{vl}` [AMBIGUOUS]", f" {d.get('source_file', '')} · relation: {d.get('relation', 'unknown')}", ] # --- Gaps section --- from .analyze import _is_file_node, _is_concept_node isolated = [ n for n in G.nodes() if G.degree(n) <= 1 and not _is_file_node(G, n) and not _is_concept_node(G, n) and G.nodes[n].get("file_type") != "rationale" ] thin_communities = { cid: nodes for cid, nodes in communities.items() if 0 < sum(1 for n in nodes if not _is_file_node(G, n)) < 3 } gap_count = len(isolated) + len(thin_communities) if gap_count > 0 or amb_pct > 20: lines += ["", "## Knowledge Gaps"] if isolated: isolated_labels = [G.nodes[n].get("label", n) for n in isolated[:5]] suffix = f" (+{len(isolated)-5} more)" if len(isolated) > 5 else "" lines.append(f"- **{len(isolated)} isolated node(s):** {', '.join(f'`{l}`' for l in isolated_labels)}{suffix}") lines.append(" These have ≤1 connection - possible missing edges or undocumented components.") if thin_communities: lines.append(f"- **{len(thin_communities)} thin communities (<{min_community_size} nodes) omitted from report** — run `graphify query` to explore isolated nodes.") if amb_pct > 20: lines.append(f"- **High ambiguity: {amb_pct}% of edges are AMBIGUOUS.** Review the Ambiguous Edges section above.") # --- Work-memory lessons (derived overlay) --- # Preferred sources come from the .graphify_learning.json sidecar; the # query-scoped dead-ends come from the reflect aggregate. Section omitted # entirely when neither is present, so a graph with no work-memory is # byte-identical to the pre-feature report. _learning_section(lines, learning) if suggested_questions: lines += ["", "## Suggested Questions"] no_signal = len(suggested_questions) == 1 and suggested_questions[0].get("type") == "no_signal" if no_signal: lines.append(f"_{suggested_questions[0]['why']}_") else: lines.append("_Questions this graph is uniquely positioned to answer:_") lines.append("") for q in suggested_questions: if q.get("question"): lines.append(f"- **{q['question']}**") lines.append(f" _{q['why']}_") return "\n".join(lines)