#!/usr/bin/env python3 """ GENESIS CONTEXT WINDOW TRACKER ============================== Monitors Claude Code context window usage and triggers handoff when approaching the degradation threshold (~100k tokens). Research indicates Claude performance degrades noticeably after 100k tokens. This system: 1. Tracks approximate token usage 2. Warns at configurable thresholds 3. Auto-generates HANDOFF.md for session continuity 4. Can auto-trigger new session Usage: from core.context_window_tracker import ContextTracker tracker = ContextTracker() tracker.add_tokens(message_tokens) if tracker.should_handoff(): tracker.generate_handoff() Author: Genesis System Version: 1.0.0 """ import os import json import time from datetime import datetime from pathlib import Path from typing import Dict, List, Optional, Any, Tuple from dataclasses import dataclass, field, asdict from enum import Enum import threading class ContextState(Enum): """Context window health states.""" HEALTHY = "healthy" # < 50% of threshold WARMING = "warming" # 50-75% of threshold CAUTION = "caution" # 75-90% of threshold CRITICAL = "critical" # 90-100% of threshold DEGRADED = "degraded" # > 100% of threshold @dataclass class ContextMetrics: """Current context window metrics.""" total_tokens: int = 0 input_tokens: int = 0 output_tokens: int = 0 message_count: int = 0 tool_calls: int = 0 files_read: int = 0 session_start: str = field(default_factory=lambda: datetime.now().isoformat()) last_update: str = field(default_factory=lambda: datetime.now().isoformat()) def to_dict(self) -> Dict: return asdict(self) @dataclass class ContextConfig: """Configuration for context tracking.""" # Token thresholds degradation_threshold: int = 100_000 # Performance degrades here warning_threshold: int = 75_000 # First warning caution_threshold: int = 90_000 # Urgent warning critical_threshold: int = 95_000 # Prepare handoff # Behavior settings auto_handoff: bool = False # Auto-generate handoff at threshold auto_suggest_new_session: bool = True # Suggest new session at threshold track_tool_usage: bool = True # Track tool call tokens estimate_system_prompt: int = 15_000 # Estimated system prompt tokens # Alerts alert_at_warning: bool = True alert_at_caution: bool = True alert_at_critical: bool = True class ContextTracker: """ Tracks Claude Code context window usage and manages session handoffs. Monitors token accumulation and provides warnings/automation when approaching the ~100k token degradation threshold. """ CONFIG_PATH = Path("E:/genesis-system/config/context_tracker_config.json") STATE_PATH = Path("E:/genesis-system/data/context_tracker_state.json") HANDOFF_PATH = Path("E:/genesis-system/HANDOFF.md") LOG_PATH = Path("E:/genesis-system/data/context_tracker.log") # Singleton _instance = None _lock = threading.Lock() def __new__(cls): with cls._lock: if cls._instance is None: cls._instance = super().__new__(cls) cls._instance._initialized = False return cls._instance def __init__(self): if self._initialized: return self.config = self._load_config() self.metrics = self._load_state() self._alert_shown = { "warning": False, "caution": False, "critical": False } # Ensure directories exist self.STATE_PATH.parent.mkdir(parents=True, exist_ok=True) self.LOG_PATH.parent.mkdir(parents=True, exist_ok=True) self._initialized = True def _load_config(self) -> ContextConfig: """Load configuration from file or use defaults.""" if self.CONFIG_PATH.exists(): try: with open(self.CONFIG_PATH) as f: data = json.load(f) return ContextConfig(**data) except Exception: pass return ContextConfig() def _load_state(self) -> ContextMetrics: """Load saved state or create new.""" if self.STATE_PATH.exists(): try: with open(self.STATE_PATH) as f: data = json.load(f) # Check if session is recent (within 2 hours) last_update = datetime.fromisoformat(data.get("last_update", "2000-01-01")) if (datetime.now() - last_update).total_seconds() < 7200: return ContextMetrics(**data) except Exception: pass return ContextMetrics() def _save_state(self): """Persist current state.""" self.metrics.last_update = datetime.now().isoformat() with open(self.STATE_PATH, "w") as f: json.dump(self.metrics.to_dict(), f, indent=2) def _log(self, message: str, level: str = "INFO"): """Log tracking events.""" timestamp = datetime.now().isoformat() log_line = f"[{timestamp}] [{level}] {message}\n" with open(self.LOG_PATH, "a") as f: f.write(log_line) # === Token Tracking === def add_tokens( self, input_tokens: int = 0, output_tokens: int = 0, is_tool_call: bool = False, file_read: bool = False ) -> ContextState: """ Add tokens to the running total. Args: input_tokens: Tokens from user/system input output_tokens: Tokens from Claude's response is_tool_call: Whether this was a tool call file_read: Whether this involved reading a file Returns: Current context state """ with self._lock: self.metrics.input_tokens += input_tokens self.metrics.output_tokens += output_tokens self.metrics.total_tokens += input_tokens + output_tokens self.metrics.message_count += 1 if is_tool_call: self.metrics.tool_calls += 1 if file_read: self.metrics.files_read += 1 self._save_state() state = self.get_state() self._check_alerts(state) return state def estimate_tokens(self, text: str) -> int: """ Estimate token count from text. Rough approximation: ~4 characters per token for English. """ return len(text) // 4 def add_message(self, message: str, is_user: bool = True) -> ContextState: """Add a message and estimate its tokens.""" tokens = self.estimate_tokens(message) if is_user: return self.add_tokens(input_tokens=tokens) else: return self.add_tokens(output_tokens=tokens) # === State Checking === def get_state(self) -> ContextState: """Get current context window state.""" effective_tokens = self.get_effective_tokens() threshold = self.config.degradation_threshold if effective_tokens >= threshold: return ContextState.DEGRADED elif effective_tokens >= self.config.critical_threshold: return ContextState.CRITICAL elif effective_tokens >= self.config.caution_threshold: return ContextState.CAUTION elif effective_tokens >= self.config.warning_threshold: return ContextState.WARMING else: return ContextState.HEALTHY def get_effective_tokens(self) -> int: """Get effective token count including system prompt estimate.""" return self.metrics.total_tokens + self.config.estimate_system_prompt def get_utilization(self) -> float: """Get context window utilization as percentage.""" return self.get_effective_tokens() / self.config.degradation_threshold def get_remaining(self) -> int: """Get estimated remaining tokens before degradation.""" return max(0, self.config.degradation_threshold - self.get_effective_tokens()) def should_handoff(self) -> bool: """Check if session should hand off.""" state = self.get_state() return state in [ContextState.CRITICAL, ContextState.DEGRADED] # === Alerts === def _check_alerts(self, state: ContextState): """Check and trigger alerts based on state.""" if state == ContextState.WARMING and not self._alert_shown["warning"]: if self.config.alert_at_warning: self._show_alert("warning") self._alert_shown["warning"] = True elif state == ContextState.CAUTION and not self._alert_shown["caution"]: if self.config.alert_at_caution: self._show_alert("caution") self._alert_shown["caution"] = True elif state == ContextState.CRITICAL and not self._alert_shown["critical"]: if self.config.alert_at_critical: self._show_alert("critical") self._alert_shown["critical"] = True # Auto-handoff if configured if self.config.auto_handoff: self.generate_handoff() def _show_alert(self, level: str): """Display alert to user.""" effective = self.get_effective_tokens() remaining = self.get_remaining() utilization = self.get_utilization() * 100 alerts = { "warning": f""" ╔══════════════════════════════════════════════════════════════╗ ║ ⚠️ CONTEXT WINDOW WARNING - {utilization:.0f}% USED ║ ╠══════════════════════════════════════════════════════════════╣ ║ Tokens used: ~{effective:,} / 100,000 ║ ║ Remaining: ~{remaining:,} tokens ║ ║ ║ ║ Consider using /clear for unrelated new tasks ║ ╚══════════════════════════════════════════════════════════════╝ """, "caution": f""" ╔══════════════════════════════════════════════════════════════╗ ║ 🟠 CONTEXT WINDOW CAUTION - {utilization:.0f}% USED ║ ╠══════════════════════════════════════════════════════════════╣ ║ Tokens used: ~{effective:,} / 100,000 ║ ║ Remaining: ~{remaining:,} tokens ║ ║ ║ ║ Performance may start degrading soon. ║ ║ Consider: /handoff to save progress ║ ╚══════════════════════════════════════════════════════════════╝ """, "critical": f""" ╔══════════════════════════════════════════════════════════════╗ ║ 🔴 CONTEXT WINDOW CRITICAL - {utilization:.0f}% USED ║ ╠══════════════════════════════════════════════════════════════╣ ║ Tokens used: ~{effective:,} / 100,000 ║ ║ Remaining: ~{remaining:,} tokens ║ ║ ║ ║ ⚠️ DEGRADATION THRESHOLD APPROACHING ║ ║ ║ ║ RECOMMENDED: Save progress and start fresh session ║ ║ ║ ║ Options: ║ ║ 1. Run /handoff to save current progress ║ ║ 2. Start new Claude Code session ║ ║ 3. Continue (may experience degraded quality) ║ ║ ║ ║ Should we save progress and start fresh? (y/n) ║ ╚══════════════════════════════════════════════════════════════╝ """ } print(alerts.get(level, "")) self._log(f"Alert shown: {level} at {effective} tokens", "ALERT") # === Handoff Generation === def generate_handoff( self, summary: str = "", active_tasks: List[str] = None, next_steps: List[str] = None, important_files: List[str] = None ) -> str: """ Generate HANDOFF.md for session continuity. Args: summary: Brief summary of what was done active_tasks: List of in-progress tasks next_steps: Recommended next actions important_files: Files relevant to continue Returns: Path to generated handoff file """ active_tasks = active_tasks or [] next_steps = next_steps or [] important_files = important_files or [] metrics = self.metrics effective = self.get_effective_tokens() handoff_content = f"""# Session Handoff - {datetime.now().strftime('%Y-%m-%d %H:%M')} ## Session Metrics - **Tokens Used**: ~{effective:,} / 100,000 - **Messages**: {metrics.message_count} - **Tool Calls**: {metrics.tool_calls} - **Files Read**: {metrics.files_read} - **Session Start**: {metrics.session_start} - **Reason**: Context window approaching degradation threshold --- ## Summary {summary if summary else "[Add summary of what was accomplished]"} --- ## Active Tasks {self._format_list(active_tasks) if active_tasks else "- [ ] [Add any in-progress tasks]"} --- ## Next Steps {self._format_numbered_list(next_steps) if next_steps else "1. [Add recommended next steps]"} --- ## Important Files {self._format_list(important_files) if important_files else "- [Add files relevant to continue]"} --- ## Context for Next Session ### What Was Working - [Add what was functioning correctly] ### Known Issues - [Add any blockers or problems encountered] ### Key Decisions Made - [Add important decisions and rationale] --- ## Quick Start for Next Session ```bash # Start new Claude Code session cd E:/genesis-system claude # First message to Claude: # "Read HANDOFF.md and continue from where we left off" ``` --- *Generated by Genesis Context Window Tracker* *Threshold: 100k tokens | Used: ~{effective:,} tokens* """ with open(self.HANDOFF_PATH, "w") as f: f.write(handoff_content) self._log(f"Handoff generated at {self.HANDOFF_PATH}", "INFO") print(f"\n✅ Handoff saved to: {self.HANDOFF_PATH}") print(" Start a new session and ask Claude to read HANDOFF.md\n") return str(self.HANDOFF_PATH) def _format_list(self, items: List[str]) -> str: """Format items as markdown bullet list.""" return "\n".join(f"- {item}" for item in items) def _format_numbered_list(self, items: List[str]) -> str: """Format items as numbered list.""" return "\n".join(f"{i+1}. {item}" for i, item in enumerate(items)) # === Session Management === def reset(self): """Reset tracker for new session.""" with self._lock: self.metrics = ContextMetrics() self._alert_shown = {"warning": False, "caution": False, "critical": False} self._save_state() self._log("Session reset", "INFO") def get_status(self) -> Dict[str, Any]: """Get current status as dictionary.""" state = self.get_state() return { "state": state.value, "total_tokens": self.metrics.total_tokens, "effective_tokens": self.get_effective_tokens(), "threshold": self.config.degradation_threshold, "utilization": f"{self.get_utilization() * 100:.1f}%", "remaining": self.get_remaining(), "message_count": self.metrics.message_count, "tool_calls": self.metrics.tool_calls, "files_read": self.metrics.files_read, "session_start": self.metrics.session_start, "should_handoff": self.should_handoff() } def print_status(self): """Print formatted status.""" status = self.get_status() state = self.get_state() state_colors = { ContextState.HEALTHY: "🟢", ContextState.WARMING: "🟡", ContextState.CAUTION: "🟠", ContextState.CRITICAL: "🔴", ContextState.DEGRADED: "⛔" } print(f""" ╔══════════════════════════════════════════════════════════════╗ ║ CONTEXT WINDOW STATUS ║ ╠══════════════════════════════════════════════════════════════╣ ║ State: {state_colors.get(state, "⚪")} {state.value.upper():<52} ║ ║ Utilization: {status['utilization']:<47} ║ ║ Tokens: {status['effective_tokens']:,} / {status['threshold']:,}{' ' * 30}║ ║ Remaining: ~{status['remaining']:,} tokens{' ' * 33}║ ╠══════════════════════════════════════════════════════════════╣ ║ Messages: {status['message_count']:<50} ║ ║ Tool Calls: {status['tool_calls']:<48} ║ ║ Files Read: {status['files_read']:<48} ║ ╚══════════════════════════════════════════════════════════════╝ """) # === Convenience Functions === _tracker = None def get_tracker() -> ContextTracker: """Get singleton tracker instance.""" global _tracker if _tracker is None: _tracker = ContextTracker() return _tracker def track(input_tokens: int = 0, output_tokens: int = 0) -> ContextState: """Quick function to track tokens.""" return get_tracker().add_tokens(input_tokens, output_tokens) def status(): """Print current status.""" get_tracker().print_status() def should_handoff() -> bool: """Check if handoff is recommended.""" return get_tracker().should_handoff() def handoff(summary: str = ""): """Generate handoff document.""" return get_tracker().generate_handoff(summary=summary) def reset(): """Reset for new session.""" get_tracker().reset() # === CLI === def main(): import argparse parser = argparse.ArgumentParser(description="Context Window Tracker") parser.add_argument("command", choices=["status", "reset", "handoff", "simulate"]) parser.add_argument("--tokens", type=int, default=0, help="Tokens to add (simulate)") parser.add_argument("--summary", type=str, default="", help="Handoff summary") args = parser.parse_args() tracker = get_tracker() if args.command == "status": tracker.print_status() elif args.command == "reset": tracker.reset() print("✅ Session reset") elif args.command == "handoff": tracker.generate_handoff(summary=args.summary) elif args.command == "simulate": if args.tokens > 0: tracker.add_tokens(input_tokens=args.tokens // 2, output_tokens=args.tokens // 2) tracker.print_status() else: # Simulate gradual increase for i in range(20): tracker.add_tokens(input_tokens=5000, output_tokens=5000) print(f"Added 10k tokens. Total: {tracker.get_effective_tokens():,}") time.sleep(0.5) if __name__ == "__main__": main()