Cogniverse Study Guide: System Integration Module¶

class TestRealVespaIntegration:
    """End-to-end system validation"""

    async def test_comprehensive_agentic_system_test(self, vespa_test_manager):
        # 1. Setup: Initialize all components across packages
        from cogniverse_foundation.config.utils import create_default_config_manager
        from cogniverse_agents.gateway_agent import GatewayAgent, GatewayDeps, GatewayInput
        from cogniverse_agents.search_agent import SearchAgent, SearchAgentDeps

        tenant_id = "test"
        config_manager = create_default_config_manager()

        # cogniverse-vespa package
        from pathlib import Path
        from cogniverse_vespa.search_backend import VespaSearchBackend
        from cogniverse_core.schemas.filesystem_loader import FilesystemSchemaLoader

        schema_loader = FilesystemSchemaLoader(Path("configs/schemas"))
        search_backend = VespaSearchBackend(
            config={
                "url": "http://localhost",
                "port": 8080,
                "profiles": {},
                "default_profiles": {},
            },
            config_manager=config_manager,
            schema_loader=schema_loader,
        )

        # cogniverse-agents package - GatewayAgent classifies simple vs.
        # complex queries with no LLM call (GLiNER + deterministic rules)
        # and routes simple queries directly to an execution agent.
        gateway = GatewayAgent(deps=GatewayDeps())

        # SearchAgent requires SearchAgentDeps with backend connection details
        # Note: profile and tenant_id are passed per-request at search() time, not at construction
        search_deps = SearchAgentDeps(
            backend_url="http://localhost",
            backend_port=8080,
        )
        # schema_loader is REQUIRED (raises ValueError if None)
        # config_manager is optional (creates default if None)
        search_agent = SearchAgent(
            deps=search_deps,
            schema_loader=schema_loader,
            config_manager=config_manager
        )

        # 2. Query Processing
        user_query = "Show me cooking videos"

        # 3. Routing Decision via GatewayAgent
        routing_result = await gateway._process_impl(
            GatewayInput(query=user_query, tenant_id=tenant_id)
        )
        assert routing_result.routed_to == "search_agent"
        assert routing_result.confidence > 0.7

        # 4. Agent Execution (SearchAgent uses search_by_text method)
        # Returns List[Dict[str, Any]] with search results
        search_results = search_agent.search_by_text(
            query=user_query,
            tenant_id=tenant_id,
            modality="video",
            top_k=10
        )

        # 5. Validation
        assert len(search_results) > 0
        # Results are list of dicts with id, score, plus metadata (video_id, etc.) at top level
        assert "id" in search_results[0]
        assert "score" in search_results[0]
        assert "video_id" in search_results[0]

# Example: Use OrchestratorAgent as the central A2A entry point
from cogniverse_agents.orchestrator_agent import OrchestratorAgent, OrchestratorDeps, OrchestratorInput
from cogniverse_core.registries.agent_registry import AgentRegistry
from cogniverse_foundation.config.utils import create_default_config_manager

tenant_id = "test_tenant"
config_manager = create_default_config_manager()

# OrchestratorAgent discovers agents via AgentRegistry (from config.json > agents section)
registry = AgentRegistry(tenant_id=tenant_id, config_manager=config_manager)
deps = OrchestratorDeps()
orchestrator = OrchestratorAgent(deps=deps, registry=registry, config_manager=config_manager)

# Execute via A2A task protocol
input_data = OrchestratorInput(query="Show me cooking videos", tenant_id=tenant_id)
result = await orchestrator._process_impl(input_data)

# Validate execution — OrchestratorOutput carries the plan, per-agent
# results, and the fused final_output (no top-level "result" field)
assert result is not None
assert result.workflow_id
assert result.final_output
assert result.agent_results

# Example: Test Vespa connection via VespaSearchBackend
from pathlib import Path
from cogniverse_vespa.search_backend import VespaSearchBackend
from cogniverse_core.schemas.filesystem_loader import FilesystemSchemaLoader
from cogniverse_foundation.config.utils import create_default_config_manager

config_manager = create_default_config_manager()
schema_loader = FilesystemSchemaLoader(Path("configs/schemas"))
search_backend = VespaSearchBackend(
    config={
        "url": "http://localhost",
        "port": 8080,
        "profiles": {},
        "default_profiles": {},
    },
    config_manager=config_manager,
    schema_loader=schema_loader,
)

# Execute search query. strategy is a rank-profile name string (e.g.
# "bm25_only"); tenant_id is required.
results = search_backend.search(
    {
        "query": "cooking tutorial",
        "type": "video",
        "strategy": "bm25_only",
        "top_k": 10,
        "tenant_id": "acme:prod",
    }
)

# Validate results (list of SearchResult with .score and .document)
assert len(results) > 0
for result in results:
    assert result.score is not None
    assert result.document.metadata["source_id"]

# Example: Phoenix telemetry validation
from cogniverse_foundation.telemetry.manager import get_telemetry_manager

telemetry_manager = get_telemetry_manager()

# Execute operation with span
with telemetry_manager.span("test_operation", tenant_id="test") as span:
    result = perform_operation()
    # Span is automatically recorded to Phoenix

# Telemetry data can be queried via Phoenix AsyncClient
# See Phoenix documentation for span query APIs

# Example: Video search integration
query = "pasta cooking tutorial"
tenant_id = "test"

# Route via GatewayAgent (GLiNER classification, no LLM call)
route = await gateway._process_impl(GatewayInput(query=query, tenant_id=tenant_id))
assert route.routed_to == "search_agent"
assert route.confidence > 0.7

# Search (using SearchAgent - returns List[Dict[str, Any]])
results = search_agent.search_by_text(query, tenant_id=tenant_id, modality="video", top_k=10)
assert len(results) > 0

# Verify (results have id, score, plus metadata like video_id at top level)
assert all("id" in r for r in results)
assert all("score" in r for r in results)
assert results[0]["score"] > 0.0

# Execute multimodal search using orchestrator
from cogniverse_agents.orchestrator_agent import OrchestratorAgent, OrchestratorDeps, OrchestratorInput
from cogniverse_core.registries.agent_registry import AgentRegistry
from cogniverse_foundation.config.utils import create_default_config_manager

query = "How does photosynthesis work?"
tenant_id = "test"

config_manager = create_default_config_manager()
registry = AgentRegistry(tenant_id=tenant_id, config_manager=config_manager)
deps = OrchestratorDeps()
orchestrator = OrchestratorAgent(deps=deps, registry=registry, config_manager=config_manager)

input_data = OrchestratorInput(query=query, tenant_id=tenant_id)
result = await orchestrator._process_impl(input_data)

# Validate orchestration occurred — final_output carries the fused
# cross-modal result, agent_results the per-agent raw outputs
assert result is not None
assert result.final_output
assert result.agent_results

# Example: Memory-enhanced routing
from cogniverse_core.memory.manager import Mem0MemoryManager
from cogniverse_core.schemas.filesystem_loader import FilesystemSchemaLoader
from pathlib import Path

tenant_id = "user123"

# 1. Initialize and add memory. Mem0MemoryManager is a per-tenant singleton;
# initialize() must run once before add_memory()/get_relevant_context() are
# usable (agents normally do this via MemoryAwareMixin.initialize_memory,
# which wraps this same call with config-driven arguments).
memory_manager = Mem0MemoryManager(tenant_id=tenant_id)
memory_manager.initialize(
    backend_host="http://localhost",
    backend_port=8080,
    llm_model="qwen2.5:0.5b",
    embedding_model="lightonai/DenseOn",
    llm_base_url="http://localhost:11434",
    embedder_base_url="http://localhost:8001",
    config_manager=config_manager,
    schema_loader=FilesystemSchemaLoader(Path("configs/schemas")),
)
memory_manager.add_memory(
    content="User prefers video tutorials",
    tenant_id=tenant_id,
    agent_name="gateway_agent"
)

# 2. Route with memory (memory automatically loaded by MemoryAwareMixin
# on agents that mix it in, e.g. GatewayAgent's downstream execution agents)
route = await gateway._process_impl(
    GatewayInput(query="Show me how to cook", tenant_id=tenant_id)
)

# 3. Verify routing decision
assert route.routed_to == "search_agent"
assert route.confidence > 0.7

# Example: Concurrent query processing.
# Reuses `orchestrator` constructed as in "Multi-Agent Orchestration" above.
# `generate_test_queries` is a test-side helper returning a list of query strings.
import asyncio
import time

queries = generate_test_queries(100)

# Execute queries concurrently
async def process_single_query(query):
    start = time.time()
    result = await orchestrator._process_impl(OrchestratorInput(query=query, tenant_id="test"))
    latency = (time.time() - start) * 1000
    return {"status": "success", "latency_ms": latency, "result": result}

tasks = [process_single_query(q) for q in queries]
results = await asyncio.gather(*tasks)

# Validate
success_rate = sum(1 for r in results if r["status"] == "success") / len(results)
assert success_rate > 0.95

# Check latencies
latencies = [r["latency_ms"] for r in results]
p95_latency = sorted(latencies)[int(len(latencies) * 0.95)]
assert p95_latency < 1000  # < 1 second

# Example: Orchestrator handles failures
from cogniverse_agents.orchestrator_agent import OrchestratorAgent, OrchestratorDeps, OrchestratorInput
from cogniverse_core.registries.agent_registry import AgentRegistry
from cogniverse_foundation.config.utils import create_default_config_manager

tenant_id = "test"
config_manager = create_default_config_manager()
registry = AgentRegistry(tenant_id=tenant_id, config_manager=config_manager)
deps = OrchestratorDeps()
orchestrator = OrchestratorAgent(deps=deps, registry=registry, config_manager=config_manager)

# Execute query that may fail
input_data = OrchestratorInput(query="test query", tenant_id=tenant_id)
result = await orchestrator._process_impl(input_data)

# On failure, orchestrator raises RuntimeError (no silent fallbacks)
assert result is not None

# Example: Verify SDK interfaces are properly implemented
from cogniverse_sdk.interfaces.backend import Backend
from cogniverse_vespa.backend import VespaBackend

# Verify Vespa implements SDK interface
assert issubclass(VespaBackend, Backend)

# Example: Verify foundation config is usable by core
from cogniverse_foundation.config.utils import create_default_config_manager
from cogniverse_core.agents.base import AgentDeps, AgentInput, AgentOutput
from cogniverse_foundation.telemetry.config import TelemetryConfig

config_manager = create_default_config_manager()

# AgentDeps carries infrastructure config (no tenant_id — it's per-request)
# Concrete agents use extended Deps classes:
# OrchestratorDeps(AgentDeps) adds orchestrator-specific config
# SearchAgentDeps(AgentDeps) adds: backend_url, backend_port, etc.
deps = AgentDeps()  # No tenant_id at construction

# Example: Verify core works with evaluation package
from cogniverse_agents.orchestrator_agent import OrchestratorAgent, OrchestratorDeps, OrchestratorInput
from cogniverse_core.registries.agent_registry import AgentRegistry
from cogniverse_evaluation.core.experiment_tracker import ExperimentTracker
from cogniverse_foundation.config.utils import create_default_config_manager

tenant_id = "test"
config_manager = create_default_config_manager()
registry = AgentRegistry(tenant_id=tenant_id, config_manager=config_manager)
agent = OrchestratorAgent(deps=OrchestratorDeps(), registry=registry, config_manager=config_manager)

# Create experiment tracker for evaluation
# Note: tenant_id is REQUIRED (no default) — experiment_project_name defaults to "experiments"
tracker = ExperimentTracker(
    experiment_project_name="test_exp",
    tenant_id=tenant_id
)

# Execute orchestration query (tracked via telemetry)
result = await agent._process_impl(OrchestratorInput(query="test query", tenant_id=tenant_id))
assert result.workflow_id

# Example: Verify Phoenix telemetry plugin integration
from cogniverse_foundation.telemetry.registry import TelemetryRegistry
from cogniverse_telemetry_phoenix.provider import PhoenixProvider

# Verify plugin registration via entry points (classmethod `get`, tenant-scoped)
# Requires config with http_endpoint and grpc_endpoint
provider = TelemetryRegistry.get(
    name="phoenix",
    tenant_id=tenant_id,
    config={"http_endpoint": "http://localhost:6006", "grpc_endpoint": "http://localhost:4317"}
)
assert isinstance(provider, PhoenixProvider)

# Example: Verify agents work with Vespa backend
from cogniverse_agents.search_agent import SearchAgent, SearchAgentDeps
from cogniverse_vespa.vespa_schema_manager import VespaSchemaManager
from cogniverse_foundation.config.utils import create_default_config_manager

tenant_id = "test"

# Setup tenant schemas
config_manager = create_default_config_manager()
schema_mgr = VespaSchemaManager(backend_endpoint="http://localhost", backend_port=8080)
tenant_schema = schema_mgr.get_tenant_schema_name(tenant_id, "video_colpali_smol500_mv_frame")

# Initialize agent with Vespa backend
search_deps = SearchAgentDeps(
    backend_url="http://localhost",
    backend_port=8080,
)
# profile is per-request, not at construction; tenant_id is per-request on search_by_text()
# schema_loader is REQUIRED (raises ValueError if None)
from pathlib import Path
from cogniverse_core.schemas.filesystem_loader import FilesystemSchemaLoader
schema_loader = FilesystemSchemaLoader(Path("configs/schemas"))
agent = SearchAgent(
    deps=search_deps,
    schema_loader=schema_loader,
    config_manager=config_manager
)

results = agent.search_by_text("test query", tenant_id=tenant_id, modality="video")
assert len(results) >= 0

# Example: Verify agents work with synthetic data
from cogniverse_agents.orchestrator_agent import OrchestratorAgent, OrchestratorDeps, OrchestratorInput
from cogniverse_core.registries.agent_registry import AgentRegistry
from cogniverse_synthetic.generators.routing import RoutingGenerator
from cogniverse_foundation.config.unified_config import (
    DSPyModuleConfig,
    OptimizerGenerationConfig,
)

# RoutingGenerator requires OptimizerGenerationConfig as REQUIRED parameter
# Configuration is REQUIRED - no fallbacks or defaults
# Create minimal config for testing (production would load from configs/config.json)
optimizer_config = OptimizerGenerationConfig(
    optimizer_type="routing",
    dspy_modules={
        "query_generator": DSPyModuleConfig(
            signature_class="cogniverse_synthetic.dspy_signatures.GenerateEntityQuery"
        )
    },
    profile_scoring_rules=[],
    agent_mappings=[],
)
# Production entity and routing callbacks are required; the generator never
# substitutes a local heuristic for either label.
generator = RoutingGenerator(
    entity_extractor=production_entity_extractor,
    routing_decider=production_routing_decider,
    optimizer_config=optimizer_config,
)

sampled_content = [
    {"video_id": "v1", "title": "Cooking tutorial", "description": "Learn to cook"},
    {"video_id": "v2", "title": "Science lecture", "description": "Physics explained"}
]
synthetic_data = await generator.generate(
    sampled_content=sampled_content,
    target_count=2,
    tenant_id=tenant_id,
)

# Test orchestrator with synthetic queries
orchestrator = OrchestratorAgent(
    deps=OrchestratorDeps(),
    registry=AgentRegistry(tenant_id=tenant_id, config_manager=config_manager),
    config_manager=config_manager,
)
for example in synthetic_data:
    result = await orchestrator._process_impl(
        OrchestratorInput(query=example.query, tenant_id=tenant_id)
    )
    assert result is not None
    assert result.workflow_id

# Example: Verify runtime integrates all layers
from cogniverse_runtime.main import app
from fastapi.testclient import TestClient

# Dependency overrides (config_manager, schema_loader) are wired in the
# app's lifespan handler, so TestClient must be used as a context manager
# to trigger startup — a bare `TestClient(app)` skips lifespan and every
# route depending on those overrides returns 503.
with TestClient(app) as client:
    # Test end-to-end search request
    response = client.post(
        "/search/",
        json={"query": "cooking videos", "tenant_id": "test", "top_k": 10}
    )

    assert response.status_code == 200
    assert "results" in response.json()

# Example: experiment tracking against Phoenix
from cogniverse_evaluation.core.experiment_tracker import ExperimentTracker

# Create experiment tracker (tenant_id is REQUIRED — no default)
tracker = ExperimentTracker(
    experiment_project_name="dash_test",
    tenant_id="test"
)

# Phoenix persistent-data management (backup/restore/clean) is handled by
# the scripts/manage_phoenix_data.py CLI.