RAG Answer Cache: เจอคำตอบเก่า ยังต้องตรวจสิทธิ์ทุกครั้ง

"""A small, process-local RAG answer-cache authorization gate (Python 3.12+).

This module is an educational mock. It does not retrieve documents, authenticate
users, or establish that a source list is complete or semantically correct.
"""

from __future__ import annotations

from dataclasses import dataclass
import hashlib
import json
import math
from typing import Mapping

MAX_SCOPE_CHARS = 256
MAX_QUERY_BYTES = 16 * 1024
MAX_ANSWER_CHARS = 64 * 1024
MAX_SOURCE_IDS = 128


class CacheInputError(ValueError):
    """Raised when the bounded mock input contract is violated."""


def _finite_number(name: str, value: object) -> float:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise CacheInputError(f"{name} must be a finite number")
    try:
        result = float(value)
    except OverflowError as exc:
        raise CacheInputError(f"{name} must be a finite number") from exc
    if not math.isfinite(result):
        raise CacheInputError(f"{name} must be a finite number")
    return result


@dataclass(frozen=True)
class TrustedScope:
    """Values supplied by a trusted server-side auth/policy layer."""

    tenant_id: str
    principal_id: str
    policy_revision: str
    corpus_revision: str
    retrieval_profile: str


@dataclass(frozen=True)
class CacheEntry:
    """Cached answer plus the source IDs the producing pipeline declared."""

    scope: TrustedScope
    query_digest: str
    answer: str
    declared_source_ids: tuple[str, ...]
    expires_at: float


def _bounded_text(name: str, value: object, limit: int) -> str:
    if not isinstance(value, str) or not value or len(value) > limit:
        raise CacheInputError(f"{name} must be non-empty text of at most {limit} chars")
    if "\x00" in value:
        raise CacheInputError(f"{name} contains a forbidden NUL character")
    return value


def _validate_scope(scope: TrustedScope) -> None:
    if not isinstance(scope, TrustedScope):
        raise CacheInputError("trusted scope is required")
    for name in ("tenant_id", "principal_id", "policy_revision", "corpus_revision", "retrieval_profile"):
        value = _bounded_text(name, getattr(scope, name), MAX_SCOPE_CHARS)
        try:
            value.encode("utf-8", errors="strict")
        except UnicodeEncodeError as exc:
            raise CacheInputError(f"{name} must be valid UTF-8 text") from exc


def query_digest(query: bytes) -> str:
    """Hash exact valid UTF-8 query bytes; intentionally performs no normalization."""
    if not isinstance(query, bytes) or not query or len(query) > MAX_QUERY_BYTES:
        raise CacheInputError(f"query must be 1..{MAX_QUERY_BYTES} exact bytes")
    try:
        query.decode("utf-8", errors="strict")
    except UnicodeDecodeError as exc:
        raise CacheInputError("query must contain valid UTF-8 bytes") from exc
    return hashlib.sha256(query).hexdigest()


def cache_key(scope: TrustedScope, query: bytes) -> str:
    """Return a versioned digest over canonical JSON; digest is not a secret."""
    _validate_scope(scope)
    payload = {
        "v": 1,
        "tenant_id": scope.tenant_id,
        "principal_id": scope.principal_id,
        "policy_revision": scope.policy_revision,
        "corpus_revision": scope.corpus_revision,
        "retrieval_profile": scope.retrieval_profile,
        "query_sha256": query_digest(query),
    }
    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def _valid_source_ids(source_ids: object) -> tuple[str, ...]:
    if not isinstance(source_ids, tuple) or not 1 <= len(source_ids) <= MAX_SOURCE_IDS:
        raise CacheInputError(f"declared source IDs must be a tuple of 1..{MAX_SOURCE_IDS} IDs")
    if any(not isinstance(item, str) or not item or len(item) > MAX_SCOPE_CHARS for item in source_ids):
        raise CacheInputError("each declared source ID must be non-empty bounded text")
    if len(set(source_ids)) != len(source_ids):
        raise CacheInputError("declared source IDs must be unique")
    return source_ids


def put(
    store: dict[str, CacheEntry],
    scope: TrustedScope,
    query: bytes,
    answer: str,
    declared_source_ids: tuple[str, ...],
    now: float,
    ttl_seconds: float,
) -> str:
    """Store one mock answer with a bounded, injected process-local TTL."""
    _validate_scope(scope)
    digest = query_digest(query)
    _bounded_text("answer", answer, MAX_ANSWER_CHARS)
    source_ids = _valid_source_ids(declared_source_ids)
    now = _finite_number("now", now)
    ttl_seconds = _finite_number("TTL", ttl_seconds)
    if ttl_seconds <= 0:
        raise CacheInputError("now and positive TTL must be finite numbers")
    expires_at = now + ttl_seconds
    if now < 0 or not math.isfinite(expires_at) or expires_at <= now:
        raise CacheInputError("invalid monotonic time or TTL overflow")
    key = cache_key(scope, query)
    store[key] = CacheEntry(scope, digest, answer, source_ids, expires_at)
    return key


def get_authorized(
    store: Mapping[str, CacheEntry],
    scope: TrustedScope | None,
    query: bytes,
    current_authorized_source_ids: frozenset[str] | None,
    now: float,
) -> str | None:
    """Return an answer only after a current fixture authorization check.

    ``None`` means miss or deny. An empty allowed-source set is valid and can
    authorize only entries with no declared sources.
    """
    if not isinstance(store, Mapping) or scope is None or current_authorized_source_ids is None:
        return None
    try:
        _validate_scope(scope)
        digest = query_digest(query)
    except CacheInputError:
        return None
    try:
        now = _finite_number("now", now)
    except CacheInputError:
        return None
    if now < 0:
        return None
    key = cache_key(scope, query)
    entry = store.get(key)
    if not isinstance(entry, CacheEntry):
        return None
    # Defense against a wrong/corrupt mapping entry; never trust the key alone.
    if entry.scope != scope or entry.query_digest != digest:
        return None
    try:
        expires_at = _finite_number("expiry", entry.expires_at)
    except CacheInputError:
        return None
    if now >= expires_at:
        return None
    if not isinstance(current_authorized_source_ids, frozenset):
        return None
    try:
        declared_ids = _valid_source_ids(entry.declared_source_ids)
    except CacheInputError:
        return None
    try:
        answer = _bounded_text("answer", entry.answer, MAX_ANSWER_CHARS)
    except CacheInputError:
        return None
    if not set(declared_ids).issubset(current_authorized_source_ids):
        return None
    return answer

"""Authored acceptance examples for cache_gate; intentionally not executed."""

import unittest
from dataclasses import replace

from cache_gate import CacheInputError, TrustedScope, cache_key, get_authorized, put


class CacheGateTests(unittest.TestCase):
    def setUp(self):
        self.scope = TrustedScope("tenant-a", "user-a", "policy-4", "corpus-9", "hybrid-v2")
        self.query = b"How do I rotate the service key?"
        self.store = {}
        put(self.store, self.scope, self.query, "Use the approved rotation runbook.", ("doc-17",), 100.0, 30.0)
        self.allowed = frozenset({"doc-17", "doc-18"})

    def test_same_scope_and_current_source_permission_returns_cached_answer(self):
        self.assertEqual(get_authorized(self.store, self.scope, self.query, self.allowed, 101.0),
                         "Use the approved rotation runbook.")

    def test_tenant_or_principal_change_is_a_miss(self):
        for field, value in (("tenant_id", "tenant-b"), ("principal_id", "user-b")):
            with self.subTest(field=field):
                other = replace(self.scope, **{field: value})
                self.assertNotEqual(cache_key(self.scope, self.query), cache_key(other, self.query))
                self.assertIsNone(get_authorized(self.store, other, self.query, self.allowed, 101.0))

    def test_policy_corpus_or_retrieval_revision_change_is_a_miss(self):
        for field, value in (("policy_revision", "policy-5"), ("corpus_revision", "corpus-10"),
                             ("retrieval_profile", "hybrid-v3")):
            with self.subTest(field=field):
                other = replace(self.scope, **{field: value})
                self.assertIsNone(get_authorized(self.store, other, self.query, self.allowed, 101.0))

    def test_exact_query_bytes_change_is_a_miss(self):
        self.assertIsNone(get_authorized(self.store, self.scope, self.query + b" ", self.allowed, 101.0))

    def test_current_reauthorization_denies_removed_source(self):
        self.assertIsNone(get_authorized(self.store, self.scope, self.query, frozenset({"doc-18"}), 101.0))

    def test_empty_allowed_set_denies_entry_with_sources(self):
        self.assertIsNone(get_authorized(self.store, self.scope, self.query, frozenset(), 101.0))

    def test_expired_entry_is_a_miss_at_expiry_boundary(self):
        self.assertIsNone(get_authorized(self.store, self.scope, self.query, self.allowed, 130.0))

    def test_inconsistent_entry_scope_is_rejected_even_under_expected_key(self):
        key = cache_key(self.scope, self.query)
        entry = self.store[key]
        self.store[key] = replace(entry, scope=replace(self.scope, tenant_id="tenant-b"))
        self.assertIsNone(get_authorized(self.store, self.scope, self.query, self.allowed, 101.0))

    def test_malformed_cached_source_ids_are_rejected(self):
        key = cache_key(self.scope, self.query)
        self.store[key] = replace(self.store[key], declared_source_ids=["doc-17"])
        self.assertIsNone(get_authorized(self.store, self.scope, self.query, self.allowed, 101.0))

    def test_empty_provenance_is_rejected_when_storing_and_reading(self):
        with self.assertRaises(CacheInputError):
            put(self.store, self.scope, b"query two", "answer", (), 100.0, 10.0)
        key = cache_key(self.scope, self.query)
        self.store[key] = replace(self.store[key], declared_source_ids=())
        self.assertIsNone(get_authorized(self.store, self.scope, self.query, frozenset(), 101.0))

    def test_corrupt_cached_answer_is_rejected(self):
        key = cache_key(self.scope, self.query)
        self.store[key] = replace(self.store[key], answer="")
        self.assertIsNone(get_authorized(self.store, self.scope, self.query, self.allowed, 101.0))

    def test_invalid_utf8_query_is_rejected(self):
        with self.assertRaises(CacheInputError):
            cache_key(self.scope, b"\xff")

    def test_ttl_overflow_is_rejected(self):
        with self.assertRaises(CacheInputError):
            put(self.store, self.scope, b"query two", "answer", ("doc-17",), 1.0e308, 1.0e308)

    def test_missing_authorization_context_denies(self):
        self.assertIsNone(get_authorized(self.store, self.scope, self.query, None, 101.0))
        self.assertIsNone(get_authorized(self.store, None, self.query, self.allowed, 101.0))


if __name__ == "__main__":
    unittest.main()

# RAG cache gate mock: usage and boundaries

This teaching example demonstrates one rule: a cache key scopes a possible reuse, while current source authorization gates the answer on every hit. It uses Python 3.12+ and the standard library only.

## Runtime and setup

- **Runtime:** Python 3.12 or newer.
- **Dependencies:** none beyond the standard library.
- **Environment variables / credentials:** none.
- **Installation:** none; use an existing Python 3.12 installation. Do not install packages for this example.
- **Network, API, model, vector store, cache service, or GPU:** none.
- **Permissions:** local read-only computation and process memory only; no file, database, or network writes.
- **Cost:** no API or GPU cost. Local CPU and memory use only; no performance claim is made.

## Run the example

From the `code` directory, the following one-line command creates a trusted fixture, stores a mock answer, then checks it against the current allowed source IDs:

```sh
python -c 'from cache_gate import *; s=TrustedScope("tenant-a","user-a","policy-4","corpus-9","hybrid-v2"); q=b"How do I rotate the service key?"; d={}; put(d,s,q,"Use the approved rotation runbook.",("doc-17",),100.0,30.0); print(get_authorized(d,s,q,frozenset({"doc-17"}),101.0))'
```

Expected output:

```text
Use the approved rotation runbook.
```

Changing the allowed set to `frozenset()` should produce `None` because the entry declares `doc-17`. Changing tenant, principal, policy revision, corpus revision, retrieval profile, or exact query bytes should also produce a miss. These are expected behaviors authored into the acceptance examples; they have **not been executed**.

## Acceptance examples

From the `code` directory, the test command is:

```sh
python -m unittest -v test_cache_gate.py
```

Expected result when run in a supported environment: fourteen passing `unittest` cases covering same-scope access, tenant/principal/revision/query misses, current reauthorization, empty permissions, expiry, missing context, malformed entries, empty provenance, corrupt answers, invalid UTF-8, and TTL overflow. **Testing status: NOT RUN.** Do not interpret the authored expectations as runtime evidence.

## Important limits

The trusted identity and allowed source IDs are fixtures, not authentication or authorization integrations. The declared source list is assumed complete; semantic provenance is not checked. The dict and injected monotonic TTL are process-local, without distributed invalidation, persistence, locking, or a revocation-freshness guarantee. Hashes are not secrets or permission proofs. The mock does not establish that cached answers are correct or production safe.