""" PM-008: Fresh Session Spawner (Claude) Spawn fresh Claude API sessions for Genesis TRUE Method. Acceptance Criteria: - [x] GIVEN task WHEN spawn_claude_session() THEN new client - [x] AND session isolated per request - [x] AND supports both Sonnet and Opus - [x] AND stateless (no conversation history reuse) Dependencies: PM-005, PM-006 """ import os import json import logging import time from datetime import datetime from typing import Optional, Dict, Any, Tuple, List from dataclasses import dataclass, field try: import anthropic except ImportError: anthropic = None from core.model_tier_loader import load_model_tiers, TierConfig from core.learning_accumulator import get_learning_accumulator, FailureLesson logger = logging.getLogger(__name__) @dataclass class ClaudeSessionResult: """Result from a Claude session execution.""" success: bool output: Optional[str] = None error_type: Optional[str] = None error_message: Optional[str] = None # Metrics input_tokens: int = 0 output_tokens: int = 0 duration_ms: int = 0 cost: float = 0.0 # Learning new_lesson: Optional[FailureLesson] = None # Metadata model: str = "" attempt: int = 0 tier: int = 2 model_variant: str = "sonnet" # "sonnet" or "opus" timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat()) def to_dict(self) -> Dict[str, Any]: result = { "success": self.success, "output": self.output, "error_type": self.error_type, "error_message": self.error_message, "input_tokens": self.input_tokens, "output_tokens": self.output_tokens, "duration_ms": self.duration_ms, "cost": self.cost, "model": self.model, "attempt": self.attempt, "tier": self.tier, "model_variant": self.model_variant, "timestamp": self.timestamp } if self.new_lesson: result["new_lesson"] = self.new_lesson.to_dict() return result class ClaudeSessionSpawner: """ Spawn fresh Claude API sessions. Key Principles: - Each session is COMPLETELY FRESH (no history reuse) - Session isolated per request (stateless) - Supports both Sonnet (Tier 2) and Opus (Tier 3) - Learnings passed as discrete system/user context """ # Model mappings SONNET_MODELS = [ "claude-sonnet-4-20250514", "claude-3-5-sonnet-20241022", "claude-3-sonnet-20240229" ] OPUS_MODELS = [ "claude-opus-4-20250514", "claude-3-opus-20240229" ] def __init__(self, api_key: Optional[str] = None, default_sonnet: str = "claude-sonnet-4-20250514", default_opus: str = "claude-opus-4-20250514"): """ Initialize ClaudeSessionSpawner. Args: api_key: Anthropic API key. Defaults to env ANTHROPIC_API_KEY. default_sonnet: Default Sonnet model. default_opus: Default Opus model. """ self.api_key = api_key or os.getenv("ANTHROPIC_API_KEY") self.default_sonnet = default_sonnet self.default_opus = default_opus self.tier_config = load_model_tiers() self.learning_accumulator = get_learning_accumulator() if not anthropic: logger.warning("anthropic package not installed") elif not self.api_key: logger.warning("ANTHROPIC_API_KEY not set") else: logger.info("Claude API configured") def _get_model_variant(self, model: str) -> str: """Determine if model is Sonnet or Opus.""" model_lower = model.lower() if "opus" in model_lower: return "opus" return "sonnet" def _build_messages(self, task_description: str, task_id: str, include_learnings: bool = True, additional_context: Optional[str] = None) -> List[Dict[str, str]]: """ Build messages array with discrete learnings. Args: task_description: The task to complete task_id: Task identifier for learning lookup include_learnings: Whether to include previous learnings additional_context: Optional additional context Returns: Messages array for Claude API """ user_content_parts = [] # Add learnings from previous attempts (DISCRETE) if include_learnings: learnings = self.learning_accumulator.format_lessons_for_prompt(task_id) if learnings: user_content_parts.append(learnings) user_content_parts.append("---\n") # Add additional context if provided if additional_context: user_content_parts.append("## CONTEXT") user_content_parts.append(additional_context) user_content_parts.append("\n---\n") # Add the main task user_content_parts.append("## TASK") user_content_parts.append(task_description) user_content_parts.append("\nComplete this task. Provide a clear, working solution.") return [ {"role": "user", "content": "\n".join(user_content_parts)} ] def _get_system_prompt(self, tier: int) -> str: """Get system prompt based on tier.""" if tier == 3: return """You are Claude Opus, executing a high-stakes task for the Genesis system. This task has failed at lower tiers. Apply maximum capability and reasoning. Be thorough, precise, and ensure correctness. Think step by step.""" else: return """You are Claude Sonnet, executing a task for the Genesis system. Be efficient, accurate, and provide working solutions. Focus on correctness and clarity.""" def spawn_session(self, task_id: str, task_description: str, attempt: int, tier: int = 2, model: Optional[str] = None, timeout_seconds: Optional[int] = None, additional_context: Optional[str] = None, temperature: float = 0.7, max_tokens: int = 8192) -> ClaudeSessionResult: """ Spawn a fresh Claude session. Args: task_id: Task identifier task_description: Description of task to complete attempt: Attempt number (for tracking) tier: Execution tier (2 for Sonnet, 3 for Opus) model: Model to use (auto-selected based on tier if not provided) timeout_seconds: Request timeout additional_context: Additional context to include temperature: Generation temperature max_tokens: Maximum output tokens Returns: ClaudeSessionResult with outcome and new learnings """ start_time = time.time() # Determine model based on tier if not specified if not model: tier_config = self.tier_config.get_tier(tier) if tier_config and tier_config.provider == "anthropic": model = tier_config.model elif tier == 3: model = self.default_opus else: model = self.default_sonnet # Get timeout from tier config if not timeout_seconds: tier_config = self.tier_config.get_tier(tier) timeout_seconds = tier_config.timeout_seconds if tier_config else 180 model_variant = self._get_model_variant(model) logger.info(f"Spawning Claude session: task={task_id}, attempt={attempt}, " f"tier={tier}, model={model} ({model_variant})") # Check if Claude is available if not anthropic or not self.api_key: return ClaudeSessionResult( success=False, error_type="ConfigurationError", error_message="Claude API not configured", model=model, attempt=attempt, tier=tier, model_variant=model_variant, duration_ms=int((time.time() - start_time) * 1000) ) try: # Create FRESH client (no session state) client = anthropic.Anthropic( api_key=self.api_key, default_headers={"anthropic-beta": "context-management-2025-06-27"} ) # Build messages with discrete learnings messages = self._build_messages( task_description=task_description, task_id=task_id, include_learnings=True, additional_context=additional_context ) # Make the API call (stateless, isolated) response = client.messages.create( model=model, max_tokens=max_tokens, temperature=temperature, system=self._get_system_prompt(tier), messages=messages, timeout=timeout_seconds ) duration_ms = int((time.time() - start_time) * 1000) # Extract token counts input_tokens = response.usage.input_tokens if response.usage else 0 output_tokens = response.usage.output_tokens if response.usage else 0 # Calculate cost cost = self._calculate_cost(model_variant, input_tokens, output_tokens) # Get response text output_text = "" if response.content: for block in response.content: if hasattr(block, 'text'): output_text += block.text logger.info(f"Claude session completed: task={task_id}, tier={tier}, " f"duration={duration_ms}ms, tokens={input_tokens}+{output_tokens}") return ClaudeSessionResult( success=True, output=output_text, input_tokens=input_tokens, output_tokens=output_tokens, duration_ms=duration_ms, cost=cost, model=model, attempt=attempt, tier=tier, model_variant=model_variant ) except Exception as e: duration_ms = int((time.time() - start_time) * 1000) error_type = type(e).__name__ error_message = str(e)[:500] logger.warning(f"Claude session failed: {error_type}: {error_message}") # Capture failure as learning new_lesson = self.learning_accumulator.capture_failure( task_id=task_id, attempt_number=attempt, tier=tier, error_type=error_type, error_message=error_message, model=model, duration_ms=duration_ms ) return ClaudeSessionResult( success=False, error_type=error_type, error_message=error_message, duration_ms=duration_ms, model=model, attempt=attempt, tier=tier, model_variant=model_variant, new_lesson=new_lesson ) def _calculate_cost(self, model_variant: str, input_tokens: int, output_tokens: int) -> float: """Calculate cost based on model and token counts.""" if model_variant == "opus": # Opus pricing input_cost = (input_tokens / 1000) * 0.015 output_cost = (output_tokens / 1000) * 0.075 else: # Sonnet pricing input_cost = (input_tokens / 1000) * 0.003 output_cost = (output_tokens / 1000) * 0.015 return input_cost + output_cost def spawn_sonnet(self, task_id: str, task_description: str, attempt: int = 1, **kwargs) -> ClaudeSessionResult: """Convenience method for Sonnet (Tier 2) sessions.""" return self.spawn_session( task_id=task_id, task_description=task_description, attempt=attempt, tier=2, model=self.default_sonnet, **kwargs ) def spawn_opus(self, task_id: str, task_description: str, attempt: int = 1, **kwargs) -> ClaudeSessionResult: """Convenience method for Opus (Tier 3) sessions.""" return self.spawn_session( task_id=task_id, task_description=task_description, attempt=attempt, tier=3, model=self.default_opus, **kwargs ) def spawn_with_retries(self, task_id: str, task_description: str, max_attempts: int = 10, tier: int = 2, **kwargs) -> Tuple[ClaudeSessionResult, int]: """ Spawn sessions with retries until success or max attempts. Args: task_id: Task identifier task_description: Task description max_attempts: Maximum attempts (default 10) tier: Execution tier (2 or 3) **kwargs: Additional arguments for spawn_session Returns: Tuple of (final ClaudeSessionResult, attempts used) """ for attempt in range(1, max_attempts + 1): result = self.spawn_session( task_id=task_id, task_description=task_description, attempt=attempt, tier=tier, **kwargs ) if result.success: return result, attempt logger.info(f"Claude attempt {attempt}/{max_attempts} failed for task {task_id}") # Return final failed result return result, max_attempts def check_availability(self) -> Dict[str, Any]: """Check if Claude API is available and configured.""" status = { "configured": bool(anthropic and self.api_key), "api_key_set": bool(self.api_key), "package_installed": bool(anthropic), "default_sonnet": self.default_sonnet, "default_opus": self.default_opus } return status # Factory function def spawn_claude_session(task_id: str, task_description: str, attempt: int = 1, tier: int = 2, **kwargs) -> ClaudeSessionResult: """ Convenience function to spawn a single Claude session. Args: task_id: Task identifier task_description: Task description attempt: Attempt number tier: Tier (2=Sonnet, 3=Opus) **kwargs: Additional arguments Returns: ClaudeSessionResult """ spawner = ClaudeSessionSpawner() return spawner.spawn_session( task_id=task_id, task_description=task_description, attempt=attempt, tier=tier, **kwargs ) if __name__ == "__main__": # Test the ClaudeSessionSpawner logging.basicConfig(level=logging.INFO) spawner = ClaudeSessionSpawner() print("Claude Availability:") print(json.dumps(spawner.check_availability(), indent=2)) # Test with a simple task (only if API is configured) if spawner.api_key and anthropic: print("\nTesting Sonnet session...") result = spawner.spawn_sonnet( task_id="test-001", task_description="What is 2 + 2? Reply with just the number." ) print("Session Result:") print(json.dumps(result.to_dict(), indent=2))