1  #!/usr/bin/env python3
  2  """
  3  Live Escalator - Watch conversation and auto-escalate high-resonance insights
  4  
  5  This is the pipeline that was missing:
  6  [Statement] → semantic_resonance.py → threshold check → RESONANCE-ALERTS/
  7  
  8  Building this to understand what "automatic escalation" means.
  9  
 10  Usage:
 11      # Watch a transcript file
 12      python scripts/live_escalator.py --watch /path/to/transcript.jsonl
 13  
 14      # Process a single statement
 15      echo "building is cheap" | python scripts/live_escalator.py
 16  
 17      # Run as daemon
 18      python scripts/live_escalator.py --daemon --transcript /path/to/session.jsonl
 19  """
 20  
 21  import os
 22  import re
 23  import sys
 24  import json
 25  import time
 26  from pathlib import Path
 27  from datetime import datetime
 28  from typing import Optional, List, Dict
 29  import argparse
 30  
 31  # Import semantic resonance if available
 32  sys.path.insert(0, str(Path(__file__).parent))
 33  
 34  try:
 35      from semantic_resonance import SemanticResonanceEngine
 36      HAS_SEMANTIC = True
 37  except ImportError:
 38      HAS_SEMANTIC = False
 39      print("Warning: semantic_resonance not available, using term-based fallback")
 40  
 41  
 42  # Fallback term-based resonance
 43  AXIOM_TERMS = {
 44      'A0': {'boundary', 'boundaries', 'markov', 'blanket', 'structure', 'inside', 'outside', 'interface'},
 45      'A1': {'integration', 'connect', 'binding', 'relation', 'isolation', 'together'},
 46      'A2': {'life', 'death', 'alive', 'primitive', 'calcified', 'motion', 'fixation', 'real'},
 47      'A3': {'pole', 'poles', 'navigation', 'tension', 'dynamic', 'balance', 'between'},
 48      'A4': {'ruin', 'survival', 'cheap', 'rebuild', 'rewrite', 'catastrophe', 'ergodic', 'terminal'},
 49  }
 50  
 51  
 52  def extract_user_statements(transcript_path: Path, after_line: int = 0) -> List[Dict]:
 53      """Extract user statements from Claude transcript."""
 54      statements = []
 55  
 56      try:
 57          with open(transcript_path, 'r') as f:
 58              for i, line in enumerate(f):
 59                  if i < after_line:
 60                      continue
 61                  try:
 62                      entry = json.loads(line)
 63                      if entry.get('type') == 'user':
 64                          content = entry.get('message', {}).get('content', '')
 65                          if isinstance(content, str) and len(content) > 20:
 66                              statements.append({
 67                                  'line': i,
 68                                  'text': content[:500],  # Truncate
 69                                  'timestamp': entry.get('timestamp', ''),
 70                              })
 71                  except json.JSONDecodeError:
 72                      continue
 73      except Exception as e:
 74          print(f"Error reading transcript: {e}", file=sys.stderr)
 75  
 76      return statements
 77  
 78  
 79  def calculate_resonance_fallback(text: str) -> Dict[str, float]:
 80      """Fallback term-based resonance."""
 81      text_lower = text.lower()
 82      words = set(re.findall(r'\w+', text_lower))
 83  
 84      scores = {}
 85      for axiom, terms in AXIOM_TERMS.items():
 86          overlap = len(words & terms)
 87          scores[axiom] = min(1.0, overlap * 0.2)
 88  
 89      return scores
 90  
 91  
 92  def calculate_resonance(text: str) -> Dict[str, float]:
 93      """Calculate resonance using semantic engine or fallback."""
 94      if HAS_SEMANTIC:
 95          try:
 96              engine = SemanticResonanceEngine()
 97              result = engine.analyze(text)
 98              return {r['axiom']: r['score'] for r in result.get('resonances', [])}
 99          except Exception:
100              pass
101  
102      return calculate_resonance_fallback(text)
103  
104  
105  def create_alert(
106      text: str,
107      resonances: Dict[str, float],
108      alerts_dir: Path,
109      threshold: float = 0.4
110  ) -> Optional[Path]:
111      """Create alert if resonance exceeds threshold."""
112      max_axiom = max(resonances, key=resonances.get)
113      max_score = resonances[max_axiom]
114  
115      if max_score < threshold:
116          return None
117  
118      timestamp = datetime.now().strftime('%Y%m%d-%H%M%S')
119      filename = f"{timestamp}-auto-escalation-{max_axiom.lower()}.md"
120      filepath = alerts_dir / filename
121  
122      lines = [
123          f"# Auto-Escalation - {max_axiom}",
124          "",
125          f"**Type:** HIGH_RESONANCE_STATEMENT",
126          f"**Detected:** {datetime.now().isoformat()}",
127          f"**Axiom:** {max_axiom}",
128          f"**Score:** {max_score:.0%}",
129          "",
130          "---",
131          "",
132          "## Statement",
133          "",
134          f"> {text[:300]}{'...' if len(text) > 300 else ''}",
135          "",
136          "## All Resonances",
137          "",
138      ]
139  
140      for axiom in ['A0', 'A1', 'A2', 'A3', 'A4']:
141          score = resonances.get(axiom, 0)
142          bar = '█' * int(score * 10)
143          lines.append(f"- {axiom}: [{score:>5.0%}] {bar}")
144  
145      lines.extend([
146          "",
147          "## Action Required",
148          "",
149          f"This statement has high resonance ({max_score:.0%}) with {max_axiom}.",
150          "Consider:",
151          "- Adding to canonical principles",
152          "- Integrating into relevant documentation",
153          "- Extracting as derived principle",
154          "",
155          "---",
156          "",
157          f"*Auto-generated by Live Escalator | {datetime.now().strftime('%Y-%m-%d %H:%M')}*",
158      ])
159  
160      filepath.write_text('\n'.join(lines), encoding='utf-8')
161      return filepath
162  
163  
164  def process_statement(
165      text: str,
166      alerts_dir: Path,
167      threshold: float = 0.4,
168      quiet: bool = False
169  ) -> Optional[Path]:
170      """Process a single statement."""
171      resonances = calculate_resonance(text)
172      max_axiom = max(resonances, key=resonances.get)
173      max_score = resonances[max_axiom]
174  
175      if not quiet:
176          print(f"[{max_score:>5.0%}] {max_axiom}: {text[:60]}...")
177  
178      if max_score >= threshold:
179          alert = create_alert(text, resonances, alerts_dir, threshold)
180          if alert and not quiet:
181              print(f"  → Alert created: {alert.name}")
182          return alert
183  
184      return None
185  
186  
187  def watch_transcript(
188      transcript_path: Path,
189      alerts_dir: Path,
190      threshold: float = 0.4,
191      interval: float = 5.0
192  ):
193      """Watch transcript and escalate new statements."""
194      print(f"Watching: {transcript_path}")
195      print(f"Alerts: {alerts_dir}")
196      print(f"Threshold: {threshold:.0%}")
197      print()
198  
199      last_line = 0
200      alerts_created = 0
201  
202      while True:
203          statements = extract_user_statements(transcript_path, last_line)
204  
205          for stmt in statements:
206              alert = process_statement(
207                  stmt['text'],
208                  alerts_dir,
209                  threshold,
210                  quiet=False
211              )
212              if alert:
213                  alerts_created += 1
214              last_line = max(last_line, stmt['line'] + 1)
215  
216          time.sleep(interval)
217  
218  
219  def main():
220      parser = argparse.ArgumentParser(
221          description='Live Escalator - Auto-escalate high-resonance insights'
222      )
223      parser.add_argument(
224          '--text', '-t',
225          type=str,
226          help='Single statement to process'
227      )
228      parser.add_argument(
229          '--watch', '-w',
230          type=Path,
231          help='Transcript file to watch'
232      )
233      parser.add_argument(
234          '--threshold',
235          type=float,
236          default=0.4,
237          help='Resonance threshold for escalation (default: 0.4)'
238      )
239      parser.add_argument(
240          '--alerts-dir',
241          type=Path,
242          default=Path(__file__).parent.parent / 'sessions' / 'RESONANCE-ALERTS',
243          help='Directory for alerts'
244      )
245      parser.add_argument(
246          '--interval',
247          type=float,
248          default=5.0,
249          help='Watch interval in seconds'
250      )
251  
252      args = parser.parse_args()
253  
254      args.alerts_dir.mkdir(parents=True, exist_ok=True)
255  
256      if args.text:
257          process_statement(args.text, args.alerts_dir, args.threshold)
258      elif args.watch:
259          watch_transcript(args.watch, args.alerts_dir, args.threshold, args.interval)
260      elif not sys.stdin.isatty():
261          text = sys.stdin.read().strip()
262          if text:
263              process_statement(text, args.alerts_dir, args.threshold)
264      else:
265          parser.print_help()
266          return 1
267  
268      return 0
269  
270  
271  if __name__ == '__main__':
272      sys.exit(main())