haive.core.schema.state
=======================

.. py:module:: haive.core.schema.state

.. autoapi-nested-parse::

   🌊 State System - Dynamic AI State Evolution Engine

   **THE LIVING MEMORY OF INTELLIGENT SYSTEMS**

   Welcome to the State System - the revolutionary foundation that transforms static data
   models into dynamic, evolving memory structures for AI agents. This isn't just another
   state management library; it's a comprehensive state evolution platform that enables
   AI systems to dynamically adapt their memory structures as they learn and grow.

   🧬 STATE EVOLUTION REVOLUTION
   -----------------------------

   The State System represents a paradigm shift in how AI agents manage and evolve their
   internal state. Every state becomes a **living, breathing data structure** that can:

   **🔄 Self-Modifying Architecture**: States that evolve their own structure based on discovered patterns
   **⚡ Real-time Adaptation**: Dynamic field addition and type evolution without restarts
   **🧠 Intelligent Merging**: Semantic-aware state combination with conflict resolution
   **📊 Temporal Memory**: Version history and time-travel debugging capabilities
   **🎯 Type-Safe Evolution**: Guaranteed type safety even during dynamic schema changes

   🌟 CORE INNOVATIONS
   -------------------

   **1. Dynamic State Architecture** 🏗️
      States that grow and evolve with your AI agents:

   .. rubric:: Examples

   >>> from haive.core.schema.state import StateSchema, Field
   >>>
   >>> class AdaptiveAgentState(StateSchema):
   >>> # Core fields that are always present
   >>> messages: List[BaseMessage] = Field(default_factory=list)
   >>> confidence: float = Field(default=0.0)
   >>>
   >>> # Fields that can be added dynamically
   >>> __dynamic_fields__ = True
   >>> __auto_evolve__ = True
   >>>
   >>> def discover_capability(self, capability: str, data: Any):
   >>> """Dynamically add new state fields based on discovered capabilities."""
   >>> if capability == "vision":
   >>> self.add_field("visual_memory", List[Image], default=[])
   >>> self.add_field("object_recognition", Dict[str, float], default={})
   >>> elif capability == "planning":
   >>> self.add_field("goal_stack", List[Goal], default=[])
   >>> self.add_field("plan_history", List[Plan], default=[])

   **2. Intelligent State Merging** 🧩
      Advanced merging strategies that understand your data:

       >>> # Define semantic merge strategies
       >>> class ConversationState(StateSchema):
       >>> messages: List[BaseMessage] = Field(default_factory=list)
       >>> knowledge: Dict[str, Any] = Field(default_factory=dict)
       >>> user_preferences: Dict[str, Any] = Field(default_factory=dict)
       >>>
       >>> # Custom reducer functions for intelligent merging
       >>> __reducer_fields__ = {
       >>> "messages": preserve_chronological_order,
       >>> "knowledge": semantic_knowledge_merge,
       >>> "user_preferences": preference_conflict_resolution
       >>> }
       >>>
       >>> # Automatic intelligent merging
       >>> state1 = ConversationState(messages=[msg1, msg2])
       >>> state2 = ConversationState(messages=[msg3, msg4])
       >>> merged = state1.merge(state2)  # Intelligent chronological merge

   **3. Temporal State Management** ⏰
      Time-aware state with history and rollback capabilities:

       >>> class TemporalState(StateSchema):
       >>> __enable_history__ = True
       >>> __snapshot_frequency__ = 10  # Snapshot every 10 updates
       >>> __max_history_size__ = 100
       >>>
       >>> current_data: Dict[str, Any] = Field(default_factory=dict)
       >>>
       >>> def rollback_to(self, timestamp: datetime):
       >>> """Rollback state to specific point in time."""
       >>> return self.restore_snapshot(timestamp)
       >>>
       >>> def get_state_at(self, timestamp: datetime):
       >>> """Get state at specific time without modifying current state."""
       >>> return self.view_snapshot(timestamp)
       >>>
       >>> def analyze_evolution(self) -> StateEvolutionAnalysis:
       >>> """Analyze how state has evolved over time."""
       >>> return StateEvolutionAnalysis(self.history)

   **4. Multi-Agent State Coordination** 🤝
      Sophisticated state sharing and synchronization:

       >>> class SharedState(StateSchema):
       >>> # Global shared fields
       >>> __shared_fields__ = ["global_knowledge", "conversation_history"]
       >>>
       >>> # Agent-specific private fields
       >>> __private_fields__ = ["internal_thoughts", "private_memory"]
       >>>
       >>> # Coordination fields
       >>> global_knowledge: KnowledgeGraph = Field(default_factory=KnowledgeGraph)
       >>> conversation_history: List[Message] = Field(default_factory=list)
       >>>
       >>> # Private to each agent instance
       >>> internal_thoughts: List[str] = Field(default_factory=list, private=True)
       >>> private_memory: Dict[str, Any] = Field(default_factory=dict, private=True)
       >>>
       >>> def share_with_agent(self, agent_id: str) -> AgentStateView:
       >>> """Create filtered view for specific agent."""
       >>> return self.create_view(
       >>> include=self.__shared_fields__,
       >>> exclude=self.__private_fields__,
       >>> agent_id=agent_id
       >>> )

   🎯 ADVANCED FEATURES
   --------------------

   **Real-time Schema Evolution** 🔮

       >>> # Schema that adapts to new patterns
       >>> class LearningState(StateSchema):
       >>> __learning_mode__ = True
       >>> __pattern_detection__ = True
       >>>
       >>> def update(self, data: Dict[str, Any]):
       >>> # Detect new patterns in incoming data
       >>> new_patterns = self.detect_patterns(data)
       >>>
       >>> # Automatically evolve schema
       >>> for pattern in new_patterns:
       >>> if pattern.confidence > 0.8:
       >>> self.add_computed_field(
       >>> pattern.field_name,
       >>> pattern.computation_logic,
       >>> pattern.type_hint
       >>> )
       >>>
       >>> # Apply update with evolved schema
       >>> super().update(data)

   **State Validation Chains** ✅

       >>> class ValidatedState(StateSchema):
       >>> data: Dict[str, Any] = Field(default_factory=dict)
       >>>
       >>> @field_validator("data")
       >>> @classmethod
       >>> def validate_data_structure(cls, v):
       >>> # Multi-level validation
       >>> structural_validation(v)
       >>> semantic_validation(v)
       >>> consistency_validation(v)
       >>> return v
       >>>
       >>> @model_validator(mode="after")
       >>> def cross_field_validation(self):
       >>> # Complex cross-field relationships
       >>> if not self.validate_field_relationships():
       >>> raise ValueError("Field relationships are inconsistent")
       >>> return self

   **State Performance Analytics** 📊

       >>> class AnalyticsState(StateSchema):
       >>> __enable_analytics__ = True
       >>> __track_performance__ = True
       >>>
       >>> def get_performance_metrics(self) -> StateMetrics:
       >>> return StateMetrics(
       >>> memory_usage=self.calculate_memory_usage(),
       >>> access_patterns=self.analyze_access_patterns(),
       >>> evolution_rate=self.calculate_evolution_rate(),
       >>> merge_efficiency=self.analyze_merge_performance()
       >>> )
       >>>
       >>> def optimize_structure(self) -> OptimizationReport:
       >>> """Automatically optimize state structure for performance."""
       >>> return self.apply_optimizations([
       >>> "field_reordering",
       >>> "type_optimization",
       >>> "compression",
       >>> "lazy_loading"
       >>> ])

   🏗️ STATE COMPOSITION PATTERNS
   ------------------------------

   **Hierarchical State Systems** 🏛️

       >>> class HierarchicalState(StateSchema):
       >>> # Root level data
       >>> session_info: SessionInfo = Field(...)
       >>>
       >>> # Nested state components
       >>> conversation_state: ConversationState = Field(default_factory=ConversationState)
       >>> knowledge_state: KnowledgeState = Field(default_factory=KnowledgeState)
       >>> planning_state: PlanningState = Field(default_factory=PlanningState)
       >>>
       >>> def get_component(self, component_name: str) -> StateSchema:
       >>> """Access specific state component."""
       >>> return getattr(self, f"{component_name}_state")
       >>>
       >>> def synchronize_components(self):
       >>> """Ensure all components are synchronized."""
       >>> shared_context = self.extract_shared_context()
       >>> for component in self.get_all_components():
       >>> component.update_from_shared_context(shared_context)

   **Plugin-based State Extensions** 🔌

       >>> class ExtensibleState(StateSchema):
       >>> __enable_plugins__ = True
       >>>
       >>> # Core state
       >>> base_data: Dict[str, Any] = Field(default_factory=dict)
       >>>
       >>> # Plugin registry
       >>> plugins: Dict[str, StatePlugin] = Field(default_factory=dict)
       >>>
       >>> def load_plugin(self, plugin: StatePlugin):
       >>> """Dynamically load state extension plugin."""
       >>> plugin_fields = plugin.get_fields()
       >>> plugin_methods = plugin.get_methods()
       >>> plugin_validators = plugin.get_validators()
       >>>
       >>> # Extend state with plugin capabilities
       >>> self.add_fields(plugin_fields)
       >>> self.add_methods(plugin_methods)
       >>> self.add_validators(plugin_validators)
       >>>
       >>> self.plugins[plugin.name] = plugin

   🔒 ENTERPRISE FEATURES
   ----------------------

   - **State Governance**: Approval workflows for schema changes
   - **Access Control**: Field-level permissions and role-based access
   - **Audit Logging**: Complete state evolution history
   - **Multi-tenancy**: Isolated state spaces per tenant
   - **Compliance**: GDPR, HIPAA compliance with automatic PII handling
   - **Backup & Recovery**: Automated state backup and disaster recovery

   🎓 BEST PRACTICES
   -----------------

   1. **Design for Evolution**: Plan for schema changes from day one
   2. **Use Typed Fields**: Leverage Pydantic's type system for safety
   3. **Implement Reducers**: Define intelligent merge strategies for complex data
   4. **Monitor Performance**: Track state size and access patterns
   5. **Version Your Schemas**: Use migration strategies for breaking changes
   6. **Test State Evolution**: Validate schema changes with comprehensive tests
   7. **Document State Contracts**: Clear documentation for shared state fields

   🚀 GETTING STARTED
   ------------------

       >>> from haive.core.schema.state import StateSchema, Field
       >>> from typing import List, Dict, Any
       >>>
       >>> # 1. Define your evolving state
       >>> class MyAgentState(StateSchema):
       >>> # Essential fields
       >>> messages: List[str] = Field(default_factory=list)
       >>> context: Dict[str, Any] = Field(default_factory=dict)
       >>>
       >>> # Enable dynamic evolution
       >>> __dynamic_fields__ = True
       >>>
       >>> # Define intelligent merging
       >>> __reducer_fields__ = {
       >>> "messages": lambda old, new: old + new,  # Append messages
       >>> "context": lambda old, new: {**old, **new}  # Merge contexts
       >>> }
       >>>
       >>> # 2. Use in your agent
       >>> state = MyAgentState()
       >>> state.messages.append("Hello!")
       >>>
       >>> # 3. Dynamic evolution
       >>> state.discover_capability("memory", {"type": "semantic", "size": 1000})
       >>>
       >>> # 4. Intelligent merging
       >>> other_state = MyAgentState(messages=["World!"])
       >>> merged = state.merge(other_state)  # ["Hello!", "World!"]

   ---

   **State System: Where Data Structures Become Living, Intelligent Memory** 🌊