1  #!/usr/bin/env python3
  2  """
  3  Hypercore Context Hook for Claude Code
  4  
  5  Injects cross-session awareness into Claude Code conversations.
  6  This runs as a Claude Code hook at the start of each user message.
  7  
  8  Install:
  9    Add to ~/.claude/settings.json:
 10    {
 11      "hooks": {
 12        "UserPromptSubmit": [
 13          {
 14            "command": "python3 /Users/rcerf/repos/Sovereign_OS/hooks/hypercore_context_hook.py"
 15          }
 16        ]
 17      }
 18    }
 19  
 20  What it does:
 21  1. Reads local FO-STATE.json (First Officer persistent state) - ALWAYS available
 22  2. Reads local Phoenix states from sessions directory - ALWAYS available
 23  3. Connects to Hypercore daemon at localhost:7777 - IF available
 24  4. Fetches cross-session attractors (hot topics) - IF Hypercore available
 25  5. Outputs awareness blocks for Claude
 26  
 27  The key insight: Local state is PRIMARY (always available), Hypercore is BONUS.
 28  This ensures cognitive continuity even when Hypercore daemon isn't running.
 29  """
 30  
 31  import json
 32  import sys
 33  import urllib.request
 34  import urllib.error
 35  from datetime import datetime
 36  from pathlib import Path
 37  
 38  DAEMON_URL = "http://localhost:7777"
 39  MAX_ATTRACTORS = 5
 40  MAX_PHOENIX_STATES = 3
 41  
 42  # Local paths (relative to hook location)
 43  HOOK_DIR = Path(__file__).parent
 44  SOVEREIGN_OS_ROOT = HOOK_DIR.parent
 45  SESSIONS_DIR = SOVEREIGN_OS_ROOT / "sessions"
 46  FO_STATE_PATH = SESSIONS_DIR / "FO-STATE.json"
 47  LIVE_COMPRESSION_PATH = SESSIONS_DIR / "LIVE-COMPRESSION.md"
 48  
 49  
 50  def fetch_json(path: str) -> dict:
 51      """Fetch JSON from daemon."""
 52      try:
 53          url = f"{DAEMON_URL}{path}"
 54          with urllib.request.urlopen(url, timeout=2) as response:
 55              return json.loads(response.read().decode('utf-8'))
 56      except:
 57          return {}
 58  
 59  
 60  # =============================================================================
 61  # LOCAL STATE READERS (ALWAYS AVAILABLE)
 62  # =============================================================================
 63  
 64  def get_local_fo_state() -> dict:
 65      """Read local First Officer state."""
 66      try:
 67          if FO_STATE_PATH.exists():
 68              return json.loads(FO_STATE_PATH.read_text())
 69      except:
 70          pass
 71      return {}
 72  
 73  
 74  def get_local_phoenix_state() -> str:
 75      """Read local LIVE-COMPRESSION.md and extract key info."""
 76      lines = []
 77  
 78      try:
 79          if LIVE_COMPRESSION_PATH.exists():
 80              content = LIVE_COMPRESSION_PATH.read_text()
 81  
 82              # Extract key metadata
 83              session_id = ""
 84              updated = ""
 85              focus = ""
 86              gravity_wells = []
 87  
 88              for line in content.split("\n"):
 89                  if "session_id::" in line or "session:" in line:
 90                      session_id = line.split("::")[-1].strip() if "::" in line else line.split(":")[-1].strip()
 91                  elif "updated::" in line:
 92                      updated = line.split("::", 1)[1].strip()
 93                  elif "**Primary work:**" in line:
 94                      focus = line.replace("**Primary work:**", "").strip()
 95                  elif "[[" in line and "]]" in line and "strength::" not in line:
 96                      import re
 97                      match = re.search(r'\[\[([^\]]+)\]\]', line)
 98                      if match:
 99                          gravity_wells.append(match.group(1))
100  
101              if session_id or updated or focus:
102                  lines.append(f"  - {session_id or 'current'} ({updated[:16] if updated else 'unknown'}): {', '.join(gravity_wells[:3]) or focus[:50]}")
103  
104      except:
105          pass
106  
107      return "\n".join(lines)
108  
109  
110  def get_local_gravity_wells() -> list:
111      """Extract gravity wells from local FO state."""
112      fo_state = get_local_fo_state()
113      wells = fo_state.get("gravity_wells", {})
114  
115      if wells:
116          sorted_wells = sorted(wells.items(), key=lambda x: -x[1])
117          return [{"concept": k, "strength": v} for k, v in sorted_wells[:10]]
118  
119      return []
120  
121  
122  def get_cross_session_context() -> str:
123      """Build cross-session awareness context."""
124      lines = []
125  
126      # Get attractors (hot topics across sessions)
127      attractors_data = fetch_json("/attractors")
128      attractors = attractors_data.get("attractors", [])
129  
130      # Filter to multi-session attractors
131      cross_session = [a for a in attractors if len(a.get("sessions", [])) >= 2]
132      multi_machine = [a for a in attractors if len(a.get("machines", [])) > 1]
133  
134      if cross_session or multi_machine:
135          lines.append("<cross-session-awareness>")
136          lines.append("Context from Hypercore P2P mesh:")
137          lines.append("")
138  
139          if cross_session:
140              lines.append("**Topics active across multiple sessions:**")
141              for a in cross_session[:MAX_ATTRACTORS]:
142                  topic = a.get("topic", "")
143                  sessions = len(a.get("sessions", []))
144                  lines.append(f"  - {topic} ({sessions} sessions)")
145              lines.append("")
146  
147          if multi_machine:
148              lines.append("**Topics spanning multiple machines:**")
149              for a in multi_machine[:MAX_ATTRACTORS]:
150                  topic = a.get("topic", "")
151                  machines = a.get("machines", [])
152                  lines.append(f"  - {topic} (on: {', '.join(machines)})")
153              lines.append("")
154  
155          lines.append("Consider how your conversation relates to these active threads.")
156          lines.append("</cross-session-awareness>")
157  
158      # Get recent Phoenix states
159      phoenix_data = fetch_json("/phoenix")
160      states = phoenix_data.get("states", [])
161  
162      if states:
163          # Sort by most recent
164          recent = sorted(states, key=lambda s: s.get("storedAt", ""), reverse=True)[:MAX_PHOENIX_STATES]
165  
166          if recent:
167              lines.append("")
168              lines.append("<recent-phoenix-states>")
169              lines.append("Recent cognitive checkpoints available for resurrection:")
170              for state in recent:
171                  session_id = state.get("sessionId", "unknown")
172                  stored_at = state.get("storedAt", "")[:16]
173                  wells = state.get("gravityWells", [])
174                  well_str = ", ".join(w.get("concept", "") if isinstance(w, dict) else str(w) for w in wells[:3])
175                  lines.append(f"  - {session_id} ({stored_at}): {well_str}")
176              lines.append("</recent-phoenix-states>")
177  
178      # Get daemon status
179      status = fetch_json("/status")
180      peers = status.get("peers", 0)
181      machine = status.get("machine", "unknown")
182  
183      if peers > 0:
184          lines.append("")
185          lines.append(f"<p2p-status>Connected to {peers} peer(s) via Hyperswarm</p2p-status>")
186  
187      return "\n".join(lines)
188  
189  
190  def register_session_start():
191      """Register this session start with Hypercore."""
192      try:
193          # Generate a session ID from timestamp
194          session_id = f"claude-code-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
195  
196          data = json.dumps({
197              "type": "session_start",
198              "sessionId": session_id,
199              "source": "claude-code-hook"
200          }).encode('utf-8')
201  
202          req = urllib.request.Request(
203              f"{DAEMON_URL}/event",
204              data=data,
205              headers={"Content-Type": "application/json"},
206              method="POST"
207          )
208  
209          with urllib.request.urlopen(req, timeout=2):
210              pass
211  
212      except:
213          pass  # Silent fail - don't break the hook
214  
215  
216  def main():
217      """Main hook entry point."""
218      # Read hook input from stdin (if any)
219      # For UserPromptSubmit, we just output context
220  
221      # Register this session
222      register_session_start()
223  
224      lines = []
225  
226      # ALWAYS: Get local phoenix state (doesn't require Hypercore daemon)
227      local_phoenix = get_local_phoenix_state()
228      local_wells = get_local_gravity_wells()
229  
230      if local_phoenix or local_wells:
231          lines.append("<recent-phoenix-states>")
232          lines.append("Recent cognitive checkpoints available for resurrection:")
233          if local_phoenix:
234              lines.append(local_phoenix)
235  
236      # BONUS: Get Hypercore cross-session context (if daemon running)
237      hypercore_context = get_cross_session_context()
238      if hypercore_context:
239          lines.append(hypercore_context)
240  
241      # Close phoenix states block if we started it
242      if local_phoenix or local_wells:
243          lines.append("</recent-phoenix-states>")
244  
245      # Output combined context
246      if lines:
247          print("\n".join(lines))
248  
249  
250  if __name__ == "__main__":
251      main()