Cogniverse Study Guide: Configuration Management¶

from cogniverse_foundation.config.unified_config import SystemConfig

# SystemConfig is global infrastructure state (one per deployment) — it
# has no tenant_id field. Per-tenant settings live in RoutingConfigUnified,
# AgentConfig, and TelemetryConfig instead (see below).
system_config = SystemConfig(
    llm_model="gpt-4",
    base_url="https://api.openai.com/v1",
    backend_url="http://localhost",
    backend_port=8080,
    telemetry_url="http://localhost:6006",
)

from cogniverse_foundation.config.unified_config import LLMEndpointConfig, LLMConfig

# Primary (student) LLM — used at runtime by all agents
primary = LLMEndpointConfig(
    model="openai/google/gemma-4-e4b-it",
    api_base="http://localhost:11434/v1",   # in-cluster vLLM or Ollama
    temperature=0.1,
    max_tokens=1000,
    request_timeout=120.0,   # fail-fast timeout in seconds (default: 120.0)
    num_retries=1,            # total attempts per call (default: 1)
)

# Teacher LLM — used only during DSPy optimization (BootstrapFewShot's teacher_settings)
teacher = LLMEndpointConfig(
    model="openai/cyankiwi/Qwen3.6-27B-AWQ-INT4",
    api_base="http://localhost:29011/v1",
    request_timeout=120.0,
    num_retries=1,
)

llm_config = LLMConfig(primary=primary, teacher=teacher)

from cogniverse_foundation.config.llm_factory import create_dspy_lm

lm = create_dspy_lm(primary)  # wires api_base/api_key/temperature/seed/extra_headers onto dspy.LM

from cogniverse_foundation.config.semantic_router import routed_lm_context_for

with routed_lm_context_for(config_manager, tenant_id="acme_corp", agent_name="search_agent"):
    ...  # DSPy calls inside this block use the resolved (routed or direct) LM

from cogniverse_foundation.config.semantic_router import routed_lm_context_for_async

routed_lm = await routed_lm_context_for_async(config_manager, tenant_id="acme_corp", agent_name="search_agent")
with routed_lm:
    ...

from cogniverse_foundation.config.agent_config import AgentConfig, ModuleConfig, DSPyModuleType, OptimizerConfig, OptimizerType

agent_config = AgentConfig(
    agent_name="video_search_agent",
    agent_version="1.0.0",
    agent_description="Video search agent with ReAct module",
    agent_url="http://localhost:8002",
    capabilities=["video_search", "rerank"],
    skills=[{"name": "vespa_search"}, {"name": "rerank"}],
    module_config=ModuleConfig(
        module_type=DSPyModuleType.REACT,
        signature="Question -> Answer",
        max_retries=5,
        temperature=0.7
    ),
    optimizer_config=OptimizerConfig(
        optimizer_type=OptimizerType.GEPA,
        max_bootstrapped_demos=4,
        max_labeled_demos=16,
        num_trials=10
    ),
    llm_model="gpt-4",
    llm_temperature=0.7
)

from cogniverse_foundation.config.utils import create_default_config_manager

# Uses the configured backend (e.g., VespaConfigStore)
manager = create_default_config_manager()

# Get the global system configuration (not per-tenant)
system_config = manager.get_system_config()
print(f"LLM: {system_config.llm_model}")
print(f"Backend: {system_config.backend_url}:{system_config.backend_port}")

from cogniverse_vespa.config.config_store import VespaConfigStore
from cogniverse_foundation.config.manager import ConfigManager

# Initialize Vespa store with URL and port
vespa_store = VespaConfigStore(
    backend_url="http://localhost",
    backend_port=8080
)

# Use with ConfigManager (tenant isolation handled via document IDs)
manager = ConfigManager(store=vespa_store)

from cogniverse_sdk.interfaces.config_store import ConfigStore, ConfigScope, ConfigEntry
from typing import Dict, Any, Optional, List
from datetime import datetime
import redis

class RedisConfigStore(ConfigStore):
    """Redis-based configuration storage - implements sdk interface"""

    def __init__(self, redis_url: str):
        self.redis_client = redis.from_url(redis_url)
        self.initialize()

    def initialize(self) -> None:
        """Setup Redis indexes and structures"""
        # Create sorted sets for versioning
        # Setup key-value structures
        pass

    def set_config(
        self,
        tenant_id: str,
        scope: ConfigScope,
        service: str,
        config_key: str,
        config_value: Dict[str, Any],
    ) -> ConfigEntry:
        """Store configuration with versioning"""
        # Generate version number
        # Store in Redis with TTL
        # Publish update event
        pass

    def get_config(
        self,
        tenant_id: str,
        scope: ConfigScope,
        service: str,
        config_key: str,
        version: Optional[int] = None,
    ) -> Optional[ConfigEntry]:
        """Retrieve configuration by version"""
        # Get from Redis cache
        # Deserialize JSON
        # Return ConfigEntry
        pass

from cogniverse_foundation.config.utils import create_default_config_manager, get_config
from cogniverse_foundation.config.unified_config import RoutingConfigUnified

manager = create_default_config_manager()

# Configure per-tenant routing for Tenant A (enterprise customer)
# routing_mode is "tiered" for both tenants below — it's the only mode
# implemented end-to-end today; other values are accepted by the schema
# for forward compatibility but produce no behavior change at dispatch time.
tenant_a_routing = RoutingConfigUnified(
    tenant_id="acme_corp",
    routing_mode="tiered",
    fast_path_confidence_threshold=0.7,
    cache_ttl_seconds=300,
    optimizer_floors={
        "profile_selection": {
            "min_samples_for_optimization": 20,
            "min_unique_queries": 6,
        },
        "entity_extraction": {
            "min_samples_for_optimization": 58,
            "min_unique_queries": 15,
        },
    },
)
manager.set_routing_config(tenant_a_routing, tenant_id="acme_corp")

# Configure per-tenant routing for Tenant B (different customer)
tenant_b_routing = RoutingConfigUnified(
    tenant_id="globex_inc",
    routing_mode="tiered",
    fast_path_confidence_threshold=0.8,
    cache_ttl_seconds=600,
)
manager.set_routing_config(tenant_b_routing, tenant_id="globex_inc")

# Per-tenant configurations are completely isolated
config_a = get_config(tenant_id="acme_corp", config_manager=manager)
config_b = get_config(tenant_id="globex_inc", config_manager=manager)
assert config_a["cache_ttl_seconds"] != config_b["cache_ttl_seconds"]

# Each tenant gets isolated Vespa schemas
# acme_corp → video_colpali_mv_frame_acme_corp
# globex_inc → video_colpali_mv_frame_globex_inc

from cogniverse_foundation.config.unified_config import RoutingConfigUnified
from cogniverse_foundation.config.utils import get_config

# Create per-tenant routing configuration for a new tenant
new_tenant_routing = RoutingConfigUnified(
    tenant_id="new_tenant",
    routing_mode="tiered",
    fast_path_confidence_threshold=0.7,
)
manager.set_routing_config(new_tenant_routing, tenant_id="new_tenant")

# Copy per-tenant config from one tenant to another
source_config = get_config(tenant_id="tenant_a", config_manager=manager)
staging_routing = RoutingConfigUnified(
    tenant_id="tenant_a_staging",
    routing_mode=source_config.get("routing_mode", "tiered"),
)
manager.set_routing_config(staging_routing, tenant_id="tenant_a_staging")

# Delete tenant configuration (removes all versions)
from cogniverse_sdk.interfaces.config_store import ConfigScope
manager.store.delete_config(
    tenant_id="old_tenant",
    scope=ConfigScope.ROUTING,
    service="gateway_agent",
    config_key="routing_config"
)

from cogniverse_foundation.config.agent_config import AgentConfig, ModuleConfig, DSPyModuleType, OptimizerConfig, OptimizerType
from cogniverse_foundation.config.utils import create_default_config_manager

manager = create_default_config_manager()

# Configure Video Search Agent with ReAct (per tenant)
video_agent_config = AgentConfig(
    agent_name="video_search_agent",
    agent_version="1.0.0",
    agent_description="Video search agent with ReAct module",
    agent_url="http://localhost:8002",
    capabilities=["video_search", "rerank", "summarize"],
    skills=[{"name": "vespa_search"}, {"name": "rerank"}],
    module_config=ModuleConfig(
        module_type=DSPyModuleType.REACT,
        signature="Question -> Answer",
        max_retries=5,
        temperature=0.7
    ),
    optimizer_config=OptimizerConfig(
        optimizer_type=OptimizerType.GEPA,
        max_bootstrapped_demos=4,
        max_labeled_demos=16,
        num_trials=10
    ),
    llm_model="gpt-4",
    llm_temperature=0.7
)

manager.set_agent_config(
    tenant_id="acme_corp",
    agent_name="video_search_agent",
    agent_config=video_agent_config
)

from cogniverse_sdk.interfaces.config_store import ConfigScope

# Get configuration history (SystemConfig stored under "_system" sentinel tenant)
history = manager.store.get_config_history(
    tenant_id="_system",
    scope=ConfigScope.SYSTEM,
    service="system",
    config_key="system_config",
    limit=10
)

for entry in history:
    print(f"Version {entry.version}:")
    print(f"  Updated: {entry.updated_at}")
    print(f"  Changes: {entry.config_value}")

from cogniverse_sdk.interfaces.config_store import ConfigScope
from cogniverse_foundation.config.unified_config import SystemConfig

# Get current global system config (no tenant_id argument)
current = manager.get_system_config()
print(f"Current LLM: {current.llm_model}")

# Rollback to specific version by retrieving and re-applying old config
# SystemConfig is stored under the "_system" sentinel tenant
old_entry = manager.store.get_config(
    tenant_id="_system",
    scope=ConfigScope.SYSTEM,
    service="system",
    config_key="system_config",
    version=5  # Specific version to rollback to
)

if old_entry:
    # Re-apply old configuration
    old_config = SystemConfig.from_dict(old_entry.config_value)
    manager.set_system_config(old_config)

    # Verify rollback
    rolled_back = manager.get_system_config()
    print(f"Rolled back LLM: {rolled_back.llm_model}")

import json
from datetime import datetime

# Export all configurations
export_data = manager.store.export_configs(
    tenant_id="production",
    include_history=True
)

# Save with timestamp
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
with open(f"config_backup_{timestamp}.json", "w") as f:
    json.dump(export_data, f, indent=2, default=str)

print(f"Exported {len(export_data['configs'])} configurations")

# Load backup
with open("config_backup_20250104_120000.json", "r") as f:
    backup_data = json.load(f)

# Import to new environment
imported_count = manager.store.import_configs(
    tenant_id="staging",
    configs=backup_data
)

print(f"Imported {imported_count} configurations")

# Check storage backend health
if manager.store.health_check():
    print("✓ Configuration storage healthy")
else:
    print("✗ Configuration storage unavailable")

# Get storage statistics
stats = manager.store.get_stats()
print(f"Total configurations: {stats['total_configs']}")
print(f"Total tenants: {stats['total_tenants']}")
print(f"Total versions: {stats['total_versions']}")
print(f"Configs by scope: {stats['configs_per_scope']}")
print(f"Storage backend: {stats['storage_backend']}")

# ✅ Good: SystemConfig is global — call with no arguments
system_config = manager.get_system_config()
print(f"Backend: {system_config.backend_url}:{system_config.backend_port}")

# ✅ Good: Per-tenant config uses get_config() with explicit tenant_id
from cogniverse_foundation.config.utils import get_config
tenant_config = get_config(tenant_id="acme_corp", config_manager=manager)

# ❌ Bad: get_system_config does not accept a tenant_id argument
# config = manager.get_system_config(tenant_id="production")  # WRONG

# Before major changes, export current configuration
backup = manager.store.export_configs(
    tenant_id="_system",
    include_history=True
)

# Save backup to file
import json
from datetime import datetime
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
with open(f"config_backup_{timestamp}.json", "w") as f:
    json.dump(backup, f, indent=2, default=str)

# Make changes (a new version is created automatically on write)
from cogniverse_foundation.config.unified_config import SystemConfig
current = manager.get_system_config()
updated = SystemConfig.from_dict({**current.to_dict(), "llm_model": "gpt-4-turbo"})
manager.set_system_config(updated)

# ✅ Good: Type-safe dataclass (SystemConfig has no tenant_id field)
from cogniverse_foundation.config.unified_config import SystemConfig
config = SystemConfig(
    llm_model="gpt-4",
    backend_url="http://localhost",
    backend_port=8080
)

# ❌ Bad: Raw dictionaries
config = {"llm_model": "gpt-4"}  # No validation

# Define reusable templates for different deployment environments
# (SystemConfig is global — no tenant_id)
TEMPLATES = {
    "development": SystemConfig(
        llm_model="gpt-3.5-turbo",
        base_url="http://localhost:11434",
        backend_url="http://localhost",
        backend_port=8080,
        telemetry_url="http://localhost:6006"
    ),
    "production": SystemConfig(
        llm_model="gpt-4",
        base_url="https://api.openai.com/v1",
        backend_url="http://production-vespa",
        backend_port=8080,
        telemetry_url="http://production-phoenix:6006"
    )
}

# Apply a template to set the global system config
template = TEMPLATES["production"]
new_config = SystemConfig(
    llm_model="claude-3-opus",  # Override
    base_url=template.base_url,
    backend_url=template.backend_url,
    backend_port=template.backend_port,
    telemetry_url=template.telemetry_url
)
manager.set_system_config(new_config)

from cogniverse_sdk.interfaces.config_store import ConfigScope
from cogniverse_foundation.config.unified_config import SystemConfig

# Check if global system configuration exists
configs = manager.store.list_configs(
    tenant_id="_system",
    scope=ConfigScope.SYSTEM
)
print(f"Available configs: {configs}")

# Initialize missing system configuration (no tenant_id — SystemConfig is global)
existing_config = manager.store.get_config(
    tenant_id="_system",
    scope=ConfigScope.SYSTEM,
    service="system",
    config_key="system_config"
)
if not existing_config:
    manager.set_system_config(SystemConfig())

from cogniverse_foundation.config.unified_config import SystemConfig

# Handle concurrent updates by checking version history
current_entry = manager.store.get_config(
    tenant_id="_system",
    scope=ConfigScope.SYSTEM,
    service="system",
    config_key="system_config"
)
if current_entry is None:
    raise RuntimeError("system_config not initialized — call set_system_config first")

# Make your changes
updated_config = SystemConfig.from_dict(current_entry.config_value)
updated_config.llm_model = "gpt-4-turbo"

# Apply update (creates new version automatically)
manager.set_system_config(updated_config)

# Verify version incremented
new_entry = manager.store.get_config(
    tenant_id="_system",
    scope=ConfigScope.SYSTEM,
    service="system",
    config_key="system_config"
)
assert new_entry.version == current_entry.version + 1

# create_default_config_manager() reads BACKEND_URL/BACKEND_PORT (env) and
# backend.type (configs/config.json) via BootstrapConfig — it raises
# ValueError if BACKEND_URL is unset or backend.type is anything other
# than "vespa" (the only supported backend). There is no silent fallback.
from cogniverse_foundation.config.utils import create_default_config_manager

manager = create_default_config_manager()