File size: 3,146 Bytes
377417d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | """Response cache — SQLite inside the container, ephemeral, TTL per source (ADR-0001 §8).
ToolUniverse's own result cache is switched off (``src.tu_client``) so this is the only one, and
it keeps the one fact a cache usually hides: ``retrieved_at`` is when the SOURCE answered, and a
hit returns that original time, never the time the cache served it. Only responses that passed
their schema check are stored, so a drifted response is never replayed.
The file lives in the temp dir by default (``KNOWLEDGE_CACHE_PATH`` overrides; ``:memory:`` for
tests). Nothing here is persistent on a Space, by design: a rebuild empties it.
"""
from __future__ import annotations
import hashlib
import json
import os
import sqlite3
import tempfile
import threading
import time
from collections.abc import Callable
from pathlib import Path
from typing import Any
from src import config
DEFAULT_PATH = str(Path(tempfile.gettempdir()) / "pdac-knowledge-cache.sqlite3")
def cache_key(tool: str, args: dict[str, Any]) -> str:
canonical = json.dumps({"tool": tool, "args": args}, sort_keys=True, separators=(",", ":"))
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
class ResponseCache:
def __init__(self, path: str | None = None, *, clock: Callable[[], float] = time.time):
self.path = path or os.environ.get("KNOWLEDGE_CACHE_PATH") or DEFAULT_PATH
self._clock = clock
self._lock = threading.Lock()
self._db = sqlite3.connect(self.path, check_same_thread=False)
self._db.execute(
"CREATE TABLE IF NOT EXISTS responses ("
" key TEXT PRIMARY KEY, source TEXT NOT NULL, tool TEXT NOT NULL,"
" data TEXT NOT NULL, retrieved_at REAL NOT NULL)"
)
self._db.commit()
def get(self, source: str, tool: str, args: dict[str, Any]) -> tuple[Any, float] | None:
"""``(data, retrieved_at)`` for a live entry, else None. Expired entries are dropped."""
key = cache_key(tool, args)
ttl = config.source(source)["ttl_seconds"]
with self._lock:
row = self._db.execute(
"SELECT data, retrieved_at FROM responses WHERE key = ?", (key,)
).fetchone()
if row is None:
return None
data, retrieved_at = row
if self._clock() - retrieved_at >= ttl:
self._db.execute("DELETE FROM responses WHERE key = ?", (key,))
self._db.commit()
return None
return json.loads(data), retrieved_at
def put(
self, source: str, tool: str, args: dict[str, Any], data: Any, retrieved_at: float
) -> None:
key = cache_key(tool, args)
with self._lock:
self._db.execute(
"INSERT OR REPLACE INTO responses (key, source, tool, data, retrieved_at)"
" VALUES (?, ?, ?, ?, ?)",
(key, source, tool, json.dumps(data), retrieved_at),
)
self._db.commit()
def clear(self) -> None:
with self._lock:
self._db.execute("DELETE FROM responses")
self._db.commit()
|