""" engine.cache — Compile cache for YAML → Term. Compilation is deterministic (FROM_CONFIG_SPEC P4), so same hash → same Term. """ from __future__ import annotations from collections import OrderedDict from typing import Any, Optional class CompileCache: def __init__(self, maxsize: int = 1000): self._cache: OrderedDict = OrderedDict() self._maxsize = maxsize self.hits = 0 self.misses = 0 def get(self, config_hash: str) -> Optional[Any]: if config_hash in self._cache: self._cache.move_to_end(config_hash) self.hits += 1 return self._cache[config_hash] self.misses += 1 return None def put(self, config_hash: str, term: Any): self._cache[config_hash] = term self._cache.move_to_end(config_hash) while len(self._cache) > self._maxsize: self._cache.popitem(last=False) def get_or_compile(self, config_hash: str, config: dict) -> Any: cached = self.get(config_hash) if cached: return cached import tempfile, yaml, os with tempfile.NamedTemporaryFile(mode='w', suffix='.yml', delete=False, encoding='utf-8') as f: yaml.dump(config, f, allow_unicode=True) tmp = f.name try: from lambdagent.fromconfig import from_config term = from_config(tmp) self.put(config_hash, term) return term finally: os.unlink(tmp) @property def stats(self): total = self.hits + self.misses return {"size": len(self._cache), "hits": self.hits, "misses": self.misses, "hit_rate": f"{self.hits/total:.1%}" if total else "N/A"}