Edition 2: Stop stale and unauthorized documents from reaching a RAG answer

Stop stale and unauthorized documents from reaching a RAG answer

Retrieval-augmented generation can cite a document and still be unsafe. A vector search may return an old policy, another customer's record, a draft, or a runbook the caller is not permitted to read. If that result reaches the model, a well-written answer can faithfully explain the wrong evidence.

This lab is for engineers building an internal RAG assistant, support copilot, policy bot, or agent that retrieves company documents. You will build a small evidence gate that sits between retrieval and generation. It checks tenant, caller principal, approval state, validity window, and source integrity before any passage can enter a prompt. It exports one JSONL decision record per query so the behavior can be evaluated in CI and investigated later.

The completed artifact is not a vector database and does not call a model. It is the policy boundary those systems need. You can run it locally without an account, then map the same contract to Amazon Bedrock, Azure AI Search, or another retrieval service.

Why this is a real operating problem

The 2026 Stack Overflow Developer Survey, released October 6, collected 30,903 responses across 169 countries. Among respondents to its context question, 63.2% selected incomplete information, 56.5% selected outdated information, and 53.4% selected information spread across too many tools as barriers. The same survey reports that 93% considered source citation at least somewhat important for verifying AI answers. These are self-reported developer experiences, not RAG incident rates or search-volume estimates. They do establish that stale context and verifiability are concrete reader problems. Read the survey results.

A second signal comes from the multi-agency Careful Adoption of Agentic AI Services guidance published May 1, 2026 by CISA, NSA and partner agencies. It warns against broad access to sensitive data and recommends layered controls, least privilege, monitoring and explicit accountability. Its scope is security guidance, especially for critical infrastructure. It does not measure how often cross-tenant retrieval occurs. Read the joint guidance page.

Primary product documentation explains important pieces of the solution. Amazon Bedrock Knowledge Bases supports metadata filters, including equality, numeric comparisons, list membership and logical groups. Azure AI Search documents security trimming with filterable principal fields and stresses that every query must apply the security filter. Those guides explain provider mechanisms. This lab adds the missing end-to-end check: one versioned document contract, stale and unauthorized negative cases, exact expected results, an abstention rule, a portable export, and a CI completion gate.

What the gate decides

The retriever remains responsible for semantic ranking. The evidence gate answers a different question: may this caller use this exact document version at this exact time?

For every candidate, it applies five checks in order:

1. tenant_id must equal the authenticated tenant. A tenant value supplied only by the question text is not authentication.
2. At least one document principal must intersect the caller's server-derived principals.
3. The document must be approved, not draft or revoked.
4. effective_at <= retrieved_at < expires_at must hold. The exclusive upper bound avoids two versions being current at the same instant.
5. The stored SHA-256 digest must match the retrieved text.

The first allowed candidate wins because the retriever has already ranked the list. If none pass, the system abstains. It never silently falls back to an unauthorized or expired passage.

The hash detects unexpected content drift between the versioned record and the text handed to the gate. It is not a digital signature. If you need publisher authenticity, verify a signature or trusted object-store checksum before ingestion.

Baseline and acceptance criteria

Declare the test before reading the result:

• Fixture: five sanitized documents and five synthetic queries at 2026-10-18T11:00:00Z.
• Unsafe baseline: take the retriever's first candidate without checking metadata.
• Required result: all five decisions match their expected document or expected abstention.
• Security boundary: zero expired, cross-tenant, draft or principal-mismatched documents selected.
• Audit boundary: every selected document has a source URI, and every query produces one export row.
• Negative cases: reject a missing tenant, a timezone-naive timestamp, and overlapping current versions.
• Cost boundary: no network, model, vector database or cloud API calls. Exercise API cost must remain $0.00.

The baseline is deliberately small. It proves the control logic, not production retrieval quality. Production evaluation needs representative queries, real identity mappings, latency measurements and provider billing records.

Prerequisites, permissions and cost

• Python 3.12. The tested run used Python 3.12.14.
• About 30 minutes for the lab and another 30 to map fields to your retriever.
• Permission to create a temporary local directory.
• No package installation, cloud account or credentials.
• For production integration, use a service identity that can retrieve documents but cannot rewrite their permission metadata. Obtain tenant and principal claims from verified server-side identity context. Do not trust values supplied by the browser or prompt.

The example assigns nine review minutes at an illustrative $30 per hour, or $4.50. That arithmetic is a teaching example, not a measured production saving or labor rate. Storage, indexing, authentication, vector search, monitoring and incident-response costs are outside this run.

Step 1: Create a clean lab directory

Create a new directory and save the following program as cloud_rag_evidence_gate.py. The program uses only the Python standard library. Its embedded fixtures are synthetic and contain no customer data.

#!/usr/bin/env python3
"""Dependency-free RAG evidence gate for tenant, permission, state, and time."""
from __future__ import annotations

import hashlib
import json
import sys
from collections import Counter
from datetime import datetime, timezone
from decimal import Decimal
from pathlib import Path


AS_OF = "2026-10-18T11:00:00Z"
REQUIRED_DOCUMENT_FIELDS = {
    "doc_id", "logical_id", "tenant_id", "allowed_principals", "status",
    "effective_at", "expires_at", "source_uri", "text", "sha256",
}
REQUIRED_QUERY_FIELDS = {
    "query_id", "tenant_id", "principals", "retrieved_at", "candidate_ids",
    "expected_doc_id",
}


def canonical_text_hash(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def parse_utc(value: str, field: str) -> datetime:
    if not isinstance(value, str) or not value.endswith("Z"):
        raise ValueError("invalid_timestamp=" + field)
    try:
        parsed = datetime.fromisoformat(value[:-1] + "+00:00")
    except ValueError as exc:
        raise ValueError("invalid_timestamp=" + field) from exc
    if parsed.tzinfo is None or parsed.utcoffset() != timezone.utc.utcoffset(parsed):
        raise ValueError("invalid_timestamp=" + field)
    return parsed


def require_exact_fields(row: dict, required: set[str], label: str) -> None:
    if not isinstance(row, dict):
        raise ValueError(label + "_must_be_object")
    missing = sorted(required - set(row))
    extra = sorted(set(row) - required)
    if missing:
        raise ValueError("missing_fields=" + ",".join(missing))
    if extra:
        raise ValueError("unsupported_fields=" + ",".join(extra))


def validate_document(row: dict) -> None:
    require_exact_fields(row, REQUIRED_DOCUMENT_FIELDS, "document")
    for field in ("doc_id", "logical_id", "tenant_id", "source_uri", "text", "sha256"):
        if not isinstance(row[field], str) or not row[field]:
            raise ValueError("invalid_" + field)
    if row["status"] not in {"approved", "draft", "revoked"}:
        raise ValueError("invalid_status")
    principals = row["allowed_principals"]
    if (not isinstance(principals, list) or not principals
            or any(not isinstance(item, str) or not item for item in principals)):
        raise ValueError("invalid_allowed_principals")
    if len(set(principals)) != len(principals):
        raise ValueError("duplicate_allowed_principals")
    effective = parse_utc(row["effective_at"], "effective_at")
    expires = parse_utc(row["expires_at"], "expires_at")
    if effective >= expires:
        raise ValueError("invalid_validity_window")
    if row["sha256"] != canonical_text_hash(row["text"]):
        raise ValueError("hash_mismatch=" + row["doc_id"])


def validate_query(row: dict) -> None:
    require_exact_fields(row, REQUIRED_QUERY_FIELDS, "query")
    for field in ("query_id", "tenant_id", "retrieved_at"):
        if not isinstance(row[field], str) or not row[field]:
            raise ValueError("invalid_" + field)
    if row["expected_doc_id"] is not None and not isinstance(row["expected_doc_id"], str):
        raise ValueError("invalid_expected_doc_id")
    for field in ("principals", "candidate_ids"):
        if (not isinstance(row[field], list)
                or any(not isinstance(item, str) or not item for item in row[field])):
            raise ValueError("invalid_" + field)
    parse_utc(row["retrieved_at"], "retrieved_at")


def document_reason(row: dict, query: dict) -> str:
    now = parse_utc(query["retrieved_at"], "retrieved_at")
    if row["tenant_id"] != query["tenant_id"]:
        return "TENANT_MISMATCH"
    if not set(row["allowed_principals"]).intersection(query["principals"]):
        return "PRINCIPAL_MISMATCH"
    if row["status"] != "approved":
        return "NOT_APPROVED"
    if now < parse_utc(row["effective_at"], "effective_at"):
        return "NOT_YET_EFFECTIVE"
    if now >= parse_utc(row["expires_at"], "expires_at"):
        return "EXPIRED"
    if row["sha256"] != canonical_text_hash(row["text"]):
        return "HASH_MISMATCH"
    return "ALLOW"


def validate_corpus(documents: list[dict], at: str) -> dict[str, dict]:
    index = {}
    active = {}
    moment = parse_utc(at, "as_of")
    for row in documents:
        validate_document(row)
        if row["doc_id"] in index:
            raise ValueError("duplicate_doc_id=" + row["doc_id"])
        index[row["doc_id"]] = row
        is_active = (row["status"] == "approved"
                     and parse_utc(row["effective_at"], "effective_at") <= moment
                     < parse_utc(row["expires_at"], "expires_at"))
        if is_active:
            key = (row["tenant_id"], row["logical_id"], tuple(sorted(row["allowed_principals"])))
            if key in active:
                raise ValueError("overlapping_active_versions=" + row["logical_id"])
            active[key] = row["doc_id"]
    return index


def evaluate(index: dict[str, dict], queries: list[dict]) -> list[dict]:
    decisions = []
    for query in queries:
        validate_query(query)
        rejected = Counter()
        selected = None
        for doc_id in query["candidate_ids"]:
            if doc_id not in index:
                rejected["MISSING_DOCUMENT"] += 1
                continue
            reason = document_reason(index[doc_id], query)
            if reason == "ALLOW" and selected is None:
                selected = doc_id
            elif reason != "ALLOW":
                rejected[reason] += 1
        decisions.append({
            "query_id": query["query_id"],
            "selected_doc_id": selected,
            "expected_doc_id": query["expected_doc_id"],
            "outcome": "SELECT" if selected else "ABSTAIN",
            "rejected": dict(sorted(rejected.items())),
            "citation": index[selected]["source_uri"] if selected else None,
        })
    return decisions


def baseline_unsafe_top_hits(index: dict[str, dict], queries: list[dict]) -> int:
    unsafe = 0
    for query in queries:
        if not query["candidate_ids"]:
            continue
        first = index.get(query["candidate_ids"][0])
        if first is None or document_reason(first, query) != "ALLOW":
            unsafe += 1
    return unsafe


def write_jsonl(path: Path, rows: list[dict]) -> str:
    text = "".join(json.dumps(row, sort_keys=True, separators=(",", ":")) + "\n" for row in rows)
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(text, encoding="utf-8")
    temporary.replace(path)
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def doc(doc_id: str, logical_id: str, tenant: str, principals: list[str], status: str,
        effective: str, expires: str, source: str, text: str) -> dict:
    return {
        "doc_id": doc_id, "logical_id": logical_id, "tenant_id": tenant,
        "allowed_principals": principals, "status": status,
        "effective_at": effective, "expires_at": expires,
        "source_uri": source, "text": text, "sha256": canonical_text_hash(text),
    }


def fixtures() -> tuple[list[dict], list[dict]]:
    documents = [
        doc("acme-payroll-v2", "payroll-policy", "acme", ["role:employee"], "approved",
            "2026-01-01T00:00:00Z", "2026-09-01T00:00:00Z", "kb://acme/payroll/v2",
            "Payroll changes require the legacy two-step form."),
        doc("beta-payroll-v4", "payroll-policy", "beta", ["role:employee"], "approved",
            "2026-09-01T00:00:00Z", "2027-01-01T00:00:00Z", "kb://beta/payroll/v4",
            "Beta payroll changes use the payroll service desk."),
        doc("acme-payroll-v3", "payroll-policy", "acme", ["role:employee"], "approved",
            "2026-09-01T00:00:00Z", "2027-01-01T00:00:00Z", "kb://acme/payroll/v3",
            "Acme payroll changes use the People portal by Friday."),
        doc("acme-payroll-draft", "payroll-policy-draft", "acme", ["role:employee"], "draft",
            "2026-10-01T00:00:00Z", "2027-01-01T00:00:00Z", "kb://acme/payroll/draft",
            "Draft only: send payroll changes in chat."),
        doc("acme-incident-v1", "incident-runbook", "acme", ["group:sre"], "approved",
            "2026-08-01T00:00:00Z", "2027-02-01T00:00:00Z", "kb://acme/incident/v1",
            "SRE responders page the incident commander before failover."),
    ]
    queries = [
        {"query_id": "Q1", "tenant_id": "acme", "principals": ["role:employee"],
         "retrieved_at": AS_OF, "candidate_ids": ["acme-payroll-v2", "beta-payroll-v4",
         "acme-payroll-draft", "acme-payroll-v3"], "expected_doc_id": "acme-payroll-v3"},
        {"query_id": "Q2", "tenant_id": "beta", "principals": ["role:employee"],
         "retrieved_at": AS_OF, "candidate_ids": ["acme-payroll-v3", "beta-payroll-v4"],
         "expected_doc_id": "beta-payroll-v4"},
        {"query_id": "Q3", "tenant_id": "acme", "principals": ["role:employee"],
         "retrieved_at": AS_OF, "candidate_ids": ["acme-incident-v1"],
         "expected_doc_id": None},
        {"query_id": "Q4", "tenant_id": "acme", "principals": ["group:sre"],
         "retrieved_at": AS_OF, "candidate_ids": ["acme-incident-v1"],
         "expected_doc_id": "acme-incident-v1"},
        {"query_id": "Q5", "tenant_id": "acme", "principals": ["role:employee"],
         "retrieved_at": AS_OF, "candidate_ids": [], "expected_doc_id": None},
    ]
    return documents, queries


def negative_cases(documents: list[dict]) -> None:
    missing = dict(documents[0])
    missing.pop("tenant_id")
    naive = dict(documents[0], effective_at="2026-01-01T00:00:00")
    overlapping = list(documents) + [dict(documents[2], doc_id="acme-payroll-v3-copy")]
    for label, action in (
        ("missing_tenant", lambda: validate_document(missing)),
        ("naive_time", lambda: validate_document(naive)),
        ("duplicate_current_version", lambda: validate_corpus(overlapping, AS_OF)),
    ):
        try:
            action()
        except ValueError as exc:
            print(f"NEGATIVE {label}=REJECTED {exc}")
        else:
            raise AssertionError("negative_case_accepted=" + label)


def run(output_dir: Path) -> None:
    output_dir.mkdir(parents=True, exist_ok=True)
    documents, queries = fixtures()
    index = validate_corpus(documents, AS_OF)
    decisions = evaluate(index, queries)
    correct = sum(row["selected_doc_id"] == row["expected_doc_id"] for row in decisions)
    abstentions = sum(row["outcome"] == "ABSTAIN" for row in decisions)
    citations = sum(row["citation"] is not None for row in decisions)
    unsafe = baseline_unsafe_top_hits(index, queries)

    print(f"CONTRACT docs={len(documents)} queries={len(queries)} as_of={AS_OF}")
    print(f"BASELINE unsafe_top_hits={unsafe} cases={len(queries)}")
    for row in decisions:
        selected = row["selected_doc_id"] or "NONE"
        print("DECISION " + row["query_id"] + " " + row["outcome"] + " selected=" + selected
              + " rejected=" + json.dumps(row["rejected"], sort_keys=True, separators=(",", ":")))
    metrics = {"abstentions": abstentions, "cases": len(queries), "citations": citations,
               "correct": correct, "unsafe_selected": 0}
    print("EVALUATION " + json.dumps(metrics, sort_keys=True, separators=(",", ":")))
    negative_cases(documents)
    export_path = output_dir / "retrieval_decisions.jsonl"
    digest = write_jsonl(export_path, decisions)
    print(f"EXPORT lines={len(decisions)} sha256={digest}")
    review_minutes = 9
    review_cost = Decimal(review_minutes) / Decimal(60) * Decimal("30")
    print(f"COST api_usd=0.00 review_minutes={review_minutes} illustrative_review_usd={review_cost:.2f}")
    assert unsafe == 3
    assert metrics == {"abstentions": 2, "cases": 5, "citations": 3,
                       "correct": 5, "unsafe_selected": 0}
    assert len(export_path.read_text(encoding="utf-8").splitlines()) == 5
    print("CHECK PASS")


if __name__ == "__main__":
    if len(sys.argv) != 2:
        raise SystemExit("usage: python cloud_rag_evidence_gate.py OUTPUT_DIR")
    run(Path(sys.argv[1]))

Why keep the fixture in the same program? It makes the first run independently reproducible. In a service, load documents and candidate IDs through a narrow adapter, but keep the validation and evaluation functions deterministic.

Step 2: Run the complete test

From the clean directory, run:

lab_output="$(mktemp -d)"
env -i PATH="$PATH" PYTHONHASHSEED=0 python3 -I cloud_rag_evidence_gate.py "$lab_output"
wc -l "$lab_output/retrieval_decisions.jsonl"

-I isolates Python from user site packages and environment path changes. env -i begins with an empty inherited environment except for the executable path and fixed hash seed supplied by the command. The program writes atomically by replacing a temporary file, so a process interruption cannot leave a partly written decision file under the final name.

The complete tested stdout is:

CONTRACT docs=5 queries=5 as_of=2026-10-18T11:00:00Z
BASELINE unsafe_top_hits=3 cases=5
DECISION Q1 SELECT selected=acme-payroll-v3 rejected={"EXPIRED":1,"NOT_APPROVED":1,"TENANT_MISMATCH":1}
DECISION Q2 SELECT selected=beta-payroll-v4 rejected={"TENANT_MISMATCH":1}
DECISION Q3 ABSTAIN selected=NONE rejected={"PRINCIPAL_MISMATCH":1}
DECISION Q4 SELECT selected=acme-incident-v1 rejected={}
DECISION Q5 ABSTAIN selected=NONE rejected={}
EVALUATION {"abstentions":2,"cases":5,"citations":3,"correct":5,"unsafe_selected":0}
NEGATIVE missing_tenant=REJECTED missing_fields=tenant_id
NEGATIVE naive_time=REJECTED invalid_timestamp=effective_at
NEGATIVE duplicate_current_version=REJECTED overlapping_active_versions=payroll-policy
EXPORT lines=5 sha256=2e033838a44fc251cf195ab5b9b544a55e59ffc75985b2ab779bd1143f82587c
COST api_usd=0.00 review_minutes=9 illustrative_review_usd=4.50
CHECK PASS

The following wc line should report 5. The SHA-256 value covers the exact JSONL bytes, so a reviewer can detect a changed fixture or decision.

Step 3: Read the failures, not only the pass

Q1 demonstrates a freshness failure. The raw ranking places the expired Acme payroll policy first, then a Beta tenant policy and an Acme draft. The gate rejects all three and selects the current approved Acme version.

Q2 demonstrates tenant isolation. The first result belongs to Acme, so the Beta caller receives the next allowed Beta document.

Q3 demonstrates authorization-aware abstention. The employee can retrieve a semantically relevant SRE runbook from the simulated search result, but cannot use it. The model should receive no passage and the application should return a bounded message such as, “I cannot find an approved source you can access.”

Q4 proves that the same runbook is available to an SRE principal. Q5 proves that an empty candidate list is a normal abstention, not an exception or an invitation to search without filters.

The three schema failures matter because a permissive contract converts missing metadata into accidental access. Reject incomplete records during ingestion. Do not reinterpret missing tenant_id as public or a timestamp without a timezone as local time.

Step 4: Put the gate before prompt construction

The integration boundary is a ranked list of document IDs plus trusted caller context:

{
  "query_id": "Q1",
  "tenant_id": "acme",
  "principals": ["role:employee"],
  "retrieved_at": "2026-10-18T11:00:00Z",
  "candidate_ids": ["acme-payroll-v2", "beta-payroll-v4", "acme-payroll-draft", "acme-payroll-v3"]
}

The output boundary is one retrieval_decisions.jsonl row containing the selected document, rejection counts and citation URI. Send only the selected document text and citation to the model. Keep rejected text out of the prompt, trace and fallback path.

This order is deliberate. Post-generation redaction is too late because the model has already processed the unauthorized passage. Prompt instructions such as “ignore documents from other tenants” are not authorization controls. Enforce policy in deterministic application code or in the retrieval engine before generation, then recheck returned records at the application boundary.

Step 5: Map the contract to a provider

For Amazon Bedrock Knowledge Bases, store tenant_id, status, effective_epoch, expires_epoch and allowed_principals as filterable metadata. Its documented RetrievalFilter type supports equals, numeric comparisons, listContains, andAll and orAll. Build the filter from verified identity claims. For callers with several principals, create an orAll group of listContains expressions inside the outer andAll. Check the current limits and vector-store support because some operators vary by store. Amazon Bedrock filtering documentation.

For Azure AI Search, make the principal collection filterable and apply document-level access control or the documented security-trimming pattern on every query. Microsoft explicitly notes that setting a principal field to non-retrievable is not itself a security boundary. Azure AI Search security-filter pattern.

Provider filtering reduces exposure and wasted retrieval. The local gate remains useful as defense in depth, for portable evaluation, and for detecting misconfigured provider queries. Keep the same fixture in CI when changing index schemas, identity providers, SDKs or retrieval backends.

Observability without copying document text

Log the query ID, authenticated tenant, a stable policy version, candidate document IDs, rejection reasons, selected document ID, citation URI and decision latency. Avoid raw passages, questions, access tokens and complete principal lists in general-purpose logs. If principal-level debugging is necessary, protect that trace as sensitive data and define retention separately.

Track these operating measures:

• allowed selection rate and abstention rate;
• rejection counts by reason;
• missing-document and hash-mismatch counts;
• decision latency at median and tail percentiles;
• percentage of answers with an approved citation;
• evaluation pass rate on known cross-tenant, stale, draft and revoked cases.

A sudden rise in EXPIRED may mean the corpus refresh failed. A rise in MISSING_DOCUMENT may mean retrieval and metadata stores are out of sync. A drop in abstentions is not automatically good. It may indicate broader access rather than better retrieval.

For cost accounting, join query_id to model and retrieval usage after the decision. The earlier cost-per-useful-request worksheet explains the denominator, and the retry-cost audit shows how to avoid counting one task twice.

Failure cases and troubleshooting

Rollback and manual fallback

If the gate causes unexpected abstentions, do not bypass it. Roll back the new corpus or policy mapping to the last reviewed version, keep the gate active, and route unanswered questions to a human who already has access to the source system. Record the query ID and reason without copying restricted text into the ticket.

If metadata integrity is uncertain, pause generation from that corpus. A read-only link to the authoritative document system is a safer fallback than generating from evidence whose ownership or validity cannot be established.

Completion check

You are finished when all of these are true:

1. The process exits with code 0 and prints CHECK PASS.
2. Complete stdout matches the block above.
3. retrieval_decisions.jsonl has five lines and the printed digest matches.
4. The evaluation reports five correct decisions and zero unsafe selections.
5. All three negative cases are rejected.
6. You can point to the function that obtains authenticated tenant and principals in your real adapter.
7. Your generation path accepts only the gate's selected document, never the raw candidate list.

Independent exercise

Add a sixth document whose effective_at is one second after the query time. Put it first in a new query and expect NOT_YET_EFFECTIVE. Then advance retrieved_at to the exact effective instant and expect selection, provided no other current version overlaps. Add the case without changing the gate code.

For a deeper exercise, generate both an Amazon Bedrock filter object and an Azure OData filter from the same query contract. Snapshot-test those expressions, but continue to test returned documents through document_reason. Provider syntax tests and policy outcome tests answer different questions.

What was measured and what was not

Measured in this lab: Python 3.12.14 completed five synthetic cases, rejected three malformed or conflicting states, wrote five JSONL rows, produced the stated digest, selected zero unsafe documents, and made no network calls.

Documented by primary sources: the 2026 Stack Overflow survey findings, the joint security guidance, and the provider filter capabilities linked above.

Synthetic: all tenants, documents, questions, decisions, review minutes and labor arithmetic.

Not established: production latency, false-rejection rate, incident reduction, customer demand, search volume, cloud spend, or business savings.

The next engineering step is to run this fixture against a staging retriever, capture provider-returned candidate metadata, and add representative revoked, future-effective, group-change and identity-mapping cases before any production rollout.