deploy: phase 3 BYOK backend (Dockerfile.hf, FastAPI on 7860)

config/settings.py

@@ -173,6 +173,13 @@ class Settings(BaseSettings):
     # deps — so it is safe to leave on everywhere.
     audit_verify_enabled: bool = True
     audit_verify_interval_hours: int = 6
+    # Optional HMAC key for the audit hash chain. When unset (default) entries
+    # are SHA-256 hashed — tamper-*evident* (any edit breaks the chain, but an
+    # attacker with file access can recompute the whole chain). When set, each
+    # entry hash is an HMAC-SHA256 keyed by this secret, making the chain
+    # tamper-*resistant* (an attacker cannot forge a valid chain without the
+    # key). Keep the key out of the audit host's filesystem (env/secret store).
+    audit_hmac_key: str | None = None
 
     # ── Citation Faithfulness Gate (NLI) ─────────────────────────────────────────
     # After synthesis, run a per-sentence NLI check: for each sentence that

ingestion/contextual.py

@@ -41,6 +41,7 @@ async def _generate_one(
     semaphore: asyncio.Semaphore,
     prefer_cloud: bool,
     max_doc_chars: int,
+    sensitivity_level: str = "low",
 ) -> str:
     """Generate a single chunk's context summary.
 
@@ -63,7 +64,9 @@ async def _generate_one(
             ctx = await call_llm_async(
                 prompt,
                 system_prompt="You generate short retrieval context summaries.",
-                sensitivity_level="low",
+                # Use the document's real sensitivity so HIGH content is summarised
+                # on local inference and never sent to a cloud provider at ingest.
+                sensitivity_level=sensitivity_level,
                 prefer_cloud=prefer_cloud,
             )
             return ctx.strip()
@@ -79,6 +82,7 @@ async def generate_chunk_contexts(
     prefer_cloud: bool = False,
     max_concurrent: int = 8,
     max_doc_chars: int = 50_000,
+    sensitivity_level: str = "low",
 ) -> list[str]:
     """Generate contexts for every chunk concurrently.
 
@@ -96,7 +100,10 @@ async def generate_chunk_contexts(
     if not chunks:
         return []
     sem = asyncio.Semaphore(max_concurrent)
-    tasks = [_generate_one(document_text, c, sem, prefer_cloud, max_doc_chars) for c in chunks]
+    tasks = [
+        _generate_one(document_text, c, sem, prefer_cloud, max_doc_chars, sensitivity_level)
+        for c in chunks
+    ]
     contexts = await asyncio.gather(*tasks, return_exceptions=False)
     logger.info(
         "contextual_retrieval_generated",

ingestion/pipeline.py

@@ -241,6 +241,9 @@ class IngestionPipeline:
                     full_doc,
                     chunk_texts,
                     prefer_cloud=False,
+                    # HIGH docs keep their context-summary LLM call on local
+                    # inference — never route sensitive content to cloud at ingest.
+                    sensitivity_level=request.sensitivity_level.value,
                 )
                 embed_inputs = merge_chunks(chunk_texts, contexts)
                 logger.info(

interfaces/api.py

@@ -809,6 +809,10 @@ if _FASTAPI_AVAILABLE:
                 )
                 for p in points:
                     payload = p.payload or {}
+                    # Skip the purge sentinel point — it carries no source_file
+                    # and must never appear as a phantom upload row.
+                    if payload.get("__sentinel__"):
+                        continue
                     src = payload.get("source_file") or ""
                     # Reduce absolute paths down to a basename + sha so the
                     # visitor sees the filename they uploaded, not the

retrieval/qdrant_client.py

@@ -191,6 +191,13 @@ class QdrantManager:
                     field_name=field,
                     field_schema=schema,
                 )
+        # Stamp the collection with a creation timestamp so the 24h purge cron
+        # can actually find and drop it later (Qdrant has no writable collection
+        # metadata slot — see retrieval/session_purge.write_session_sentinel).
+        with contextlib.suppress(Exception):
+            from retrieval.session_purge import write_session_sentinel
+
+            write_session_sentinel(mgr._client, sess_collection, settings.embedding_dim)
         self._tenant_cache[sess_collection] = mgr
         logger.info(
             "byok_session_collection_cached",

retrieval/session_purge.py

@@ -23,6 +23,7 @@ See ``launch-plan/03-backend-byok.md`` § Session purge cron.
 
 from __future__ import annotations
 
+import uuid
 from datetime import UTC, datetime, timedelta
 from typing import TYPE_CHECKING, Any
 
@@ -35,6 +36,84 @@ if TYPE_CHECKING:
 logger = get_logger(__name__)
 
 
+# Deterministic namespace so the sentinel point id is stable across processes.
+_SENTINEL_NS = uuid.UUID("5a1d0000-0000-4000-8000-000000000001")
+
+
+def session_sentinel_id(collection_name: str) -> str:
+    """Stable UUID for a session collection's creation-timestamp sentinel point.
+
+    A fixed namespace UUID5 of the collection name, so the writer (at
+    collection-creation time) and the purge sweep agree on the id without any
+    shared state.
+    """
+    return str(uuid.uuid5(_SENTINEL_NS, collection_name))
+
+
+def write_session_sentinel(client: QdrantClient, collection_name: str, dim: int) -> None:
+    """Stamp a session collection with a creation timestamp the purge can read.
+
+    Qdrant ``CollectionInfo.config.params`` has no writable metadata slot, so the
+    original purge (which read ``config.params.metadata.created_at``) could never
+    find a timestamp and skipped every collection forever — they accumulated
+    until the 1 GB free tier filled. Instead we upsert one tiny sentinel point
+    carrying ``created_at``. It can never surface in retrieval: it has no
+    ``org_id`` / ``roles`` / ``sensitivity_level_int`` payload, so the RBAC
+    must-filter excludes it on every query.
+
+    Best-effort: a failure here only means the collection won't be auto-purged
+    (it still bounds itself by visitor inactivity), never a hard error.
+    """
+    from qdrant_client.http.models import PointStruct
+
+    sid = session_sentinel_id(collection_name)
+    try:
+        client.upsert(
+            collection_name=collection_name,
+            points=[
+                PointStruct(
+                    id=sid,
+                    vector=[0.0] * int(dim),
+                    payload={
+                        "__sentinel__": True,
+                        "created_at": datetime.now(UTC).isoformat(),
+                    },
+                )
+            ],
+        )
+    except Exception as exc:  # pragma: no cover - defensive
+        logger.warning("session_sentinel_write_failed", collection=collection_name, error=str(exc))
+
+
+def _read_created_at(client: QdrantClient, name: str) -> datetime | None:
+    """Resolve a session collection's creation time.
+
+    Production path: read the sentinel point's ``created_at`` payload. Legacy
+    fallback: the old ``config.params.metadata.created_at`` slot (kept so
+    pre-sentinel deployments + existing unit mocks still resolve).
+    """
+    # 1. Sentinel point (the real, durable path).
+    try:
+        points = client.retrieve(
+            collection_name=name,
+            ids=[session_sentinel_id(name)],
+            with_payload=True,
+        )
+        for p in points:
+            dt = _parse_created_at(getattr(p, "payload", None))
+            if dt is not None:
+                return dt
+    except Exception:
+        pass
+    # 2. Legacy collection-config metadata.
+    try:
+        info = client.get_collection(name)
+        meta = getattr(info.config.params, "metadata", None) or {}
+        return _parse_created_at(meta)
+    except Exception:
+        return None
+
+
 SESSION_COLLECTION_PREFIX = "_sess_"
 """Suffix introduced into the collection name by ``get_collection_name`` when
 ``byok_mode`` is on and a ``session_id`` is supplied. Used here to filter the
@@ -108,9 +187,7 @@ def purge_expired_sessions(
             continue
         inspected += 1
         try:
-            info = client.get_collection(name)
-            meta = getattr(info.config.params, "metadata", None) or {}
-            created = _parse_created_at(meta)
+            created = _read_created_at(client, name)
             if created is None:
                 # Undated -> skip; we don't delete what we can't time-stamp.
                 skipped += 1

utils/audit.py

@@ -13,6 +13,7 @@ re-ordering of past entries breaks the chain and is detected by
 from __future__ import annotations
 
 import hashlib
+import hmac
 import json
 import threading
 from datetime import UTC, date, datetime
@@ -60,10 +61,19 @@ class AuditEntry(BaseModel):
     entry_hash: str = ""
 
     def compute_hash(self) -> str:
-        """Compute the SHA-256 over canonical JSON of this entry except ``entry_hash``."""
+        """Hash the canonical JSON of this entry (excluding ``entry_hash``).
+
+        SHA-256 by default (tamper-evident). When ``settings.audit_hmac_key`` is
+        set the digest is HMAC-SHA256 keyed by that secret (tamper-resistant) —
+        ``verify_chain`` recomputes the same way, so flipping the key on a fresh
+        chain upgrades the integrity guarantee with no other code change.
+        """
         payload = self.model_dump(mode="json", exclude={"entry_hash"})
-        canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
-        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+        canonical = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
+        key = settings.audit_hmac_key
+        if key:
+            return hmac.new(key.encode("utf-8"), canonical, hashlib.sha256).hexdigest()
+        return hashlib.sha256(canonical).hexdigest()
 
 
 class AuditLogger: