Spaces:
Running
Running
| from __future__ import annotations | |
| import hashlib | |
| import os | |
| import threading | |
| import time | |
| from collections import OrderedDict | |
| from pathlib import Path | |
| from typing import Any | |
| from oplog import atomic_write_json, log_event | |
| # Per-space prompt cache - the panel's top free-tier utility lever. Each user runs their | |
| # OWN single-tenant space, so caching their own (prompt, budget) -> output is safe and | |
| # saves a real provider call on an exact repeat. In-memory LRU first, with an optional | |
| # disk mirror (atomic writes) so the cache survives a restart/redeploy alongside jobs. | |
| # | |
| # Correctness rules: the key includes every output-affecting field; we ONLY cache | |
| # successful calls; and a job marked no_store is never written. Never raises. | |
| CACHE_DIR = Path(os.environ.get("CACHE_DIR", Path(__file__).resolve().parent / "data" / "cache")) | |
| CACHE_ENABLED = os.environ.get("CACHE_ENABLED", "1").strip() not in ("0", "false", "no", "") | |
| CACHE_TTL_S = float(os.environ.get("CACHE_TTL_S", str(24 * 3600))) | |
| CACHE_MAX = int(os.environ.get("CACHE_MAX", "512")) | |
| class PromptCache: | |
| def __init__(self, *, enabled: bool = CACHE_ENABLED, dirpath: Path = CACHE_DIR, | |
| ttl_s: float = CACHE_TTL_S, max_entries: int = CACHE_MAX) -> None: | |
| self.enabled = enabled | |
| self.ttl_s = ttl_s | |
| self.max_entries = max(1, max_entries) | |
| self.dir = Path(dirpath) | |
| self._mem: "OrderedDict[str, dict]" = OrderedDict() | |
| self._lock = threading.Lock() | |
| self.hits = 0 | |
| self.misses = 0 | |
| if self.enabled: | |
| try: | |
| self.dir.mkdir(parents=True, exist_ok=True) | |
| except OSError as e: | |
| log_event("cache_dir_error", error=repr(e)[:200]) | |
| self.enabled = False | |
| def key(*, logical: str, system_prompt: str, user_msg: str, max_tokens: int, | |
| reasoning: bool, research: bool, privacy: str = "off") -> str: | |
| # privacy is part of the key: a result produced under privacy=off may have | |
| # come from a training/logging host, and must never be served to a strict | |
| # request (panel-found bypass, whole-repo review 2026-06). | |
| h = hashlib.sha256() | |
| for part in (logical, system_prompt, user_msg, str(int(max_tokens)), | |
| "1" if reasoning else "0", "1" if research else "0", privacy): | |
| h.update(part.encode("utf-8", "replace")) | |
| h.update(b"\x00") | |
| return h.hexdigest() | |
| def _path(self, k: str) -> Path: | |
| return self.dir / f"{k}.json" | |
| def _fresh(self, entry: dict) -> bool: | |
| return (time.time() - entry.get("ts", 0)) <= self.ttl_s | |
| def get(self, k: str) -> dict | None: | |
| if not self.enabled: | |
| return None | |
| with self._lock: | |
| entry = self._mem.get(k) | |
| if entry is not None: | |
| if self._fresh(entry): | |
| self._mem.move_to_end(k) | |
| self.hits += 1 | |
| return dict(entry["value"]) | |
| self._mem.pop(k, None) | |
| # cold mem (restart) -> try the disk mirror. | |
| try: | |
| p = self._path(k) | |
| if p.exists(): | |
| import json | |
| entry = json.loads(p.read_text(encoding="utf-8")) | |
| if self._fresh(entry): | |
| with self._lock: | |
| self._mem[k] = entry | |
| self._mem.move_to_end(k) | |
| self.hits += 1 | |
| return dict(entry["value"]) | |
| p.unlink(missing_ok=True) | |
| except (OSError, ValueError): | |
| pass | |
| with self._lock: | |
| self.misses += 1 | |
| return None | |
| def put(self, k: str, value: dict) -> None: | |
| if not self.enabled: | |
| return | |
| entry = {"ts": time.time(), "value": value} | |
| with self._lock: | |
| self._mem[k] = entry | |
| self._mem.move_to_end(k) | |
| while len(self._mem) > self.max_entries: | |
| old_k, _ = self._mem.popitem(last=False) | |
| try: | |
| self._path(old_k).unlink(missing_ok=True) | |
| except OSError: | |
| pass | |
| try: | |
| atomic_write_json(self._path(k), entry) | |
| except Exception as e: # noqa: BLE001 - cache write must never break a call | |
| log_event("cache_write_error", error=repr(e)[:200]) | |
| def stats(self) -> dict: | |
| with self._lock: | |
| total = self.hits + self.misses | |
| return {"enabled": self.enabled, "entries": len(self._mem), | |
| "hits": self.hits, "misses": self.misses, | |
| "hit_rate": round(self.hits / total, 3) if total else 0.0, | |
| "ttl_s": self.ttl_s, "max_entries": self.max_entries} | |