# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/within-what-bounded-cache-and-object-lifecycle-drains-must-an-instance-constrain-in-memory-state-retention-to-prevent-reference-cycle-leaks-and-oom-kills.md
# Remnant Warrant:    https://corpusfetch.com/c/within-what-bounded-cache-and-object-lifecycle-drains-must-an-instance-constrain-in-memory-state-retention-to-prevent-reference-cycle-leaks-and-oom-kills.warrant.json
# Markdown SHA-256:   68d1aaf291fa38a14d33b34fc26baf440e893277934cab8211eb09d393b1674a
# Code AST SHA-256:   027fb10f7aeb208eb59ecc8a9e631664076f358e798ef807cd585d1da6e346f8
# Verified Classes:   BoundedTTLCache
# Defensive Guards:   1 fail-closed invariants (zero external deps)

import sys
import threading
import time
from collections import OrderedDict
from typing import Callable, Generic, Hashable, TypeVar

K = TypeVar("K", bound=Hashable)
V = TypeVar("V")


class BoundedTTLCache(Generic[K, V]):
    '''
    A thread-safe in-memory LRU cache bounded simultaneously by:
    1. Maximum entry count (`max_entries`)
    2. Maximum estimated payload byte size (`max_total_bytes`)
    3. Monotonic Time-To-Live expiration (`ttl_sec`)
    Prevents unbounded heap growth and OOM kills in long-running workers.
    '''

    def __init__(
        self,
        *,
        max_entries: int = 512,
        max_total_bytes: int = 32 * 1024 * 1024,  # 32 MiB hard ceiling
        ttl_sec: float = 300.0,
    ) -> None:
        if max_entries < 1 or max_total_bytes < 1024 or ttl_sec <= 0:
            raise ValueError("Invalid cache bounds")
        self.max_entries = max_entries
        self.max_total_bytes = max_total_bytes
        self.ttl_sec = ttl_sec
        self._lock = threading.Lock()
        # Maps key -> (value, expires_at_monotonic, estimated_bytes)
        self._store: OrderedDict[K, tuple[V, float, int]] = OrderedDict()
        self._current_bytes = 0

    @staticmethod
    def _estimate_bytes(obj: object) -> int:
        if isinstance(obj, (bytes, bytearray, memoryview)):
            return len(obj)
        if isinstance(obj, str):
            return len(obj.encode("utf-8", errors="ignore"))
        return sys.getsizeof(obj)

    def _evict_expired_and_overflow_locked(self, now: float) -> None:
        # Step 1: Purge expired keys from oldest to newest
        expired_keys = [
            k for k, (_, exp, _) in self._store.items() if exp <= now
        ]
        for k in expired_keys:
            _, _, sz = self._store.pop(k)
            self._current_bytes -= sz

        # Step 2: Evict least-recently-used entries until under entry & byte ceilings
        while (
            len(self._store) > self.max_entries
            or self._current_bytes > self.max_total_bytes
        ) and self._store:
            _, (_, _, sz) = self._store.popitem(last=False)
            self._current_bytes -= sz

    def get(self, key: K) -> V | None:
        now = time.monotonic()
        with self._lock:
            entry = self._store.get(key)
            if entry is None:
                return None
            val, expires_at, sz = entry
            if expires_at <= now:
                self._store.pop(key, None)
                self._current_bytes -= sz
                return None
            # Promote to most-recently-used
            self._store.move_to_end(key, last=True)
            return val

    def put(self, key: K, value: V) -> bool:
        val_bytes = self._estimate_bytes(key) + self._estimate_bytes(value)
        # Gate: Never cache a single giant object larger than 25% of the entire cache budget
        if val_bytes > (self.max_total_bytes // 4):
            return False

        now = time.monotonic()
        with self._lock:
            if key in self._store:
                _, _, old_sz = self._store.pop(key)
                self._current_bytes -= old_sz

            self._store[key] = (value, now + self.ttl_sec, val_bytes)
            self._current_bytes += val_bytes
            self._evict_expired_and_overflow_locked(now)
            return True


def strip_exception_frames(exc: BaseException) -> str:
    '''
    Extracts a clean diagnostic string from an exception and explicitly clears
    its __traceback__ chain so local stack-frame buffers are immediately freed.
    '''
    summary = f"{type(exc).__name__}: {exc}"
    exc.__traceback__ = None
    return summary