cleveractors-core/docs/registry/integration.md

Integration Examples

End-to-end workflows that combine multiple modules from the Package Registry subsystem. Each example is a complete, copy-paste-runnable script (aside from the registry server URL, which must be configured for your environment).

---

1. Electronic Component Ordering Pipeline

Registers a supplier agent that validates part numbers in purchase orders by resolving an order-validator template from the package registry at startup.

This workflow demonstrates the full pipeline: parse reference → resolve (version alias) → fetch → cache → instantiate agent.

"""
Electronic component ordering pipeline.

Resolves an order-validator template from the registry, fetches its
full content, and produces a supplier agent configuration. Illustrates
the complete client → cache → resolver pipeline.
"""

import asyncio
from typing import Any

from cleveractors.registry import (
    CacheFactory,
    PackageContentResolver,
    RegistryCache,
    RegistryClient,
    ReferenceResolver,
)
from cleveractors.registry.exceptions import (
    InvalidPackageReferenceError,
    PackageNotFoundError,
    RegistryError,
    RegistryNetworkError,
)
from cleveractors.registry.types import PackageReference, ReferenceType


async def order_pipeline() -> dict[str, Any]:
    client = RegistryClient(
        base_url="https://registry.example.com",
        api_key="supplier-api-key",
    )

    # Transparent content-level cache per §10.3
    cache_factory = CacheFactory(max_size=512, ttl=300.0)
    cache = cache_factory.create(client)

    # Resolver for parsing and resolving references
    resolver = ReferenceResolver(client=client)

    # 1. Parse the reference string
    ref_str = "registry.example.com:acme/order-validator@v1.x"
    ref: PackageReference = resolver.parse(ref_str)

    if ref.reference_type != ReferenceType.REGISTRY:
        raise RuntimeError(f"Expected registry reference, got {ref.reference_type}")

    print(f"[1] Parsed reference: {ref.namespace}/{ref.name}@{ref.version}")

    # 2. Resolve the version alias to a concrete Package ID
    resolved = await cache.resolve_package(
        package_type="actor",
        namespace=ref.namespace,
        name=ref.name,
        version=ref.version,
    )
    package_id = resolved["package_id"]
    print(f"[2] Resolved {ref.version} → {package_id}")

    # 3. Fetch full package content (cache hit if already fetched)
    content = await cache.get_package(package_id)
    print(f"[3] Fetched content: {content.get('name', package_id)}")

    # 4. Build supplier agent config from the template
    agent_config = {
        "type": "supplier",
        "template_package_id": package_id,
        "validators": content.get("validators", []),
        "cache_stats": {
            "hits": cache.stats.hits,
            "misses": cache.stats.misses,
            "evictions": cache.stats.evictions,
        },
    }
    print(f"[4] Agent configured with {len(agent_config['validators'])} validators")
    print(f"    Cache: hits={cache.stats.hits} misses={cache.stats.misses}")

    await cache.close()
    return agent_config


if __name__ == "__main__":
    config = asyncio.run(order_pipeline())
    print(f"\nDone. Supplier agent config: {list(config.keys())}")

---

2. Email Categorization System

Resolves categorization templates from a registry, fetches tool packages for domain-specific agents, caches for repeated use, and gracefully handles missing or inaccessible packages.

"""
Email categorization system.

Resolves multiple skill packages from the registry, loads tool packages
for domain-specific agents, and builds a categorisation pipeline. Uses
the content resolver for multi-server client pooling and transparent
two-tier caching.
"""

import asyncio
from typing import Any

from cleveractors.registry import (
    CacheFactory,
    PackageContentResolver,
    RegistryClient,
)
from cleveractors.registry.exceptions import (
    AccessDeniedError,
    AuthenticationRequiredError,
    PackageNotFoundError,
    RegistryNetworkError,
    VersionNotFoundError,
)
from cleveractors.registry.types import PackageReference


CATEGORY_TEMPLATES = [
    ("skill", "acme", "email-categorizer", "v1.x"),
    ("skill", "acme", "spam-detector", "latest"),
    ("skill", "acme", "priority-classifier", "v2.0.0"),
    ("actor", "acme", "response-generator", "v1.0.0"),
]

TOOL_PACKAGES = [
    "pkg_skl_aaa1111111111111111111111111111111111",
    "pkg_skl_bbb2222222222222222222222222222222222",
]


async def build_categorization_system() -> dict[str, Any]:
    client = RegistryClient(
        base_url="https://registry.example.com",
        api_key="categorizer-key",
    )

    cache_factory = CacheFactory(max_size=256, ttl=600.0)
    resolver = PackageContentResolver(
        api_key="categorizer-key",
        cache_factory=cache_factory,
    )

    # Manually pre-warm the resolver's client pool
    resolver.clients["https://registry.example.com"] = client

    # Phase 1: Resolve template packages
    template_contents: dict[str, dict] = {}
    failed_templates: list[str] = []

    for pkg_type, ns, name, version in CATEGORY_TEMPLATES:
        ref = PackageReference.from_string(
            f"registry.example.com:{ns}/{name}@{version}"
        )
        try:
            content = await resolver.aresolve(ref, package_type=pkg_type)
            if content is not None:
                template_contents[f"{ns}/{name}"] = content
                print(
                    f"  ✓ {ns}/{name}@{version} → "
                    f"{content.get('_package_id', 'unknown')}"
                )
            else:
                failed_templates.append(f"{ns}/{name}")
        except (PackageNotFoundError, VersionNotFoundError) as exc:
            print(f"  ✗ {ns}/{name}@{version}: {exc}")
            failed_templates.append(f"{ns}/{name}")
        except AccessDeniedError:
            print(f"  ✗ {ns}/{name}@{version}: access denied")
            failed_templates.append(f"{ns}/{name}")

    # Phase 2: Load tool packages from cache (second fetch hits cache)
    print("\nLoading tool packages...")
    tool_contents: dict[str, dict] = {}
    for pkg_id in TOOL_PACKAGES:
        try:
            content = await resolver.aresolve(
                PackageReference.from_string(f"ID:{pkg_id}"),
                package_type="skill",
            )
            if content is not None:
                tool_contents[pkg_id] = content
                print(f"  ✓ {pkg_id}")
        except Exception as exc:
            print(f"  ✗ {pkg_id}: {exc}")

    # Phase 3: Aggregate cache statistics
    stats = resolver.total_stats
    print(
        f"\nCache summary: hits={stats.hits} misses={stats.misses}"
        f" evictions={stats.evictions}"
    )

    await resolver.close_all()
    return {
        "templates_loaded": len(template_contents),
        "tools_loaded": len(tool_contents),
        "failed_templates": failed_templates,
        "cache_stats": {
            "hits": stats.hits,
            "misses": stats.misses,
            "evictions": stats.evictions,
        },
    }


if __name__ == "__main__":
    result = asyncio.run(build_categorization_system())
    print(f"\nDone: {result}")

---

3. CI/CD Pipeline Actor Verification

Canonicalizes an actor configuration, compares hashes across environments, and detects configuration drift. Demonstrates the canonicalization pipeline and content validation workflow.

"""
CI/CD pipeline actor verification.

Compares actor configurations across staging and production environments
using content-addressed SHA-1 hashes. Detects drift, verifies integrity
of cached templates, and reports mismatches with full Package IDs.
"""

import asyncio
import hashlib
from typing import Any

from cleveractors.registry import (
    Canonicalizer,
    PackageType,
    RegistryCache,
    RegistryClient,
)
from cleveractors.registry.types import PackageId


def canonical_hash(content: dict[str, Any]) -> str:
    """Compute the canonical SHA-1 hash of a content dict."""
    canon = Canonicalizer()
    canonical = canon.canonicalize(content)
    return hashlib.sha1(
        canonical.encode("utf-8"), usedforsecurity=False
    ).hexdigest()


async def verify_deployment(
    staging_url: str,
    production_url: str,
    package_id_str: str,
) -> dict[str, Any]:
    """
    Verify that a deployed actor config is identical across environments.

    Args:
        staging_url: URL of the staging registry.
        production_url: URL of the production registry.
        package_id_str: The Package ID to verify.
    """
    pid = PackageId.from_string(package_id_str)
    result: dict[str, Any] = {"package_id": pid.id_string}

    # Fetch from both environments
    async with (
        RegistryClient(base_url=staging_url) as staging_client,
        RegistryClient(base_url=production_url) as prod_client,
    ):
        staging_cache = RegistryCache(staging_client, max_size=1, ttl=60.0)
        prod_cache = RegistryCache(prod_client, max_size=1, ttl=60.0)

        staging_content = await staging_cache.get_package(package_id_str)
        prod_content = await prod_cache.get_package(package_id_str)

        result["staging_content_name"] = staging_content.get("name")
        result["prod_content_name"] = prod_content.get("name")

        # Compute canonical hashes
        staging_hash = canonical_hash(staging_content)
        prod_hash = canonical_hash(prod_content)
        result["staging_hash"] = staging_hash
        result["prod_hash"] = prod_hash

        # Verify canonical hash matches the ID
        canon = Canonicalizer()
        computed_id = canon.compute_package_id(staging_content, pid.package_type)
        result["hash_matches_id"] = computed_id.sha1_hex == pid.sha1_hex

        # Compare environments
        if staging_hash != prod_hash:
            result["drift_detected"] = True
            result["staging_canonical_id"] = computed_id.id_string

            computed_prod = canon.compute_package_id(
                prod_content, pid.package_type
            )
            result["prod_canonical_id"] = computed_prod.id_string

            print(
                f"DRIFT DETECTED for {package_id_str}\n"
                f"  Staging hash: {staging_hash}\n"
                f"  Production hash: {prod_hash}"
            )
        else:
            result["drift_detected"] = False
            print(f"No drift — {pid.id_string} matches across environments")

        return result


async def main() -> None:
    PKG_ID = "pkg_act_a1b2c3d4e5f67890abcdef1234567890abcdef"

    result = await verify_deployment(
        staging_url="https://staging-registry.example.com",
        production_url="https://registry.example.com",
        package_id_str=PKG_ID,
    )

    print(f"\nVerification result:")
    for key, value in result.items():
        print(f"  {key}: {value}")


if __name__ == "__main__":
    asyncio.run(main())

---

4. Multi-Tenant SaaS Provisioning

Resolves agent templates per tenant namespace, validates content integrity via SHA-1, detects tampered cache entries, and provisions tenant-specific agent configurations.

"""
Multi-tenant SaaS provisioning.

Provisions agent configurations for multiple tenants by resolving
tenant-specific templates from the registry. Validates content integrity
and detects tampered or stale cache entries.
"""

import asyncio
from typing import Any

from cleveractors.registry import (
    CacheFactory,
    Canonicalizer,
    PackageContentResolver,
    RegistryClient,
    PackageType,
)
from cleveractors.registry.exceptions import (
    RegistryError,
    RegistryNetworkError,
)
from cleveractors.registry.types import PackageReference


TENANTS = {
    "tenant-alpha": "registry.example.com:alpha/workflow-agent@v2.x",
    "tenant-beta": "registry.example.com:beta/workflow-agent@v1.x",
    "tenant-gamma": "registry.example.com:gamma/workflow-agent@latest",
}


async def provision_tenants() -> dict[str, dict[str, Any]]:
    client = RegistryClient(
        base_url="https://registry.example.com",
        api_key="provisioning-key",
    )
    cache_factory = CacheFactory(max_size=512, ttl=300.0)
    resolver = PackageContentResolver(
        api_key="provisioning-key",
        cache_factory=cache_factory,
    )
    resolver.clients["https://registry.example.com"] = client
    canon = Canonicalizer()

    results: dict[str, dict[str, Any]] = {}

    for tenant_id, ref_str in TENANTS.items():
        print(f"\nProvisioning {tenant_id}...")
        ref = PackageReference.from_string(ref_str)

        try:
            # Resolve via the two-tier cache
            content = await resolver.aresolve(ref, package_type="actor")
            if content is None:
                print(f"  ✗ Failed to resolve {ref_str}")
                results[tenant_id] = {"status": "failed", "reason": "not_found"}
                continue

            package_id_str = content.get("_package_id", "unknown")

            # Validate content integrity
            canon.compute_package_id(
                content, package_type=PackageType.ACTOR
            )

            results[tenant_id] = {
                "status": "provisioned",
                "package_id": package_id_str,
                "namespace": ref.namespace,
                "template_name": content.get("name", "unknown"),
            }
            print(f"  ✓ Resolved to {package_id_str}")

        except RegistryNetworkError as exc:
            print(f"  ✗ Network error: {exc}")
            results[tenant_id] = {
                "status": "failed",
                "reason": f"network: {exc.status_code}",
            }
        except RegistryError as exc:
            print(f"  ✗ Registry error: {exc}")
            results[tenant_id] = {
                "status": "failed",
                "reason": type(exc).__name__,
            }

    # Report aggregated cache statistics
    stats = resolver.total_stats
    print(
        f"\nAggregated cache: hits={stats.hits} "
        f"misses={stats.misses} evictions={stats.evictions}"
    )

    await resolver.close_all()
    return results


if __name__ == "__main__":
    results = asyncio.run(provision_tenants())
    provisioned = sum(
        1 for r in results.values() if r["status"] == "provisioned"
    )
    print(f"\nProvisioned {provisioned}/{len(results)} tenants")