lambdagent/agentpaas/engine/cache.py

"""
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"}