core/cache.py

"""Thread-safe metadata cache used by orchestration components."""

from __future__ import annotations

from collections import OrderedDict
from threading import RLock
from typing import Generic, TypeVar


K = TypeVar("K")
V = TypeVar("V")


class LRUCache(Generic[K, V]):
    """A minimal bounded least-recently-used cache."""

    def __init__(self, capacity: int) -> None:
        if capacity < 1:
            raise ValueError("capacity must be positive")
        self.capacity = capacity
        self._items: OrderedDict[K, V] = OrderedDict()
        self._lock = RLock()

    def get(self, key: K) -> V | None:
        with self._lock:
            value = self._items.get(key)
            if value is not None:
                self._items.move_to_end(key)
            return value

    def put(self, key: K, value: V) -> tuple[K, V] | None:
        with self._lock:
            self._items[key] = value
            self._items.move_to_end(key)
            if len(self._items) > self.capacity:
                return self._items.popitem(last=False)
            return None

    def clear(self) -> list[V]:
        with self._lock:
            values = list(self._items.values())
            self._items.clear()
            return values

    def values(self) -> list[V]:
        """Return a snapshot of cached values without exposing mutable state."""
        with self._lock:
            return list(self._items.values())