haive.core.schema.prebuilt.dynamic_activation_state

Dynamic Activation State for Component Management.

This module provides DynamicActivationState, a specialized state schema for dynamic component activation patterns. It extends StateSchema with registry management, MetaStateSchema integration, and activation tracking.

Based on the Dynamic Activation Pattern: @project_docs/active/patterns/dynamic_activation_pattern.md

Classes

DynamicActivationState

State schema for dynamic component activation patterns.

Module Contents

class haive.core.schema.prebuilt.dynamic_activation_state.DynamicActivationState(/, **data)

Bases: haive.core.schema.state_schema.StateSchema

State schema for dynamic component activation patterns.

This state schema provides the foundation for dynamic component activation using MetaStateSchema for component wrapping and DynamicRegistry for component management. It supports discovery, activation, and tracking of components with full type safety.

Key Features:

• Generic component registry with activation tracking

• MetaStateSchema integration for component wrapping

• Discovery configuration for component finding

• Activation history with detailed tracking

• Component lifecycle management

• Type-safe component access

Parameters:

• registry – Registry of available components

• active_meta_states – MetaStateSchema instances for active components

• discovery_config – Configuration for component discovery

• activation_history – History of activation/deactivation events

• current_task – Current task being processed

• required_capabilities – Capabilities needed for current task

• missing_capabilities – Capabilities currently missing

• data (Any)

Examples

Basic usage:

from haive.core.schema.prebuilt.dynamic_activation_state import DynamicActivationState
from haive.core.registry import DynamicRegistry, RegistryItem
from langchain_core.tools import tool

@tool
def calculator(expression: str) -> float:
    '''Calculate mathematical expression.'''
    return eval(expression)

# Create state with registry
state = DynamicActivationState()

# Register a component
item = RegistryItem(
    id="calc",
    name="calculator",
    description="Math operations",
    component=calculator
)
state.registry.register(item)

# Activate component
meta_state = state.activate_component("calc")
assert meta_state is not None

With discovery configuration:

state = DynamicActivationState(
    discovery_config={
        "source": "@haive-tools/docs",
        "auto_discover": True,
        "query": "math tools"
    }
)

# Track required capabilities
state.required_capabilities = ["math", "calculation"]
state.missing_capabilities = ["math"]

# Activate component to fill capability gap
meta_state = state.activate_component("calc")
state.missing_capabilities.remove("math")

With activation history:

# Check activation history
recent_activations = [
    event for event in state.activation_history
    if event["action"] == "activate"
]

print(f"Recent activations: {len(recent_activations)}")
for event in recent_activations[-3:]:
    print(f"- {event['component_name']} at {event['timestamp']}")

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

activate_component(component_id)

Activate a component and wrap in MetaStateSchema.

Parameters:

component_id (str) – ID of component to activate

Returns:

MetaStateSchema instance if activation succeeded, None otherwise

Return type:

haive.core.schema.prebuilt.meta_state.MetaStateSchema | None

Examples

Activate a registered component:

# Register component first
item = RegistryItem(
    id="my_tool",
    name="My Tool",
    description="Useful tool",
    component=tool_instance
)
state.registry.register(item)

# Activate component
meta_state = state.activate_component("my_tool")
if meta_state:
    print(f"Component {component_id} activated successfully")

Check activation result:

meta_state = state.activate_component("calculator")
if meta_state:
    # Use the wrapped component
    result = await meta_state.execute_agent(
        input_data={"expression": "2 + 2"}
    )
    print(f"Result: {result}")

deactivate_component(component_id)

Deactivate a component and remove from active states.

Parameters:

component_id (str) – ID of component to deactivate

Returns:

True if deactivation succeeded, False otherwise

Return type:

bool

Examples

Deactivate a component:

success = state.deactivate_component("my_tool")
if success:
    print("Component deactivated successfully")

Deactivate all components:

active_ids = list(state.active_meta_states.keys())
for comp_id in active_ids:
    state.deactivate_component(comp_id)

assert len(state.active_meta_states) == 0

get_activation_stats()

Get statistics about component activation.

Returns:

Dictionary with activation statistics

Return type:

dict[str, Any]

Examples

Show activation statistics:

stats = state.get_activation_stats()
print(f"Total components: {stats['total_components']}")
print(f"Active components: {stats['active_components']}")
print(f"Activation rate: {stats['activation_rate']:.1%}")
print(f"Most activated: {stats['most_activated_component']}")

get_active_components()

Get all active component instances.

Returns:

List of active component instances

Return type:

list[Any]

Examples

Get active components:

active_components = state.get_active_components()
print(f"Active components: {len(active_components)}")

for component in active_components:
    print(f"- {type(component).__name__}")

Use active components:

tools = state.get_active_components()
for tool in tools:
    if hasattr(tool, 'invoke'):
        result = tool.invoke({"input": "test"})

get_meta_state(component_id)

Get MetaStateSchema for a specific component.

Parameters:

component_id (str) – ID of component to get meta state for

Returns:

MetaStateSchema instance if found, None otherwise

Return type:

haive.core.schema.prebuilt.meta_state.MetaStateSchema | None

Examples

Get meta state for execution:

meta_state = state.get_meta_state("calculator")
if meta_state:
    result = await meta_state.execute_agent(
        input_data={"expression": "10 * 5"}
    )
    print(f"Calculation result: {result}")

get_unsatisfied_capabilities()

Get list of capabilities that are still unsatisfied.

Returns:

List of capability names that are still missing

Return type:

list[str]

Examples

Check what capabilities are still needed:

unsatisfied = state.get_unsatisfied_capabilities()
if unsatisfied:
    print(f"Still need: {', '.join(unsatisfied)}")
else:
    print("All capabilities satisfied!")

is_capability_satisfied(capability)

Check if a capability is satisfied.

Parameters:

capability (str) – Name of capability to check

Returns:

True if capability is satisfied (not in missing), False otherwise

Return type:

bool

Examples

Check capability status:

if state.is_capability_satisfied("math"):
    print("Math capability is available")
else:
    print("Need to activate math tool")

mark_capability_satisfied(capability)

Mark a capability as satisfied (remove from missing).

Parameters:

capability (str) – Name of capability that was satisfied

Returns:

True if capability was removed from missing, False otherwise

Return type:

bool

Examples

Mark capability as satisfied after activation:

# Activate math tool
meta_state = state.activate_component("calculator")
if meta_state:
    # Mark math capability as satisfied
    state.mark_capability_satisfied("math")

Check satisfaction status:

if state.mark_capability_satisfied("web_search"):
    print("Web search capability now satisfied")
else:
    print("Web search was not in missing capabilities")

reset_state()

Reset the activation state to initial values.

Examples

Reset state for new task:

# Clear previous state
state.reset_state()

# Set up for new task
state.current_task = "new task"
state.required_capabilities = ["new", "capabilities"]

Return type:

None

setup_dynamic_activation()

Setup dynamic activation state after model creation.

This validator: 1. Initializes discovery configuration if empty 2. Sets up execution context 3. Validates capability consistency 4. Initializes tracking systems

Return type:

Self

update_capabilities(required, missing=None)

Update required and missing capabilities.

Parameters:

• required (list[str]) – List of required capabilities

• missing (list[str] | None) – List of missing capabilities (defaults to copy of required)

Return type:

None

Examples

Update capabilities for new task:

state.update_capabilities(
    required=["math", "visualization", "data_processing"],
    missing=["visualization", "data_processing"]
)

Auto-detect missing capabilities:

state.update_capabilities(
    required=["math", "web_search", "file_processing"]
)
# missing will be set to all required capabilities

classmethod validate_current_task(v)

Validate current task is properly formatted.

Parameters:

v (str)

Return type:

str

classmethod validate_missing_capabilities(v)

Validate missing capabilities list.

Parameters:

v (list[str])

Return type:

list[str]

classmethod validate_required_capabilities(v)

Validate required capabilities list.

Parameters:

v (list[str])

Return type:

list[str]