1  """
  2  Live Graph Renderer for Sovereign OS
  3  
  4  Generates and updates Mermaid graphs in real-time based on
  5  GRAPH-STATE.md and thread activity.
  6  
  7  The output is a Mermaid-compatible markdown file that Obsidian
  8  can render as a live updating graph.
  9  """
 10  
 11  import re
 12  import logging
 13  from pathlib import Path
 14  from dataclasses import dataclass, field
 15  from typing import Dict, List, Set, Optional, Tuple
 16  from datetime import datetime
 17  
 18  logger = logging.getLogger(__name__)
 19  
 20  
 21  @dataclass
 22  class GraphNode:
 23      """A node in the knowledge graph."""
 24      
 25      id: str
 26      label: str
 27      node_type: str  # axiom, protocol, thread, concept, artifact
 28      connections: Set[str] = field(default_factory=set)
 29      weight: float = 1.0
 30      created: Optional[datetime] = None
 31      
 32      @property
 33      def style(self) -> str:
 34          """Get Mermaid style based on node type."""
 35          styles = {
 36              "axiom": "fill:#FF6B6B",      # Red
 37              "protocol": "fill:#4ECDC4",    # Teal
 38              "thread": "fill:#95E1D3",      # Green
 39              "concept": "fill:#F38181",     # Pink
 40              "artifact": "fill:#FCE38A",    # Yellow
 41          }
 42          return styles.get(self.node_type, "fill:#DDD")
 43  
 44  
 45  @dataclass
 46  class GraphEdge:
 47      """An edge in the knowledge graph."""
 48      
 49      source: str
 50      target: str
 51      edge_type: str  # depends_on, enables, blocks, resonates_with
 52      weight: float = 1.0
 53      
 54      @property
 55      def style(self) -> str:
 56          """Get Mermaid edge style."""
 57          if self.edge_type == "blocks":
 58              return "stroke:#FF0000,stroke-width:2px"
 59          elif self.edge_type == "resonates_with":
 60              return "stroke:#00FF00,stroke-width:2px,stroke-dasharray: 5 5"
 61          elif self.edge_type == "depends_on":
 62              return "stroke:#0066FF,stroke-width:2px"
 63          return ""
 64  
 65  
 66  class LiveGraphRenderer:
 67      """
 68      Renders knowledge graphs as Mermaid markdown.
 69      
 70      Watches GRAPH-STATE.md and produces LIVE-GRAPH.md with
 71      real-time updates.
 72      """
 73      
 74      def __init__(self, sessions_dir: Path):
 75          self.sessions_dir = Path(sessions_dir)
 76          self._nodes: Dict[str, GraphNode] = {}
 77          self._edges: List[GraphEdge] = []
 78      
 79      def load_from_graph_state(self) -> None:
 80          """Load graph from GRAPH-STATE.md."""
 81          graph_state_path = self.sessions_dir / "GRAPH-STATE.md"
 82          
 83          if not graph_state_path.exists():
 84              logger.warning(f"GRAPH-STATE.md not found at {graph_state_path}")
 85              return
 86          
 87          content = graph_state_path.read_text(encoding="utf-8")
 88          self._parse_graph_state(content)
 89      
 90      def _parse_graph_state(self, content: str) -> None:
 91          """Parse GRAPH-STATE.md and extract nodes/edges."""
 92          # Extract nodes from tables
 93          self._extract_nodes_from_tables(content)
 94          
 95          # Extract edges from dependency map
 96          self._extract_edges_from_content(content)
 97      
 98      def _extract_nodes_from_tables(self, content: str) -> None:
 99          """Extract nodes from markdown tables in GRAPH-STATE."""
100          # Pattern for table rows
101          table_row = re.compile(r"\|\s*`([^`]+)`\s*\|\s*\[\[([^\]]+)\]\]")
102          
103          for match in table_row.finditer(content):
104              node_id = match.group(1)
105              label = match.group(2)
106              
107              # Determine type from ID prefix
108              if node_id.startswith("axiom"):
109                  node_type = "axiom"
110              elif node_id.startswith("proto"):
111                  node_type = "protocol"
112              elif node_id.startswith("thread"):
113                  node_type = "thread"
114              elif node_id.startswith("ui"):
115                  node_type = "artifact"
116              else:
117                  node_type = "concept"
118              
119              self._nodes[node_id] = GraphNode(
120                  id=node_id,
121                  label=label,
122                  node_type=node_type,
123                  created=datetime.now()
124              )
125      
126      def _extract_edges_from_content(self, content: str) -> None:
127          """Extract edges from content patterns."""
128          # Look for dependency patterns like "proto-017 → proto-018"
129          arrow_pattern = re.compile(r"`([^`]+)`\s*[→\-]+>\s*`([^`]+)`")
130          
131          for match in arrow_pattern.finditer(content):
132              source = match.group(1)
133              target = match.group(2)
134              
135              self._edges.append(GraphEdge(
136                  source=source,
137                  target=target,
138                  edge_type="depends_on"
139              ))
140              
141              # Update node connections
142              if source in self._nodes:
143                  self._nodes[source].connections.add(target)
144      
145      def add_node(self, node: GraphNode) -> None:
146          """Add or update a node."""
147          self._nodes[node.id] = node
148      
149      def add_edge(self, edge: GraphEdge) -> None:
150          """Add an edge."""
151          self._edges.append(edge)
152          
153          # Update node connections
154          if edge.source in self._nodes:
155              self._nodes[edge.source].connections.add(edge.target)
156      
157      def add_resonance_edge(self, thread1: str, thread2: str, pattern: str) -> None:
158          """Add a resonance edge between threads."""
159          self._edges.append(GraphEdge(
160              source=f"thread/{thread1}",
161              target=f"thread/{thread2}",
162              edge_type="resonates_with",
163              weight=0.8
164          ))
165      
166      def render_mermaid(self) -> str:
167          """Render the graph as Mermaid markdown."""
168          lines = ["```mermaid", "graph TD"]
169          
170          # Group nodes by type
171          by_type: Dict[str, List[GraphNode]] = {}
172          for node in self._nodes.values():
173              if node.node_type not in by_type:
174                  by_type[node.node_type] = []
175              by_type[node.node_type].append(node)
176          
177          # Render subgraphs by type
178          type_labels = {
179              "axiom": "AXIOMS",
180              "protocol": "PROTOCOLS",
181              "thread": "THREADS",
182              "concept": "CONCEPTS",
183              "artifact": "ARTIFACTS"
184          }
185          
186          for node_type, nodes in by_type.items():
187              if nodes:
188                  label = type_labels.get(node_type, node_type.upper())
189                  lines.append(f"    subgraph {label}")
190                  for node in nodes:
191                      safe_id = self._safe_mermaid_id(node.id)
192                      safe_label = node.label.replace('"', "'")
193                      lines.append(f'        {safe_id}["{safe_label}"]')
194                  lines.append("    end")
195          
196          lines.append("")
197          
198          # Render edges
199          for edge in self._edges:
200              source_id = self._safe_mermaid_id(edge.source)
201              target_id = self._safe_mermaid_id(edge.target)
202              
203              if edge.edge_type == "resonates_with":
204                  lines.append(f"    {source_id} -.-|resonates| {target_id}")
205              elif edge.edge_type == "blocks":
206                  lines.append(f"    {source_id} -->|blocks| {target_id}")
207              else:
208                  lines.append(f"    {source_id} --> {target_id}")
209          
210          lines.append("")
211          
212          # Add styles
213          lines.append("    %% Styling")
214          for node in self._nodes.values():
215              safe_id = self._safe_mermaid_id(node.id)
216              lines.append(f"    style {safe_id} {node.style}")
217          
218          lines.append("```")
219          
220          return "\n".join(lines)
221      
222      def _safe_mermaid_id(self, node_id: str) -> str:
223          """Convert node ID to safe Mermaid identifier."""
224          # Replace characters that Mermaid doesn't like
225          safe = re.sub(r"[^a-zA-Z0-9_]", "_", node_id)
226          return safe
227      
228      def render_full_markdown(self) -> str:
229          """Render complete LIVE-GRAPH.md file."""
230          now = datetime.now()
231          mermaid = self.render_mermaid()
232          
233          # Build node registry table
234          node_rows = []
235          for node in sorted(self._nodes.values(), key=lambda n: n.id):
236              node_rows.append(
237                  f"| {node.id} | {node.node_type} | {len(node.connections)} |"
238              )
239          node_table = "\n".join(node_rows) if node_rows else "| (no nodes) | - | - |"
240          
241          # Build edge count
242          edge_count = len(self._edges)
243          
244          return f"""# Live Graph - Real-Time Visualization
245  
246  *Auto-generated by Mission Control*
247  
248  **Updated:** {now.strftime('%Y-%m-%d %H:%M')}
249  **Nodes:** {len(self._nodes)}
250  **Edges:** {edge_count}
251  
252  ---
253  
254  ## Live Mermaid Graph
255  
256  {mermaid}
257  
258  ---
259  
260  ## Node Registry
261  
262  | ID | Type | Connections |
263  |----|------|-------------|
264  {node_table}
265  
266  ---
267  
268  ## How This Updates
269  
270  This file is regenerated by Mission Control when:
271  - New nodes are created in any thread
272  - New edges (dependencies, resonances) are detected
273  - FO checkpoints update thread state
274  
275  Open in Obsidian to see live graph rendering.
276  
277  ---
278  
279  *Live Graph | Auto-generated by Mission Control*
280  """
281      
282      def write(self, output_path: Optional[Path] = None) -> Path:
283          """Write the rendered graph to file."""
284          if output_path is None:
285              output_path = self.sessions_dir / "LIVE-GRAPH.md"
286          
287          content = self.render_full_markdown()
288          output_path.write_text(content, encoding="utf-8")
289          
290          logger.info(f"Live graph written to {output_path}")
291          return output_path
292  
293  
294  # CLI for testing
295  if __name__ == "__main__":
296      import sys
297      
298      sessions_dir = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("./sessions")
299      
300      renderer = LiveGraphRenderer(sessions_dir)
301      renderer.load_from_graph_state()
302      
303      # Add some test nodes
304      renderer.add_node(GraphNode(
305          id="test-node-1",
306          label="Test Node 1",
307          node_type="concept"
308      ))
309      
310      output = renderer.write()
311      print(f"Graph written to {output}")