cache.cache_manager

Overview

CacheManager is the single entry point for the entire semantic-cache pipeline, wiring together every other module documented in this reference: normalization, security, embeddings, storage, cost modeling, policy learning, tenant management, and metrics. Its module docstring includes an architecture diagram and a description of the lookup and store pipelines, both of which are reproduced (and cross-checked against the actual code) in Architecture.

_InflightCoalescer (private)

class _InflightCoalescer:
    def __init__(self) -> None
    async def coalesce(self, key: str, coro_fn: Callable) -> Any

Purpose

Deduplicates concurrent, identical embedding requests: if two callers request an embedding for the same normalized query at the same time, only one actually calls the (potentially expensive/rate-limited) embedding provider; the second waits for and reuses the first's result.

Design rationale

The class docstring explains the choice explicitly: it uses threading.Lock/ threading.Event rather than asyncio.Lock/asyncio.Event, specifically so it works correctly even if different callers are running their own separate event loops in different threads (e.g. sync callers each spinning up a loop in a thread pool via CacheManager.get_sync).

Algorithm

1. Under self._lock: check if key is already in-flight. If so, join as a follower (is_leader = False) and grab the existing (event, holder) pair. If not, become the leader: create a fresh threading.Event and an empty holder list, register them, and set is_leader = True.
2. Leader: await coro_fn(). On success, append (result, None) to holder; on exception, append (None, exc). In a finally block, set the event and remove the in-flight registration (so the next caller for this key starts fresh). Then re-raise if an exception was captured, or return the result.
3. Follower: await loop.run_in_executor(None, event.wait) — waits for the leader's event on a thread-pool thread so the event loop itself isn't blocked — then inspects the same holder list and returns/raises identically to the leader.

Complexity

O(1) beyond the cost of coro_fn() itself for the leader; followers pay one thread-pool round-trip to wait.

Thread Safety / Async Safety

Explicitly designed to be safe across multiple event loops in multiple threads, per its own docstring — this is one of the more carefully-reasoned concurrency primitives in the codebase.

_L1ExactCache (private)

class _L1ExactCache:
    def __init__(self, maxsize: int) -> None

An OrderedDict-backed LRU cache of IntelligentCacheEntry objects, keyed by the exact-match hash. This is CacheManager's only in-process cache tier (see the note on config_cache.md about l2_max_items not corresponding to a second CacheManager tier).

Thread-safe via its own RLock.

CacheManager

class CacheManager:
    def __init__(self, config: CacheConfig) -> None

Constructor

Builds, in order: QueryNormalizer, SecurityGuard, CachedEmbedder, storage (via build_storage), CachePolicyLearner, CostModel, DecisionEngine, TenantManager, CacheMetricsCollector, _InflightCoalescer, the _L1ExactCache, and an EmbeddingIndex sized from the embedder's dimension and config.semantic.max_index_size. Sets self._sim_threshold = config.semantic.similarity_threshold as the mutable, adaptive starting point. Finally starts the background eviction timer (_start_eviction_timer()) and logs an INFO summary line.

Note: the synchronous constructor does not warm up the embedding provider — use create() for that.

create(cls, config: Optional[CacheConfig] = None) -> CacheManager (async classmethod)

manager = await CacheManager.create()             # defaults
manager = await CacheManager.create(my_config)    # explicit config

Constructs a CacheManager(config or CacheConfig()), then attempts one warm-up call (await mgr._embedder.embed_single("warmup")), catching and logging (not raising) any failure. This is the recommended construction path for async applications.

async get(self, query: str, tenant_id: str = "default", top_k: int = 1) -> CacheLookupResult

Overview

The full cache-lookup-with-routing operation. See Architecture → get() lookup flow for the complete flowchart; this section is the textual, step-by-step equivalent, verified line-by-line against the source.

Why it exists: to answer "do we already have a good-enough cached response for this query" while enforcing security, quota, and tenant-isolation policy along the way, and while recording every relevant metric.

When to use it: as the first step of handling any incoming query, before invoking an LLM.

When not to use it: if you need a raw, policy-free key/value get — use the storage backend directly, or use MultiLevelCache for that simpler use case.

Parameters

Returns

A CacheLookupResult. hit=True only for EXACT_HIT or SEMANTIC_HIT decisions; every other outcome (security rejection, quota rejection, no candidates, below-threshold similarity, or DecisionEngine choosing the LLM) returns hit=False with decision=RoutingDecision.LLM_FALLBACK.

Raises

get() does not propagate SecurityViolation or QuotaExceeded to the caller — both are caught internally, logged, and translated into a miss result. No exception type is documented as escaping this method under normal operation in the analyzed source (unhandled exceptions from deeper layers, e.g. a storage backend bug, would still propagate, but there is no explicit except Exception catch-all around the whole method body).

Internal Workflow (verified against source, in order)

1. Record the request in metrics (_metrics.record_request).
2. SecurityGuard.validate_query — on SecurityViolation, record a "security_violation" error and return a miss.
3. TenantManager.check_and_charge_request — on QuotaExceeded, record a "quota_exceeded" error and return a miss.
4. Normalize the query and compute the exact-match key.
5. L1 exact check: if present, not expired, and integrity-verified (SecurityGuard.verify_entry_integrity), record an exact hit and return immediately with similarity=1.0. If present but integrity fails, invalidate the L1 entry, record an "integrity_fail_l1" error, and continue to the next step (does not return early).
6. Record an exact miss (_metrics.record_exact_miss()) — this happens regardless of whether step 5 found nothing or found an integrity failure.
7. Storage exact check: if present, not expired, integrity-verified, and tenant-accessible (TenantManager.is_accessible), promote to L1, touch() it, record an exact hit, and return.
8. Semantic search: embed the normalized query (via the _InflightCoalescer, keyed on the normalized query text, so concurrent identical queries share one embedding call), record embedding latency, then search the vector index with min_similarity = self._sim_threshold * 0.80 (a looser pre-filter than the "real" threshold, giving the ranking/decision stages a wider candidate pool to work with).
9. For each raw candidate, load the full entry from storage, skipping it if missing, expired, not tenant-accessible, or failing integrity verification (the latter also records an "integrity_fail_semantic" error).
10. If any candidates survive: re-rank via CachePolicyLearner.rank_candidates, take the top result, and pass it to DecisionEngine.decide().
11. If the decision is EXACT_HIT or SEMANTIC_HIT: touch() the entry, promote it to L1, record a semantic hit, and return hit=True with decision=SEMANTIC_HIT (note: even if DecisionEngine.decide() internally returned RoutingDecision.EXACT_HIT for a near-identical semantic match, CacheManager.get()'s own return statement here hard-codes decision=RoutingDecision.SEMANTIC_HIT — see Known Issues for this discrepancy).
12. Otherwise (no candidates, or the decision engine chose the LLM path): record a semantic miss and an LLM fallback (with its estimated cost), and return hit=False.

Side Effects

• Mutates the L1 cache (inserts/invalidates).
• May mutate entry.usage_count/last_access via touch().
• Extensive metrics recording (see CacheMetricsCollector).
• Logging: warnings on security/quota rejection.

Thread Safety / Async Safety

Safe to call concurrently — every sub-component it touches (_l1, _metrics, _vector_idx, _tenant_mgr) has its own internal locking, and the _InflightCoalescer specifically protects the one operation (embedding) that would otherwise be duplicated under concurrent identical queries.

Examples

# 1. Basic
result = await manager.get("What is RAG?")

# 2. Intermediate — multi-tenant
result = await manager.get("What is RAG?", tenant_id="acme-corp")

# 3. Advanced — wider candidate pool for the ranker
result = await manager.get("What is RAG?", tenant_id="acme-corp", top_k=5)

# 4. Error handling — get() itself won't raise for security/quota issues;
#    treat any miss uniformly:
result = await manager.get(suspicious_input, tenant_id="acme-corp")
if not result.hit:
    # Could be a genuine miss, a rejected query, or an exhausted quota —
    # get() intentionally does not distinguish these in its return value.
    answer = await call_llm(suspicious_input)

# 5. Production-ready usage
result = await manager.get(query, tenant_id=tenant_id)
if result.hit:
    response = result.response
else:
    response = await call_llm(query)
    await manager.put(query, response, tenant_id=tenant_id, response_tokens=len(response.split()))

Best Practices

• Always branch on result.hit, not on result.decision, to decide whether to call an LLM — decision is informative/diagnostic, hit is the actionable boolean.
• Pass a real tenant_id even in single-tenant deployments if you plan to add tenants later — the "default" tenant has no special exemption from quotas.

Common Mistakes

• Assuming a non-hit means "definitely not in the cache" — it can also mean the query/ response was rejected by SecurityGuard, the tenant's quota was exhausted, or the decision engine judged the LLM path better even though a similar entry existed.
• Treating result.decision == RoutingDecision.SEMANTIC_HIT as proof the entry was found via approximate similarity rather than an exact string match — per the note in step 11 above, get()'s semantic branch always reports SEMANTIC_HIT, even for a similarity of 1.0.

async put(self, query: str, response: Any, tenant_id: str = "default", ttl_s: Optional[float] = None, quality_score: float = 1.0, response_tokens: int = 0, is_shared: bool = False, input_tokens: int = 0) -> Optional[IntelligentCacheEntry]

Overview

Stores a query/response pair. See Architecture → put() store flow for the flowchart.

Parameters

Returns

The newly-created IntelligentCacheEntry, or None if rejected by security or quota checks.

Internal Workflow

1. SecurityGuard.validate_query + validate_response — on violation, record "security_violation_put" and return None.
2. SecurityGuard.scrub_pii on the query, and on the response only if it's a str (non-string responses pass through scrub_pii unchanged, since the scrubber operates on text).
3. TenantManager.check_entry_quota + check_memory_quota — on violation, record "quota_exceeded_put" and return None.
4. Normalize the (cleaned) query and compute the exact key.
5. Embed the normalized query (directly via CachedEmbedder.embed_single — not routed through the _InflightCoalescer the way get()'s embedding call is).
6. Estimate embedding and LLM costs via CostModel, building a CostRecord.
7. Build the IntelligentCacheEntry, using the exact key as entry_id.
8. Persist to L1, storage, and the vector index (add).
9. Charge the tenant's entry-count and memory quotas.
10. Log at DEBUG and return the entry.

Side Effects

Writes to _l1, _storage, _vector_idx; charges tenant quota counters.

Examples

# Basic
await manager.put("What is RAG?", "Retrieval-Augmented Generation is...")

# With explicit TTL and token accounting
await manager.put(
    "What is RAG?", answer,
    ttl_s=1800, response_tokens=120, input_tokens=8,
)

# Opting into cross-tenant sharing (requires the tenant to have allow_shared_write=True)
await manager.put("common FAQ answer", answer, tenant_id="acme", is_shared=True)

Common Mistakes

• Passing ttl_s=0 expecting an immediate/no-op cache write — it is falsy, so it falls back to config.default_ttl_s instead of expiring immediately (same pattern as the storage backends' set() methods).
• Expecting is_shared=True to always take effect — it's silently downgraded to False if the tenant isn't registered with allow_shared_write=True.

async feedback(self, entry_id: str, positive: bool, magnitude: float = 1.0, tenant_id: str = "default", source: str = "user") -> None

Overview

Records a quality signal for a previously-served entry, updating its bandit arm and the pipeline's adaptive similarity threshold.

Internal Workflow

1. Look up the entry by entry_id — tries storage first, then L1, as a fallback. Logs a warning and returns early if not found in either.
2. Build a FeedbackSignal and pass it to CachePolicyLearner.record_feedback.
3. Record the entry's (now-updated) bandit_arm.expected_reward into metrics (record_reward).
4. Call DecisionEngine.update_threshold(positive).
5. If config.semantic.adaptive_threshold is enabled, nudge self._sim_threshold itself (separately from DecisionEngine's own internal bonus) — relaxing it on positive feedback (down to threshold_min) or tightening it on negative feedback (up to threshold_max), by half of threshold_step on the positive side and a full threshold_step on the negative side (an intentional asymmetry favoring caution).
6. Persist the updated entry back to storage (self._storage.set(entry_id, entry, ttl_s=entry.ttl_s)).

[!NOTE] Two independent threshold-adjustment mechanisms exist CacheManager._sim_threshold (adjusted directly by feedback(), used as the baseline for get()'s pre-filter) and DecisionEngine._current_threshold_bonus (adjusted by update_threshold(), added on top of whatever threshold is passed into decide()) are two separate, independently-moving numbers that both respond to the same feedback event but at different asymmetric rates. Their combined effect is what actually determines the effective threshold inside DecisionEngine.decide() — see Known Issues for a closer look at how these interact.

Examples

await manager.feedback(entry.entry_id, positive=True, tenant_id="acme")
await manager.feedback(entry.entry_id, positive=False, magnitude=0.5, source="llm_eval")

Tenant management delegates

[!WARNING] flush_tenant also deregisters the tenant, inside its own L1-scanning loop Reading _flush_tenant's body closely: for every L1 key belonging to tenant_id, it calls self._tenant_mgr.deregister(tenant_id) — inside the loop over matching keys, meaning it is called once per matching L1 entry, not once overall (harmless since deregister is idempotent for an already-removed tenant, but redundant). More significantly, this means flush_tenant() always deregisters the tenant (as long as at least one matching entry existed in L1), which is a stronger operation than its name ("flush", i.e. clear the data) suggests — callers expecting to flush a tenant's cache and continue using it under the same tenant configuration should be aware that its registration (quotas, shared-cache flags) may be removed as a side effect. See Known Issues.

Background eviction

Metrics & monitoring

Synchronous wrappers

_run_sync(coro) (static, private)

Overview: the mechanism behind every *_sync method — runs a coroutine to completion regardless of whether the calling thread already has a running event loop.

Algorithm:

• If a loop is already running in the current thread (e.g. inside Jupyter or a framework like FastAPI that manages its own loop): submit the coroutine to that loop via asyncio.run_coroutine_threadsafe(coro, running_loop) and block on the resulting concurrent.futures.Future.
• Otherwise: use asyncio.run(coro), which creates an isolated event loop, runs the coroutine to completion, and closes the loop — appropriate for plain scripts or thread-pool worker threads that don't already own a loop.

get_sync, put_sync, feedback_sync

Thin wrappers calling _run_sync(self.get(...)) / self.put(...) / self.feedback(...) with the same parameters (positionally forwarded, plus **kwargs for put_sync).

Lifecycle

[!NOTE] No __aenter__/__aexit__ Unlike MultiLevelCache, CacheManager does not implement __aenter__/__aexit__ in the analyzed source, despite being an otherwise async-first class and having an aclose() method. Use the synchronous with manager: form, or call await manager.aclose() explicitly.

Examples: complete lifecycle

import asyncio
from zerda.cache import CacheManager, CacheConfig

async def main():
    manager = await CacheManager.create(CacheConfig())
    try:
        result = await manager.get("hello")
        if not result.hit:
            await manager.put("hello", "hi!")
    finally:
        manager.close()

asyncio.run(main())

# Synchronous context manager form
from zerda.cache import CacheManager, CacheConfig

with CacheManager(CacheConfig()) as manager:
    result = manager.get_sync("hello")
    if not result.hit:
        manager.put_sync("hello", "hi!")

Related APIs

• Architecture — full pipeline diagrams
• CacheConfig — the sole constructor input
• Every module under API Overview — CacheManager composes all of them