1  """
  2  Edge Drawer for Sovereign OS
  3  
  4  Manages graph edges - creating, updating, and removing connections
  5  between nodes based on detected resonance and thread state.
  6  
  7  Works with both GRAPH-STATE.md (the God Database) and 
  8  LIVE-GRAPH.md (Mermaid visualization).
  9  """
 10  
 11  import re
 12  import logging
 13  from pathlib import Path
 14  from dataclasses import dataclass
 15  from typing import Dict, List, Optional, Set, Tuple
 16  from datetime import datetime
 17  from enum import Enum
 18  
 19  logger = logging.getLogger(__name__)
 20  
 21  
 22  class EdgeType(Enum):
 23      """Types of edges in the knowledge graph."""
 24      DEPENDS_ON = "depends_on"      # A requires B
 25      ENABLES = "enables"            # A makes B possible
 26      BLOCKS = "blocks"              # A prevents B
 27      RESONATES_WITH = "resonates"   # A and B share patterns
 28      CONVERGES_WITH = "converges"   # A and B reaching same conclusion
 29      DIVERGES_FROM = "diverges"     # A and B contradict
 30  
 31  
 32  @dataclass
 33  class Edge:
 34      """An edge in the knowledge graph."""
 35      
 36      source: str
 37      target: str
 38      edge_type: EdgeType
 39      pattern: Optional[str] = None  # For resonance edges
 40      created: Optional[datetime] = None
 41      weight: float = 1.0
 42      
 43      def to_mermaid(self) -> str:
 44          """Convert to Mermaid edge syntax."""
 45          source_safe = re.sub(r"[^a-zA-Z0-9_]", "_", self.source)
 46          target_safe = re.sub(r"[^a-zA-Z0-9_]", "_", self.target)
 47          
 48          if self.edge_type == EdgeType.RESONATES_WITH:
 49              return f"    {source_safe} -.-|{self.pattern or 'resonates'}| {target_safe}"
 50          elif self.edge_type == EdgeType.BLOCKS:
 51              return f"    {source_safe} -->|blocks| {target_safe}"
 52          elif self.edge_type == EdgeType.DIVERGES_FROM:
 53              return f"    {source_safe} x--x|diverges| {target_safe}"
 54          elif self.edge_type == EdgeType.CONVERGES_WITH:
 55              return f"    {source_safe} <-->|converges| {target_safe}"
 56          else:
 57              label = self.edge_type.value.replace("_", " ")
 58              return f"    {source_safe} -->|{label}| {target_safe}"
 59      
 60      def to_markdown_row(self) -> str:
 61          """Convert to markdown table row."""
 62          return (
 63              f"| {self.source} | {self.edge_type.value} | {self.target} | "
 64              f"{self.weight:.2f} | {self.pattern or '-'} |"
 65          )
 66  
 67  
 68  class EdgeDrawer:
 69      """
 70      Manages edges in the Sovereign OS knowledge graph.
 71      
 72      Responsibilities:
 73      - Track all edges between nodes
 74      - Add edges when resonance detected
 75      - Update GRAPH-STATE.md with new edges
 76      - Update LIVE-GRAPH.md Mermaid diagram
 77      """
 78      
 79      def __init__(self, sessions_dir: Path):
 80          self.sessions_dir = Path(sessions_dir)
 81          self._edges: Dict[Tuple[str, str], Edge] = {}
 82          self._pending_updates: List[Edge] = []
 83      
 84      def add_edge(self, edge: Edge) -> bool:
 85          """
 86          Add or update an edge.
 87          
 88          Returns True if edge was new, False if updated existing.
 89          """
 90          key = (edge.source, edge.target)
 91          
 92          is_new = key not in self._edges
 93          self._edges[key] = edge
 94          self._pending_updates.append(edge)
 95          
 96          if is_new:
 97              logger.info(f"New edge: {edge.source} --{edge.edge_type.value}--> {edge.target}")
 98          else:
 99              logger.debug(f"Updated edge: {edge.source} --{edge.edge_type.value}--> {edge.target}")
100          
101          return is_new
102      
103      def add_resonance_edge(
104          self, 
105          thread1: str, 
106          thread2: str, 
107          pattern: str,
108          strength: float = 0.8
109      ) -> Edge:
110          """Convenience method to add a resonance edge."""
111          edge = Edge(
112              source=thread1,
113              target=thread2,
114              edge_type=EdgeType.RESONATES_WITH,
115              pattern=pattern,
116              created=datetime.now(),
117              weight=strength
118          )
119          self.add_edge(edge)
120          return edge
121      
122      def add_dependency_edge(self, dependent: str, dependency: str) -> Edge:
123          """Convenience method to add a dependency edge."""
124          edge = Edge(
125              source=dependent,
126              target=dependency,
127              edge_type=EdgeType.DEPENDS_ON,
128              created=datetime.now()
129          )
130          self.add_edge(edge)
131          return edge
132      
133      def add_blocking_edge(self, blocker: str, blocked: str) -> Edge:
134          """Convenience method to add a blocking edge."""
135          edge = Edge(
136              source=blocker,
137              target=blocked,
138              edge_type=EdgeType.BLOCKS,
139              created=datetime.now()
140          )
141          self.add_edge(edge)
142          return edge
143      
144      def remove_edge(self, source: str, target: str) -> bool:
145          """Remove an edge. Returns True if edge existed."""
146          key = (source, target)
147          if key in self._edges:
148              del self._edges[key]
149              logger.info(f"Removed edge: {source} --> {target}")
150              return True
151          return False
152      
153      def get_edges_for_node(self, node_id: str) -> List[Edge]:
154          """Get all edges involving a node."""
155          return [
156              edge for edge in self._edges.values()
157              if edge.source == node_id or edge.target == node_id
158          ]
159      
160      def get_resonance_edges(self) -> List[Edge]:
161          """Get all resonance edges."""
162          return [
163              edge for edge in self._edges.values()
164              if edge.edge_type == EdgeType.RESONATES_WITH
165          ]
166      
167      def flush_to_graph_state(self) -> None:
168          """
169          Flush pending edge updates to GRAPH-STATE.md.
170          
171          Appends new edges to the Recent Graph Changes section.
172          """
173          if not self._pending_updates:
174              return
175          
176          graph_state_path = self.sessions_dir / "GRAPH-STATE.md"
177          
178          if not graph_state_path.exists():
179              logger.warning("GRAPH-STATE.md not found, cannot flush edges")
180              return
181          
182          content = graph_state_path.read_text(encoding="utf-8")
183          
184          # Find "Recent Graph Changes" section
185          changes_section = re.search(
186              r"(## Recent Graph Changes.*?\n\|.*?\n\|.*?\n)(.*?)(?=\n---|\Z)",
187              content,
188              re.DOTALL
189          )
190          
191          if changes_section:
192              header = changes_section.group(1)
193              existing = changes_section.group(2)
194              
195              # Add new entries
196              now = datetime.now().strftime("%H:%M")
197              new_entries = []
198              for edge in self._pending_updates:
199                  new_entries.append(
200                      f"| {now} | Added edge | {edge.source} → {edge.target} |"
201                  )
202              
203              updated_section = header + "\n".join(new_entries) + "\n" + existing
204              content = content[:changes_section.start()] + updated_section + content[changes_section.end():]
205              
206              graph_state_path.write_text(content, encoding="utf-8")
207              logger.info(f"Flushed {len(self._pending_updates)} edge updates to GRAPH-STATE.md")
208          
209          self._pending_updates.clear()
210      
211      def render_mermaid_edges(self) -> str:
212          """Render all edges as Mermaid syntax."""
213          lines = []
214          for edge in self._edges.values():
215              lines.append(edge.to_mermaid())
216          return "\n".join(lines)
217      
218      def update_live_graph(self) -> None:
219          """
220          Update LIVE-GRAPH.md with current edges.
221          
222          Finds the Mermaid code block and updates the edges section.
223          """
224          live_graph_path = self.sessions_dir / "LIVE-GRAPH.md"
225          
226          if not live_graph_path.exists():
227              logger.warning("LIVE-GRAPH.md not found, cannot update edges")
228              return
229          
230          content = live_graph_path.read_text(encoding="utf-8")
231          
232          # Find the mermaid block
233          mermaid_match = re.search(
234              r"(```mermaid\n.*?)(    %% Edges.*?)(```)",
235              content,
236              re.DOTALL
237          )
238          
239          if mermaid_match:
240              # Insert new edges before closing
241              before = mermaid_match.group(1)
242              
243              edges_section = "    %% Edges (auto-generated)\n"
244              edges_section += self.render_mermaid_edges() + "\n"
245              
246              content = before + edges_section + "```"
247              
248              # Update timestamp
249              content = re.sub(
250                  r"\*\*Updated:\*\* [^\n]+",
251                  f"**Updated:** {datetime.now().strftime('%Y-%m-%d %H:%M')}",
252                  content
253              )
254              
255              live_graph_path.write_text(content, encoding="utf-8")
256              logger.info("Updated LIVE-GRAPH.md with current edges")
257      
258      def get_statistics(self) -> Dict:
259          """Get edge statistics."""
260          by_type = {}
261          for edge in self._edges.values():
262              t = edge.edge_type.value
263              by_type[t] = by_type.get(t, 0) + 1
264          
265          return {
266              "total_edges": len(self._edges),
267              "by_type": by_type,
268              "pending_updates": len(self._pending_updates)
269          }
270  
271  
272  # CLI for testing
273  if __name__ == "__main__":
274      import sys
275      
276      sessions_dir = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("./sessions")
277      
278      drawer = EdgeDrawer(sessions_dir)
279      
280      # Add some test edges
281      drawer.add_resonance_edge(
282          "alignment-protocol",
283          "three-repo",
284          "boundary permeability",
285          0.85
286      )
287      
288      drawer.add_dependency_edge(
289          "permeability-contracts",
290          "three-repo"
291      )
292      
293      print("Edge Statistics:")
294      stats = drawer.get_statistics()
295      for key, value in stats.items():
296          print(f"  {key}: {value}")
297      
298      print("\nMermaid Output:")
299      print(drawer.render_mermaid_edges())