cleveractors-core/docs/registry/cache.md

RegistryCache

Transparent client-side LRU cache with TTL expiration and SHA-1 tamper detection, implementing Package Registry Standard v1.0.0 §10.3.

from cleveractors.registry import RegistryCache, CacheFactory, CacheStats

---

Overview

RegistryCache wraps a RegistryClient and adds:

Feature	Mechanism
LRU eviction	Least-recently-used entry evicted when max_size exceeded
TTL expiration	Entries older than ttl seconds are re-fetched
SHA-1 tamper detection	Cached content hash is validated against stored PackageId
Singleflight coalescing	Concurrent cold-key requests share a single upstream fetch
Thread safety	All operations protected by asyncio.Lock

---

Constructor

RegistryCache(
    client: RegistryClient,
    *,
    max_size: int = 256,
    ttl: float = 300.0,
    validate_content: bool = True,
)

Parameter	Type	Default	Description
client	RegistryClient	required	The upstream client to cache responses for
max_size	int	256	Maximum number of cached entries before LRU eviction
ttl	float	300.0	Time-to-live in seconds before re-fetch
validate_content	bool	True	Whether to validate cached content by recomputing SHA-1

---

CacheStats

Observable cache statistics exposed via RegistryCache.stats:

@dataclass
class CacheStats:
    hits: int = 0
    misses: int = 0
    evictions: int = 0

    def record_hit(self) -> None: ...
    def record_miss(self) -> None: ...
    def record_eviction(self) -> None: ...
    def reset(self) -> None: ...

cache = RegistryCache(client, max_size=512, ttl=600.0)
# ... perform operations ...
print(f"hits={cache.stats.hits} misses={cache.stats.misses} evictions={cache.stats.evictions}")

---

Methods

get_package

async def get_package(self, package_id: str) -> dict[str, Any]

Retrieve package content, using cache when available and valid.

Cache-hit flow:

1. Entry found in cache and within TTL → validate SHA-1 (CPU-bound, outside lock).
2. Hash matches → return cached content.
3. Hash mismatch → evict (tampered), re-fetch from upstream.

Cache-miss flow:

1. Entry missing, expired, or tampered.
2. Concurrent callers for the same key share a single upstream fetch (singleflight).
3. Fetch from upstream, populate cache, return.

resolve_package

async def resolve_package(
    self,
    package_type: str,
    namespace: str,
    name: str,
    version: str = "latest",
) -> dict[str, Any]

Mirrors RegistryClient.resolve_package with transparent caching. Maintains a separate resolve-store keyed by (type, namespace, name, version).

put

async def put(self, package_id: str, content: dict[str, Any]) -> None

Insert content directly into the cache without an upstream fetch. Useful when content was obtained through another channel (e.g. resolve_package) and you want to warm the cache for subsequent get_package hits.

invalidate

async def invalidate(self, package_id: str) -> bool

Remove a specific entry from the cache. Returns True if the entry was present.

removed = await cache.invalidate("pkg_act_...")
print(f"Cache entry removed: {removed}")

clear

async def clear(self) -> None

Clear all entries from both stores (package content and resolution) and reset statistics.

close

async def close(self) -> None

Clear caches, reset statistics, and close the underlying RegistryClient. If you need post-mortem statistics, snapshot cache.stats before calling close().

__contains__

def __contains__(self, package_id: str) -> bool

Raw membership test; returns True if the key is present (does not check TTL).

if "pkg_act_..." in cache:
    print("Entry exists in cache (may be expired)")

---

Async Context Manager

async with RegistryCache(client, max_size=512) as cache:
    content = await cache.get_package("pkg_act_...")
    print(f"Final stats: hits={cache.stats.hits}")
# cache.close() called automatically

---

CacheFactory

Encapsulates cache configuration so consumers depend on the factory abstraction rather than constructing RegistryCache directly.

factory = CacheFactory(max_size=1000, ttl=900.0, validate_content=True)
cache = factory.create(client)

Parameter	Type	Default	Description
max_size	int	256	Maximum cache entries
ttl	float	300.0	TTL in seconds
validate_content	bool	True	Enable SHA-1 validation

---

Real-Life Example: Email Categorization with Caching

A categorization system that resolves templates once, caches them, and detects tampered cache entries:

import asyncio
from cleveractors.registry import (
    RegistryClient,
    RegistryCache,
    CacheStats,
    PackageId,
)


async def categorize_emails() -> None:
    client = RegistryClient(
        base_url="https://registry.example.com",
        api_key="categorizer-key",
    )
    cache = RegistryCache(client, max_size=256, ttl=600.0)

    CATEGORY_TEMPLATES = [
        "pkg_skl_aaa1111111111111111111111111111111111",
        "pkg_skl_bbb2222222222222222222222222222222222",
        "pkg_skl_ccc3333333333333333333333333333333333",
    ]

    templates: dict[str, dict] = {}

    for pkg_id in CATEGORY_TEMPLATES:
        try:
            content = await cache.get_package(pkg_id)
            templates[pkg_id] = content
            print(f"Loaded template: {content.get('name', pkg_id)}")
        except Exception as exc:
            print(f"Failed to load {pkg_id}: {exc}")

    # Warm the cache explicitly via put (content from resolve)
    resolved = await cache.resolve_package(
        package_type="skill",
        namespace="acme",
        name="priority-classifier",
        version="v2.0.0",
    )
    await cache.put(resolved["package_id"], {"name": "priority-classifier"})

    # Check cache statistics
    print(f"Cache stats: {cache.stats}")
    print(f"Template count: {len(templates)}")

    await cache.close()


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