1  """
  2  Context Bootstrap - Fast context loading for new sessions.
  3  
  4  Problem: When you open a new Claude session, it starts cold.
  5  You have to re-explain context, re-establish what you're working on.
  6  
  7  Solution: The attention system has been tracking all your sessions.
  8  When a new session starts, bootstrap it with distilled context from
  9  recent activity across all sessions.
 10  
 11  This enables:
 12  1. Cold start → warm start in seconds
 13  2. Branch sessions that share foundation
 14  3. Resume work without re-explaining
 15  4. Context continuity across session restarts
 16  
 17  Usage:
 18      bootstrap = ContextBootstrap(sessions_dir, daily_dir)
 19  
 20      # Get briefing for a new session
 21      briefing = bootstrap.get_briefing(
 22          hours_back=4,
 23          max_tokens=2000
 24      )
 25  
 26      # Inject into session as system context
 27      print(briefing.to_markdown())
 28  """
 29  
 30  from dataclasses import dataclass, field
 31  from datetime import datetime, timedelta
 32  from typing import Optional, List, Dict, Tuple
 33  from pathlib import Path
 34  import re
 35  
 36  from .cross_session import create_cross_session_system, UnifiedAttentionState
 37  from .coherence_risk import create_coherence_system, DecisionType
 38  
 39  
 40  @dataclass
 41  class ContextBriefing:
 42      """
 43      A briefing to bootstrap a new session with context.
 44  
 45      Contains distilled knowledge from recent sessions:
 46      - What topics were active
 47      - What decisions were made
 48      - What questions are unresolved
 49      - What direction attention was heading
 50      """
 51      generated_at: datetime = field(default_factory=datetime.now)
 52  
 53      # Time window covered
 54      hours_covered: float = 4.0
 55      sessions_analyzed: int = 0
 56  
 57      # Active topics (what you were thinking about)
 58      hot_topics: List[str] = field(default_factory=list)
 59      cross_session_topics: List[str] = field(default_factory=list)
 60  
 61      # Decisions made (so new session knows the choices)
 62      recent_decisions: List[Dict] = field(default_factory=list)
 63  
 64      # Unresolved items (what still needs attention)
 65      unresolved: List[str] = field(default_factory=list)
 66  
 67      # Trajectory (where attention was heading)
 68      trajectory_summary: str = ""
 69  
 70      # Key insights/ahas detected
 71      recent_ahas: List[Dict] = field(default_factory=list)
 72  
 73      # Session summaries (brief per-session context)
 74      session_summaries: List[Dict] = field(default_factory=list)
 75  
 76      def to_markdown(self) -> str:
 77          """Convert briefing to markdown for injection."""
 78          lines = [
 79              "## Context Briefing",
 80              f"*Generated from {self.sessions_analyzed} sessions over {self.hours_covered:.1f} hours*",
 81              ""
 82          ]
 83  
 84          # Hot topics
 85          if self.hot_topics:
 86              lines.append("### Active Topics")
 87              lines.append(f"Currently working on: **{', '.join(self.hot_topics[:5])}**")
 88              if self.cross_session_topics:
 89                  lines.append(f"Cross-session focus: {', '.join(self.cross_session_topics[:3])}")
 90              lines.append("")
 91  
 92          # Recent decisions
 93          if self.recent_decisions:
 94              lines.append("### Recent Decisions")
 95              for d in self.recent_decisions[:5]:
 96                  dtype = d.get('type', 'decision')
 97                  topic = d.get('topic', '')
 98                  content = d.get('content', '')[:60]
 99                  lines.append(f"- [{dtype}] **{topic}**: {content}...")
100              lines.append("")
101  
102          # Unresolved
103          if self.unresolved:
104              lines.append("### Unresolved Items")
105              for item in self.unresolved[:5]:
106                  lines.append(f"- [ ] {item}")
107              lines.append("")
108  
109          # Trajectory
110          if self.trajectory_summary:
111              lines.append("### Direction")
112              lines.append(self.trajectory_summary)
113              lines.append("")
114  
115          # Ahas
116          if self.recent_ahas:
117              lines.append("### Recent Insights")
118              for aha in self.recent_ahas[:3]:
119                  emoji = "💡" if aha.get('type') == 'discovery' else "🔧"
120                  lines.append(f"- {emoji} {aha.get('summary', '')[:80]}")
121              lines.append("")
122  
123          return '\n'.join(lines)
124  
125      def to_system_message(self) -> str:
126          """Convert to system message format for Claude."""
127          md = self.to_markdown()
128          return f"""<context-bootstrap>
129  {md}
130  </context-bootstrap>
131  
132  You are continuing work from previous sessions. The above briefing summarizes recent context.
133  Use this to orient yourself, but the user may want to go in a different direction."""
134  
135  
136  class ContextBootstrap:
137      """
138      Generates context briefings from recent session history.
139  
140      Analyzes session streams, extracts key information, and
141      produces a briefing that can bootstrap a new session.
142      """
143  
144      def __init__(
145          self,
146          sessions_dir: str,
147          daily_notes_dir: str
148      ):
149          self.sessions_dir = Path(sessions_dir)
150          self.daily_notes_dir = Path(daily_notes_dir)
151  
152          # Initialize trackers
153          self.cross_tracker, self.session_watcher = create_cross_session_system(
154              str(sessions_dir)
155          )
156          self.coherence_detector, _ = create_coherence_system()
157  
158          # Process current state
159          self.session_watcher.scan_sessions()
160          self.session_watcher.process_updates()
161  
162      def get_briefing(
163          self,
164          hours_back: float = 4.0,
165          max_tokens: int = 2000,
166          focus_session: str = None
167      ) -> ContextBriefing:
168          """
169          Generate a context briefing from recent sessions.
170  
171          Args:
172              hours_back: How far back to look
173              max_tokens: Approximate token budget for briefing
174              focus_session: Optional session to focus on
175  
176          Returns:
177              ContextBriefing ready for injection
178          """
179          cutoff = datetime.now() - timedelta(hours=hours_back)
180  
181          # Get cross-session state
182          state = self.cross_tracker.get_state()
183  
184          # Filter to recent sessions
185          recent_sessions = {
186              sid: info for sid, info in state.sessions.items()
187              if info.last_activity and info.last_activity > cutoff
188          }
189  
190          briefing = ContextBriefing(
191              hours_covered=hours_back,
192              sessions_analyzed=len(recent_sessions)
193          )
194  
195          # Extract hot topics from sessions
196          all_topics = []
197          for info in recent_sessions.values():
198              all_topics.extend(info.topics)
199  
200          # Count topic frequency
201          topic_counts = {}
202          for topic in all_topics:
203              if topic and len(topic) > 2:
204                  topic_counts[topic] = topic_counts.get(topic, 0) + 1
205  
206          # Sort by frequency
207          sorted_topics = sorted(topic_counts.items(), key=lambda x: x[1], reverse=True)
208          briefing.hot_topics = [t for t, c in sorted_topics[:10]]
209  
210          # Cross-session attractors
211          briefing.cross_session_topics = [
212              a for a in state.cross_session_attractors
213              if a and len(a) > 2 and '\n' not in a
214          ][:5]
215  
216          # Extract decisions from session content
217          briefing.recent_decisions = self._extract_recent_decisions(
218              recent_sessions, cutoff
219          )
220  
221          # Build trajectory summary
222          if briefing.hot_topics:
223              briefing.trajectory_summary = self._build_trajectory_summary(
224                  briefing.hot_topics,
225                  briefing.cross_session_topics
226              )
227  
228          # Session summaries
229          briefing.session_summaries = self._build_session_summaries(
230              recent_sessions, max_per_session=100
231          )
232  
233          return briefing
234  
235      def _extract_recent_decisions(
236          self,
237          sessions: Dict,
238          cutoff: datetime
239      ) -> List[Dict]:
240          """Extract decisions from recent sessions."""
241          decisions = []
242  
243          # Decision patterns
244          patterns = [
245              (r"(?:decided|choosing|going with)\s+(.{10,60})", "decision"),
246              (r"(?:principle|rule):\s*(.{10,60})", "principle"),
247              (r"(?:architecture|design):\s*(.{10,60})", "architecture"),
248          ]
249  
250          for sid, info in sessions.items():
251              # Read session file
252              session_files = list(self.sessions_dir.glob(f"*{sid}*-live.md"))
253              if not session_files:
254                  continue
255  
256              try:
257                  content = session_files[0].read_text()
258              except:
259                  continue
260  
261              # Only look at recent content (last 20% of file or last 10KB)
262              if len(content) > 10000:
263                  content = content[-10000:]
264  
265              for pattern, dtype in patterns:
266                  matches = re.finditer(pattern, content, re.IGNORECASE)
267                  for match in matches:
268                      text = match.group(1).strip()
269                      if len(text) > 10:
270                          decisions.append({
271                              'type': dtype,
272                              'topic': self._extract_topic(content, match.start()),
273                              'content': text,
274                              'session': sid[:8]
275                          })
276  
277          # Deduplicate
278          seen = set()
279          unique = []
280          for d in decisions:
281              key = d['content'][:30].lower()
282              if key not in seen:
283                  seen.add(key)
284                  unique.append(d)
285  
286          return unique[:10]
287  
288      def _extract_topic(self, content: str, position: int) -> str:
289          """Extract likely topic from context around a position."""
290          start = max(0, position - 100)
291          end = min(len(content), position + 100)
292          context = content[start:end]
293  
294          # Look for code terms
295          terms = re.findall(r'\b([a-z]+_[a-z_]+)\b', context)
296          if terms:
297              return terms[0]
298  
299          # Look for hashtags
300          tags = re.findall(r'#(\w+)', context)
301          if tags:
302              return tags[0]
303  
304          return "general"
305  
306      def _build_trajectory_summary(
307          self,
308          hot_topics: List[str],
309          cross_topics: List[str]
310      ) -> str:
311          """Build a summary of where attention is heading."""
312          if cross_topics:
313              return f"Converging on: **{cross_topics[0]}**. Related: {', '.join(hot_topics[:3])}"
314          elif hot_topics:
315              return f"Exploring: {', '.join(hot_topics[:3])}"
316          return "No clear direction detected"
317  
318      def _build_session_summaries(
319          self,
320          sessions: Dict,
321          max_per_session: int = 100
322      ) -> List[Dict]:
323          """Build brief summaries of each session."""
324          summaries = []
325  
326          for sid, info in sessions.items():
327              topics = list(info.topics)[:5]
328              summaries.append({
329                  'session': sid[:8],
330                  'topics': topics,
331                  'atoms': info.atom_count,
332                  'last_active': info.last_activity.strftime('%H:%M') if info.last_activity else '?'
333              })
334  
335          return summaries
336  
337  
338  def create_bootstrap_system(
339      sessions_dir: str,
340      daily_notes_dir: str
341  ) -> ContextBootstrap:
342      """Create a context bootstrap system."""
343      return ContextBootstrap(sessions_dir, daily_notes_dir)
344  
345  
346  # Quick command to get a briefing
347  def get_quick_briefing() -> str:
348      """Get a quick briefing for the current moment."""
349      bootstrap = ContextBootstrap(
350          sessions_dir='/Users/rcerf/repos/Sovereign_Estate/daily/sessions/',
351          daily_notes_dir='/Users/rcerf/repos/Sovereign_Estate/daily/'
352      )
353      briefing = bootstrap.get_briefing(hours_back=4)
354      return briefing.to_markdown()
355  
356  
357  if __name__ == "__main__":
358      print("=== Context Bootstrap ===\n")
359  
360      bootstrap = ContextBootstrap(
361          sessions_dir='/Users/rcerf/repos/Sovereign_Estate/daily/sessions/',
362          daily_notes_dir='/Users/rcerf/repos/Sovereign_Estate/daily/'
363      )
364  
365      briefing = bootstrap.get_briefing(hours_back=8)
366  
367      print(f"Sessions analyzed: {briefing.sessions_analyzed}")
368      print(f"Hours covered: {briefing.hours_covered}")
369      print()
370      print(briefing.to_markdown())
371      print()
372      print("--- System Message Format ---")
373      print(briefing.to_system_message()[:500] + "...")