feat: RAG upload sanitisation layer before vector indexing

LLM + regex strip of PII/confidential data from tenant uploads and runtime RAG sync before embedding. On-disk files unchanged. Configurable via ENABLE_RAG_UPLOAD_SANITISATION.

Co-authored-by: Cursor <cursoragent@cursor.com>

.env.example

@@ -48,6 +48,13 @@ OPENAI_API_KEY=sk-replace-me
 # GENERATION_TIMEOUT_SECONDS=3600
 # GENERATION_STALE_SWEEP_SECONDS=120
 
+# ── RAG upload sanitisation (strip PII before vector indexing; on-disk files unchanged) ─
+# ENABLE_RAG_UPLOAD_SANITISATION=true
+# RAG_SANITISATION_SKIP_KB_TENANT=true
+# RAG_SANITISATION_CHUNK_CHARS=12000
+# RAG_SANITISATION_MAX_OUTPUT_TOKENS=4096
+# RAG_SANITISATION_FAIL_CLOSED=true
+
 # ── Database ──────────────────────────────────────────────────────────────────
 DATABASE_URL=sqlite+aiosqlite:///./dev.db
 

app/config.py

@@ -504,6 +504,38 @@ class Settings(BaseSettings):
         ),
     )
 
+    # ── RAG upload sanitisation (PII/confidential stripping before indexing) ─
+    enable_rag_upload_sanitisation: bool = Field(
+        default=True,
+        description=(
+            "When true, tenant uploads are sanitised (LLM + regex fallback) before "
+            "text is embedded into the vector index for style/RAG retrieval. On-disk files are not rewritten."
+        ),
+    )
+    rag_sanitisation_skip_kb_tenant: bool = Field(
+        default=True,
+        description="Skip sanitisation for the reserved knowledge-base tenant (e.g. __rics_kb__).",
+    )
+    rag_sanitisation_chunk_chars: int = Field(
+        default=12_000,
+        ge=2_000,
+        le=50_000,
+        description="Max characters per LLM sanitisation call for long reports.",
+    )
+    rag_sanitisation_max_output_tokens: int = Field(
+        default=4096,
+        ge=512,
+        le=16_384,
+        description="Max tokens per sanitisation LLM response chunk.",
+    )
+    rag_sanitisation_fail_closed: bool = Field(
+        default=True,
+        description=(
+            "If true, ingestion/runtime RAG sync fails when sanitisation would store empty text. "
+            "When false, regex fallback is used even when the LLM returns nothing."
+        ),
+    )
+
     # ── Rate limiting ────────────────────────────────────────────────────────
     rate_limit_generate_rpm: int = Field(
         default=20,

app/ingest/pipeline.py

@@ -30,6 +30,7 @@ from app.db.models import IngestStatus
 from app.ingest.hierarchy import build_hierarchical_documents
 from app.ingest.parser_docx import parse_docx
 from app.ingest.parser_pdf import parse_pdf
+from app.services.document_sanitiser import DocumentSanitisationError
 from app.vectorstore.factory import get_vectorstore
 
 logger = logging.getLogger(__name__)
@@ -73,6 +74,21 @@ async def ingest_document(doc_id: str, file_path: Path) -> None:
                 raw_docs = _load_file(file_path)
                 logger.debug("doc=%s loaded %d raw document(s)", doc_id, len(raw_docs))
 
+                from app.services.document_sanitiser import (
+                    sanitise_langchain_documents_sync,
+                    should_sanitise_for_rag,
+                )
+
+                if should_sanitise_for_rag(doc.tenant_id):
+                    logger.info(
+                        "Sanitising doc=%s tenant=%s before RAG indexing",
+                        doc_id,
+                        doc.tenant_id,
+                    )
+                    raw_docs = sanitise_langchain_documents_sync(
+                        raw_docs, tenant_id=doc.tenant_id
+                    )
+
                 # ── 2. Hierarchical chunks (document → section → paragraph) ───
                 hier = build_hierarchical_documents(
                     raw_docs,
@@ -109,6 +125,15 @@ async def ingest_document(doc_id: str, file_path: Path) -> None:
                 chunks_count,
             )
 
+        except DocumentSanitisationError as exc:
+            logger.error(
+                "Ingestion blocked by sanitisation doc=%s tenant=%s: %s",
+                doc_id,
+                doc.tenant_id,
+                exc,
+            )
+            doc.status = IngestStatus.failed
+            doc.error_message = f"Document sanitisation failed: {exc}"
         except TimeoutError:
             logger.error(
                 "Ingestion timed out doc=%s tenant=%s after %ss",

app/optimization/health_warnings.py

@@ -64,6 +64,11 @@ def collect_optimization_warnings() -> list[str]:
             "ENABLE_JOB_QUEUE=true requires REDIS_URL; "
             "generation will fall back to in-process tasks until Redis is configured."
         )
+    if settings.enable_rag_upload_sanitisation and not (settings.openai_api_key or "").strip():
+        warnings.append(
+            "ENABLE_RAG_UPLOAD_SANITISATION=true but OPENAI_API_KEY is unset; "
+            "uploads are indexed using regex-only redaction (weaker than the LLM sanitiser)."
+        )
     return warnings
 
 

app/services/document_sanitiser.py

@@ -0,0 +1,333 @@
+"""LLM + regex sanitisation before tenant uploads are indexed for RAG / style learning.
+
+On-disk uploads are unchanged; only text written to the vector index is sanitised.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from typing import TYPE_CHECKING
+
+from app.config import settings
+
+if TYPE_CHECKING:
+    from langchain_core.documents import Document
+
+logger = logging.getLogger(__name__)
+
+
+class DocumentSanitisationError(RuntimeError):
+    """Raised when sanitisation is required but cannot complete safely."""
+
+
+RAG_UPLOAD_SANITISATION_SYSTEM_PROMPT = """\
+You are a document sanitisation assistant for a professional property report-writing platform.
+
+The purpose of this process is to prepare users' previously written reports for storage in a private user-specific RAG system. These reports will be used solely to learn the user's preferred writing style, report structure, technical reasoning, and professional wording.
+
+Your task is to remove all personal, confidential, identifying, financial, legal, and transaction-specific information while preserving the technical content, report format, headings, narrative style, and professional observations.
+
+IMPORTANT INSTRUCTIONS
+
+- Preserve the report's structure, headings, section order, and formatting wherever possible.
+- Preserve the user's writing style, sentence structure, tone, and terminology.
+- Preserve all technical observations, building pathology descriptions, assumptions, caveats, and recommendations.
+- Preserve generic market commentary and valuation reasoning.
+- Preserve standard boilerplate text and report templates.
+- Remove sensitive information entirely rather than replacing it with placeholders.
+- Do not summarise, rewrite, shorten, or paraphrase unless necessary to remove identifying information.
+- Do not invent any new information.
+- Ensure that the resulting document contains no information that could identify a specific individual, property, company, or transaction.
+
+REMOVE THE FOLLOWING INFORMATION
+
+PERSONAL INFORMATION
+- Client names
+- Occupier names
+- Tenant names
+- Vendor names
+- Purchaser names
+- Solicitor names
+- Estate agent names
+- Surveyor names
+- Witness names
+- Signatures
+- Initials where they identify a person
+- Dates of birth
+
+CONTACT INFORMATION
+- Email addresses
+- Telephone numbers
+- Mobile numbers
+- Fax numbers
+- Website URLs where they identify a specific business or individual
+
+ADDRESS INFORMATION
+- Full property addresses
+- Flat numbers
+- House numbers
+- Building names
+- Street names
+- Towns and cities where they identify the subject property
+- Postcodes
+- Correspondence addresses
+
+PROPERTY IDENTIFIERS
+- Land Registry title numbers
+- UPRNs
+- EPC certificate numbers
+- Planning application numbers
+- Building regulation references
+- Local authority references
+- Council tax account numbers
+
+PROFESSIONAL IDENTIFIERS
+- RICS membership numbers
+- Registration numbers
+- Company registration numbers
+- VAT numbers
+- Insurance policy numbers
+- Certificate numbers
+- Job numbers
+- File references
+- Invoice numbers
+
+FINANCIAL INFORMATION
+- Valuation figures
+- Purchase prices
+- Sale prices
+- Premiums
+- Mortgage balances
+- Ground rent amounts
+- Service charge amounts
+- Insurance claim amounts
+- Bank account details
+- Sort codes
+- IBAN numbers
+
+LEGAL INFORMATION
+- Tribunal case numbers
+- Court claim numbers
+- Lease numbers
+- Policy claim references
+- Solicitor reference numbers
+
+DATES
+- Inspection dates
+- Report dates
+- Exchange dates
+- Completion dates
+- Any specific dates linked to an identifiable matter or transaction
+
+DIGITAL IDENTIFIERS
+- IP addresses
+- GPS coordinates
+- Metadata references
+- QR code content
+
+SPECIAL CATEGORY OR CONFIDENTIAL INFORMATION
+- Medical or health information
+- Disability information
+- Family circumstances
+- Complaints
+- Litigation details
+- Insurance disputes
+- Employment information
+- Any other confidential notes
+
+OUTPUT REQUIREMENTS
+
+- Return only the sanitised report text.
+- Preserve as much useful stylistic and technical content as possible.
+- Remove all identifying and confidential information.
+- Ensure the resulting text is suitable for storage in a private user-specific RAG system for personalised AI report generation.\
+"""
+
+_POSTCODE_RE = re.compile(
+    r"\b([A-Z]{1,2}\d[A-Z\d]?\s*\d[A-Z]{2})\b", re.IGNORECASE
+)
+_ADDRESS_LINE_RE = re.compile(
+    r"\b(\d{1,4}\s+[A-Za-z][A-Za-z'\-]*(?:\s+[A-Za-z][A-Za-z'\-]*){0,6}\s+"
+    r"(Road|Rd|Street|St|Avenue|Ave|Lane|Ln|Drive|Dr|Crescent|Close|Place|Way|Gardens|Gdns|Court|Ct|Terrace|Terr))\b",
+    re.IGNORECASE,
+)
+_MONEY_RE = re.compile(
+    r"(£\s*\d[\d,]*(?:\.\d+)?|\b\d[\d,]*(?:\.\d+)?\s*(?:gbp|pounds)\b)",
+    re.IGNORECASE,
+)
+_DATE_RE = re.compile(
+    r"\b(?:\d{1,2}[/-]\d{1,2}[/-]\d{2,4}|\d{1,2}\s+"
+    r"(?:jan|feb|mar|apr|may|jun|jul|aug|sep|sept|oct|nov|dec)[a-z]*\s+\d{2,4})\b",
+    re.IGNORECASE,
+)
+_LONG_NUMBER_RE = re.compile(r"\b\d{5,}\b")
+_EMAIL_RE = re.compile(r"\b[\w.+-]+@[\w.-]+\.[A-Za-z]{2,}\b")
+_PHONE_RE = re.compile(
+    r"\b(?:\+44\s?|0)(?:\d[\s-]?){9,12}\b|\b\d{3,4}[\s-]\d{3,4}[\s-]\d{3,4}\b"
+)
+_URL_RE = re.compile(r"https?://[^\s<>\"']+", re.IGNORECASE)
+
+
+def should_sanitise_for_rag(tenant_id: str) -> bool:
+    """Whether uploads for this tenant should be sanitised before vector indexing."""
+    if not settings.enable_rag_upload_sanitisation:
+        return False
+    if (
+        settings.rag_sanitisation_skip_kb_tenant
+        and tenant_id == settings.knowledge_base_tenant_id
+    ):
+        return False
+    return True
+
+
+def regex_sanitise_text(text: str) -> str:
+    """Deterministic redaction when the LLM path is unavailable or as a safety net."""
+    t = (text or "").strip()
+    if not t:
+        return ""
+    t = _EMAIL_RE.sub("", t)
+    t = _URL_RE.sub("", t)
+    t = _PHONE_RE.sub("", t)
+    t = _POSTCODE_RE.sub("", t)
+    t = _ADDRESS_LINE_RE.sub("", t)
+    t = _MONEY_RE.sub("", t)
+    t = _DATE_RE.sub("", t)
+    t = _LONG_NUMBER_RE.sub("", t)
+    t = re.sub(r"[ \t]{2,}", " ", t)
+    t = re.sub(r"\n{3,}", "\n\n", t)
+    return t.strip()
+
+
+def _split_text_chunks(text: str, max_chars: int) -> list[str]:
+    body = (text or "").strip()
+    if not body:
+        return []
+    if len(body) <= max_chars:
+        return [body]
+    parts = re.split(r"(\n\s*\n)", body)
+    chunks: list[str] = []
+    current = ""
+    for part in parts:
+        if not part:
+            continue
+        candidate = current + part
+        if len(candidate) <= max_chars:
+            current = candidate
+            continue
+        if current.strip():
+            chunks.append(current.strip())
+        if len(part) <= max_chars:
+            current = part
+        else:
+            for i in range(0, len(part), max_chars):
+                segment = part[i : i + max_chars].strip()
+                if segment:
+                    chunks.append(segment)
+            current = ""
+    if current.strip():
+        chunks.append(current.strip())
+    return chunks or [body[:max_chars]]
+
+
+def _llm_sanitise_chunk_sync(chunk: str, *, part: int, total: int) -> str:
+    from app.llm.openai_chat import chat_completions_create
+
+    import asyncio
+
+    prefix = (
+        f"Sanitise this excerpt from a RICS property survey report "
+        f"(part {part} of {total}).\n\n"
+    )
+    user_content = prefix + chunk
+
+    async def _run() -> str:
+        return await chat_completions_create(
+            messages=[
+                {"role": "system", "content": RAG_UPLOAD_SANITISATION_SYSTEM_PROMPT},
+                {"role": "user", "content": user_content},
+            ],
+            model=settings.chat_model,
+            max_tokens=int(settings.rag_sanitisation_max_output_tokens),
+            temperature=0.0,
+            phase="rag_upload_sanitise",
+        )
+
+    try:
+        return asyncio.run(_run())
+    except RuntimeError:
+        # Already inside a running loop (should not happen from ingest thread).
+        import concurrent.futures
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
+            return pool.submit(asyncio.run, _run()).result()
+
+
+def sanitise_text_for_rag_sync(text: str, *, tenant_id: str) -> str:
+    """Sanitise plain text before it is embedded into the tenant vector index."""
+    if not should_sanitise_for_rag(tenant_id):
+        return (text or "").strip()
+
+    raw = (text or "").strip()
+    if not raw:
+        return ""
+
+    max_chars = int(settings.rag_sanitisation_chunk_chars)
+    chunks = _split_text_chunks(raw, max_chars)
+    has_key = bool((settings.openai_api_key or "").strip())
+    out_parts: list[str] = []
+
+    for i, chunk in enumerate(chunks, start=1):
+        cleaned = ""
+        if has_key:
+            try:
+                cleaned = (_llm_sanitise_chunk_sync(chunk, part=i, total=len(chunks)) or "").strip()
+            except Exception as exc:
+                logger.warning(
+                    "RAG sanitisation LLM failed tenant=%s part=%d/%d: %s",
+                    tenant_id,
+                    i,
+                    len(chunks),
+                    exc,
+                )
+        if not cleaned:
+            cleaned = regex_sanitise_text(chunk)
+            if has_key:
+                logger.info(
+                    "RAG sanitisation using regex fallback tenant=%s part=%d/%d",
+                    tenant_id,
+                    i,
+                    len(chunks),
+                )
+        if not cleaned and settings.rag_sanitisation_fail_closed:
+            raise DocumentSanitisationError(
+                f"Sanitisation produced empty text for tenant={tenant_id} part={i}/{len(chunks)}"
+            )
+        out_parts.append(cleaned)
+
+    combined = "\n\n".join(p for p in out_parts if p).strip()
+    if not combined and settings.rag_sanitisation_fail_closed:
+        raise DocumentSanitisationError(
+            f"Sanitisation produced empty document for tenant={tenant_id}"
+        )
+    return combined or regex_sanitise_text(raw)
+
+
+def sanitise_langchain_documents_sync(
+    documents: list[Document],
+    *,
+    tenant_id: str,
+) -> list[Document]:
+    """Return documents whose ``page_content`` has been sanitised for RAG indexing."""
+    if not should_sanitise_for_rag(tenant_id):
+        return documents
+
+    from langchain_core.documents import Document as LCDocument
+
+    out: list[LCDocument] = []
+    for doc in documents:
+        meta = dict(doc.metadata or {})
+        cleaned = sanitise_text_for_rag_sync(doc.page_content or "", tenant_id=tenant_id)
+        meta["rag_sanitised"] = True
+        out.append(LCDocument(page_content=cleaned, metadata=meta))
+    return out

app/services/runtime_rag_index.py

@@ -37,6 +37,14 @@ def upsert_runtime_section_index(
     if not text:
         logger.info("runtime RAG: cleared index for doc=%s", virtual_doc_id)
         return 0
+
+    from app.services.document_sanitiser import sanitise_text_for_rag_sync
+
+    text = sanitise_text_for_rag_sync(text, tenant_id=tenant_id)
+    if not text:
+        logger.info("runtime RAG: empty after sanitisation doc=%s", virtual_doc_id)
+        return 0
+
     label = (section_title or "").strip() or virtual_doc_id
     raw = [Document(page_content=text, metadata={"source": label})]
     hier = build_hierarchical_documents(

app/tests/test_document_sanitiser.py

@@ -0,0 +1,45 @@
+"""Tests for RAG upload sanitisation (regex path; no live OpenAI)."""
+
+from __future__ import annotations
+
+import pytest
+
+from app.services.document_sanitiser import (
+    regex_sanitise_text,
+    sanitise_text_for_rag_sync,
+    should_sanitise_for_rag,
+)
+
+
+def test_should_skip_kb_tenant(monkeypatch: pytest.MonkeyPatch) -> None:
+    from app.config import settings
+
+    monkeypatch.setattr(settings, "enable_rag_upload_sanitisation", True)
+    monkeypatch.setattr(settings, "rag_sanitisation_skip_kb_tenant", True)
+    monkeypatch.setattr(settings, "knowledge_base_tenant_id", "__rics_kb__")
+    assert should_sanitise_for_rag("tenant-a") is True
+    assert should_sanitise_for_rag("__rics_kb__") is False
+
+
+def test_regex_removes_postcode_and_email() -> None:
+    raw = (
+        "The roof covering is slate. Client: jane.doe@example.com. "
+        "Property at 12 High Street, London SW1A 1AA. Price £450,000."
+    )
+    out = regex_sanitise_text(raw)
+    assert "jane.doe@example.com" not in out
+    assert "SW1A" not in out
+    assert "450,000" not in out
+    assert "slate" in out.lower()
+
+
+def test_sanitise_sync_without_openai_key(monkeypatch: pytest.MonkeyPatch) -> None:
+    from app.config import settings
+
+    monkeypatch.setattr(settings, "enable_rag_upload_sanitisation", True)
+    monkeypatch.setattr(settings, "openai_api_key", "")
+    monkeypatch.setattr(settings, "rag_sanitisation_fail_closed", False)
+    text = "Defect noted at 10 Church Road, Bristol BS1 4ST."
+    out = sanitise_text_for_rag_sync(text, tenant_id="firm-1")
+    assert "BS1" not in out
+    assert "defect" in out.lower()