Test retrieval freshness and permissions

# Retrieval freshness and access regression lab

An offline, synthetic experiment comparing rank-only, top-1 then filter, and eligibility-first retrieval. It extends a metadata predicate into five test cases and three measured strategies.

## Run

Requires Python 3.10+; standard library only. No API keys, network requests or installations.

```bash
cd engineering-labs/retrieval-freshness
PYTHONDONTWRITEBYTECODE=1 python3 -m unittest -v
python3 lab.py --output my-results.json
python3 -m json.tool my-results.json
python3 lab.py --help
```

- `fixtures.json`: six synthetic documents, authoritative revision map, five queries with explicit expected IDs.
- `lab.py`: lexical Jaccard scoring, eligibility predicate, three retrieval strategies and exact counts.
- `results.json`: actual reference run; fixture/code SHA-256 hashes.
- `test_lab.py`: eight tests, including revocation, metadata omissions, validity boundary and candidate starvation.
- `article.md`: reader-facing explanation and adaptation steps.
- `SOURCES.md`: primary-source research notes.

Passing results: rank-only 1/5 correct with 4 wrong selections; filter-after-top1 3/5 with 2 false abstentions; filter-before-ranking 5/5 with 3 correct abstentions. These are hand-designed regression cases, not population accuracy estimates.

## Change the fixture

Copy `fixtures.json`, add sanitised documents/cases and execute:

```bash
python3 lab.py --fixture your-fixture.json --output your-results.json
```

Expected IDs come from an independent content/permission review. Dates are ISO calendar dates. The active revision map uses tenant/logical-ID keys; groups are an explicit allowlist, with any shared group permitting access. No implicit public access. The identity fields are trusted test inputs, not an authentication API.

## Limitations

No embeddings, model generation, database integration, timing benchmark or security certification. Lexical score threshold 0.25 is solely illustrative. This scans the entire fixture, does not reproduce ANN filtering, and does not cover cache invalidation, index lag, concurrent permission updates, denies overriding allows, nested groups, or historical revision queries. Adapt the policy before applying it to a real service.

# Research sources

Checked 8 September 2026. Primary vendor documentation supports the mechanisms; all numeric results come from this repository's synthetic run.

1. [Microsoft: security filters for trimming results](https://learn.microsoft.com/en-us/azure/search/search-security-trimming-for-azure-search)
   - Read the distinction between filtering by a principal string and authenticating a caller.
   - Supports the article's warning that metadata matching alone is not authentication.
   - No Azure latency claims or access-control certification inferred.
2. [Microsoft: vector query filters](https://learn.microsoft.com/en-us/azure/search/vector-search-filters)
   - Read the filtering-mode descriptions and small-k/selective-filter false-negative discussion.
   - Supports the candidate-starvation explanation.
   - Our exhaustive lexical scan does not reproduce Azure's sharding or HNSW implementation. No vendor performance claims are measured here.
3. [Qdrant: filtering](https://qdrant.tech/documentation/search/filtering/)
   - Supports availability of payload conditions and logical combinations as an implementation mechanism.
   - No Qdrant instance was deployed or tested.
4. [Swarm Signal: knowledge graphs for AI agents](https://swarmsignal.net/knowledge-graphs-for-ai-agents/)
   - Background internal link for validity and permission concepts, not an external research authority.

## Evidence provenance

The six-document corpus and five queries are authored synthetic fixtures, not sampled production traffic. Jaccard scores are computed from text, not manually supplied. Results are generated by `lab.py`; hashes in `results.json` bind the code and fixture bytes. Unit tests exercise all expected retrieval outcomes plus individual eligibility boundaries. No external inference or paid calls were made.

# Test retrieval freshness and permissions

A retrieved passage can match a question perfectly and still be unusable. It might belong to another tenant, describe a superseded policy, or contain information the caller cannot access. Asking the model to ignore it leaves the wrong material in the context.

This lab gives you a small regression harness for that boundary. It compares ranking alone, filtering after the first result, and ranking eligible documents. You can run it without an API key or a vector database, inspect every decision, then replace the synthetic corpus with a sanitised slice of your own.

**The result is a worked failure demonstration, not a retrieval benchmark.** The five cases deliberately expose specific mistakes. We use token overlap, not embeddings, and make no claim about how often these failures occur in production.

## What you will test

The fixture contains six short documents and five queries. Two documents represent different revisions of a retry policy. Two describe signing-key procedures with different group permissions. One policy is not effective until October. A final document belongs to another tenant.

The texts are intentionally simple. An exact match scores 1.0. The eligible replacements contain the same four query terms plus two extra words, scoring 0.666667. This makes the ranking problem visible without hiding it behind an embedding model.

The score is Jaccard similarity: shared unique tokens divided by all unique tokens. Capitalisation is ignored. Synonyms do not match. A score below 0.25 produces no result; that threshold is a fixture choice, not a recommended production setting.

Each document carries a tenant, logical document ID, revision, status, permitted groups and effective date. The separate active-revision map is authoritative for this experiment. Being the newest item returned by search is insufficient: an older revision remains invalid even when the current one is missing from the index.

This extends the validity and permission example in our [knowledge graph guide](/knowledge-graphs-for-ai-agents/) into an executable retrieval regression suite.

## Run the experiment

Download the lab ZIP below and extract it. From the extracted folder:

```bash
cd engineering-labs/retrieval-freshness
python3 -m unittest -v
python3 lab.py --output my-results.json
python3 -m json.tool my-results.json
```

Python 3.10 or later is sufficient. The code uses the standard library and makes no network calls. The included `results.json` records the run used here, with SHA-256 hashes for the fixture and code.

There are three strategies:

- **Rank only:** take the best lexical match if it clears the threshold.
- **Filter after top-1:** select one candidate, then reject it if ineligible.
- **Filter before ranking:** consider all eligible documents, then select the best match.

These are deliberately small algorithms. The last strategy scans the entire fixture. It is a reference for expected behaviour, not an implementation of approximate nearest-neighbour search.

## The actual results

The Fedora run on 8 September 2026 produced:

| Case | Rank only | Filter after top-1 | Filter before ranking |
|---|---|---|---|
| Superseded revision | retry-v1 | Abstain | retry-v2 |
| Restricted document | keys-private-v1 | Abstain | keys-public-v1 |
| Caller has no access | retry-v1 | Abstain | Abstain |
| Policy not yet effective | refund-v2 | Abstain | Abstain |
| No lexical match | Abstain | Abstain | Abstain |

Ranking alone made four wrong selections. Filtering after top-1 prevented those selections but lost two valid answers. Filtering before ranking matched all five expected outcomes, including three required abstentions.

Do not turn five deliberately constructed successes into a general accuracy percentage. The useful result is the pair of false abstentions: rejecting the highest-ranked document does not automatically recover the eligible document underneath it.

Microsoft documents a related risk in Azure AI Search: filtering after candidate selection can miss matching documents, particularly with selective filters and small candidate counts. Its implementation includes shard and graph traversal behaviour that this script does not reproduce. [Azure vector query filters](https://learn.microsoft.com/en-us/azure/search/vector-search-filters)

## Decide eligibility before building model context

Our predicate requires the correct tenant, an active status, the authoritative revision, an allowed group and an effective date range. Start dates are inclusive; expiry dates are exclusive. Missing required metadata, empty permissions and an unknown active revision fail closed.

The eight tests cover those boundaries, malformed permission metadata, the missing-result problem and a permission change between queries. One test revokes the engineering group's access to the selected revision and verifies that the next retrieval abstains.

The supplied identities are trusted fixture inputs. This does not authenticate a user. In a service, derive identity and group membership from your trusted authentication layer; do not accept an arbitrary group list submitted with a question.

That distinction matters in real products. Microsoft's security-filter pattern explicitly describes string matching rather than authentication through the principal string. A working filter still needs a trustworthy identity source and consistent application to queries. [Azure security filtering](https://learn.microsoft.com/en-us/azure/search/search-security-trimming-for-azure-search)

Metadata filters are available in retrieval systems such as Qdrant, including combinations of required and excluded conditions. That is a mechanism you can map the predicate onto, not evidence that a particular deployment enforces your policy correctly. [Qdrant filtering documentation](https://qdrant.tech/documentation/search/filtering/)

## Adapt it to your own corpus

Start with a small set of real failure cases, scrubbed of private content. Give each query an expected permitted document ID or an explicit no-answer outcome. Include an old revision that matches unusually well and a valid replacement with different wording.

Keep the fixture oracle independent of your search response. If expected answers are copied from whatever the retriever returned, the test cannot detect a regression.

Replace the lexical ranking function with an adapter around your existing retriever. Record the candidate count, selected IDs and document revisions. Compare the actual result against the same expected IDs. Keep sensitive rejected snippets out of ordinary logs.

Then test transitions: publish a replacement, remove a document, revoke access and change the active revision. Measure how long stale state remains observable. The current lab uses an immediate in-memory update; it does not test indexing lag, caches, replicas or concurrent requests.

Check permission handling again before any external reranker or model receives text. Apply the same rules to cached answers and follow-up retrieval calls. A successful first query says nothing about an unfiltered second path.

## What passing means

Passing means these cases still behave as specified after a code or index change. It does not prove semantic relevance, answer correctness, adversarial resistance or production access control. There is no generated answer to evaluate here.

Extend the corpus until it represents your actual document lifecycle. Track wrong-document selections separately from false abstentions. Both hurt usefulness, but a permission failure requires a different response from an overly narrow candidate set.


For the broader choice of where knowledge should live, see [RAG, fine-tuning or long context](/rag-vs-fine-tuning-vs-long-context-2026/). This lab tests one retrieval boundary after that architectural choice.

## Sources

- [Microsoft: security filters for trimming search results](https://learn.microsoft.com/en-us/azure/search/search-security-trimming-for-azure-search)
- [Microsoft: vector query filtering modes](https://learn.microsoft.com/en-us/azure/search/vector-search-filters)
- [Qdrant: payload filtering](https://qdrant.tech/documentation/search/filtering/)

Primary documentation checked on 8 September 2026. The downloadable fixture, code, tests and results contain the experiment evidence.

{
  "description": "Deliberately synthetic lexical retrieval regression fixture; not a model benchmark or real policy advice.",
  "active_revisions": {
    "acme/retry-policy": 2,
    "acme/private-key-policy": 1,
    "acme/public-key-policy": 1,
    "acme/refund-policy": 2,
    "beta/retry-policy": 1
  },
  "documents": [
    {
      "id": "retry-v1",
      "tenant": "acme",
      "logical_id": "retry-policy",
      "revision": 1,
      "status": "superseded",
      "groups": [
        "engineering"
      ],
      "valid_from": "2026-01-01",
      "text": "retry timeout payment worker"
    },
    {
      "id": "retry-v2",
      "tenant": "acme",
      "logical_id": "retry-policy",
      "revision": 2,
      "status": "active",
      "groups": [
        "engineering"
      ],
      "valid_from": "2026-08-01",
      "text": "payment worker retry timeout backoff budget"
    },
    {
      "id": "keys-private-v1",
      "tenant": "acme",
      "logical_id": "private-key-policy",
      "revision": 1,
      "status": "active",
      "groups": [
        "security"
      ],
      "valid_from": "2026-01-01",
      "text": "rotation signing key service"
    },
    {
      "id": "keys-public-v1",
      "tenant": "acme",
      "logical_id": "public-key-policy",
      "revision": 1,
      "status": "active",
      "groups": [
        "engineering"
      ],
      "valid_from": "2026-01-01",
      "text": "service signing key rotation request approval"
    },
    {
      "id": "refund-v2",
      "tenant": "acme",
      "logical_id": "refund-policy",
      "revision": 2,
      "status": "active",
      "groups": [
        "engineering"
      ],
      "valid_from": "2026-10-01",
      "text": "refund account recovery"
    },
    {
      "id": "other-tenant-v1",
      "tenant": "beta",
      "logical_id": "retry-policy",
      "revision": 1,
      "status": "active",
      "groups": [
        "engineering"
      ],
      "valid_from": "2026-01-01",
      "text": "retry timeout payment worker"
    }
  ],
  "cases": [
    {
      "id": "superseded",
      "query": "retry timeout payment worker",
      "tenant": "acme",
      "groups": [
        "engineering"
      ],
      "as_of": "2026-09-08",
      "expected": "retry-v2"
    },
    {
      "id": "restricted",
      "query": "rotation signing key service",
      "tenant": "acme",
      "groups": [
        "engineering"
      ],
      "as_of": "2026-09-08",
      "expected": "keys-public-v1"
    },
    {
      "id": "no-access",
      "query": "retry timeout payment worker",
      "tenant": "acme",
      "groups": [
        "finance"
      ],
      "as_of": "2026-09-08",
      "expected": null
    },
    {
      "id": "not-yet-effective",
      "query": "refund account recovery",
      "tenant": "acme",
      "groups": [
        "engineering"
      ],
      "as_of": "2026-09-08",
      "expected": null
    },
    {
      "id": "no-match",
      "query": "orbital telescope calibration",
      "tenant": "acme",
      "groups": [
        "engineering"
      ],
      "as_of": "2026-09-08",
      "expected": null
    }
  ]
}

#!/usr/bin/env python3
"""Synthetic retrieval regression experiment. No network, embeddings or model calls."""
import argparse
from datetime import date
import hashlib
import json
from pathlib import Path
import re

HERE = Path(__file__).resolve().parent
MIN_SCORE = 0.25  # Demonstration threshold, not calibrated for a real corpus.


def similarity(query, text):
    left, right = (set(re.findall(r"[a-z0-9]+", value.lower())) for value in (query, text))
    return len(left & right) / len(left | right) if left | right else 0.0


def eligible(doc, case, active_revisions):
    """The fixture supplies trusted metadata and identity; this is not authentication."""
    required = {"id", "tenant", "logical_id", "revision", "status", "groups", "valid_from", "text"}
    if not required.issubset(doc):
        return False
    if not isinstance(doc["groups"], list) or not doc["groups"]:
        return False
    if not all(isinstance(group, str) for group in doc["groups"]):
        return False
    if doc["tenant"] != case["tenant"] or doc["status"] != "active":
        return False
    if not set(doc["groups"]) & set(case["groups"]):
        return False
    key = f'{doc["tenant"]}/{doc["logical_id"]}'
    if key not in active_revisions or doc["revision"] != active_revisions[key]:
        return False
    try:
        now = date.fromisoformat(case["as_of"])
        start = date.fromisoformat(doc["valid_from"])
        end = date.fromisoformat(doc["valid_until"]) if doc.get("valid_until") else None
    except (ValueError, TypeError):
        return False
    return start <= now and (end is None or now < end)


def rank(documents, query):
    # Stable input order breaks ties; not a relevance claim.
    return sorted(((similarity(query, doc["text"]), doc) for doc in documents),
                  key=lambda item: item[0], reverse=True)


def choose(ranked):
    if not ranked or ranked[0][0] < MIN_SCORE:
        return None
    score, doc = ranked[0]
    return {"id": doc["id"], "score": round(score, 6)}


def run(fixture):
    rows = []
    for case in fixture["cases"]:
        documents = fixture["documents"]
        allowed = lambda doc: eligible(doc, case, fixture["active_revisions"])
        ranked = rank(documents, case["query"])
        choices = {
            "rank_only": choose(ranked),
            "filter_after_top1": choose([(score, doc) for score, doc in ranked[:1] if allowed(doc)]),
            "filter_before_rank": choose(rank([doc for doc in documents if allowed(doc)], case["query"]))
        }
        rows.append({"case": case["id"], "expected": case["expected"], **choices})
    summary = {}
    for method in ("rank_only", "filter_after_top1", "filter_before_rank"):
        actual = [row[method]["id"] if row[method] else None for row in rows]
        summary[method] = {
            "cases": len(rows),
            "correct": sum(value == row["expected"] for value, row in zip(actual, rows)),
            "wrong_document": sum(value is not None and value != row["expected"] for value, row in zip(actual, rows)),
            "false_abstention": sum(value is None and row["expected"] is not None for value, row in zip(actual, rows)),
            "correct_abstention": sum(value is None and row["expected"] is None for value, row in zip(actual, rows))
        }
    return {"experiment": "Synthetic lexical retrieval; five designed cases, not a population estimate",
            "min_score": MIN_SCORE, "rows": rows, "summary": summary}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--fixture", type=Path, default=HERE / "fixtures.json")
    parser.add_argument("--output", type=Path, help="Write JSON; otherwise print it")
    args = parser.parse_args()
    raw = args.fixture.read_bytes()
    result = run(json.loads(raw))
    result["fixture_sha256"] = hashlib.sha256(raw).hexdigest()
    result["code_sha256"] = hashlib.sha256(Path(__file__).read_bytes()).hexdigest()
    output = json.dumps(result, indent=2) + "\n"
    if args.output:
        args.output.write_text(output)
    else:
        print(output, end="")


if __name__ == "__main__":
    main()

{
  "experiment": "Synthetic lexical retrieval; five designed cases, not a population estimate",
  "min_score": 0.25,
  "rows": [
    {
      "case": "superseded",
      "expected": "retry-v2",
      "rank_only": {
        "id": "retry-v1",
        "score": 1.0
      },
      "filter_after_top1": null,
      "filter_before_rank": {
        "id": "retry-v2",
        "score": 0.666667
      }
    },
    {
      "case": "restricted",
      "expected": "keys-public-v1",
      "rank_only": {
        "id": "keys-private-v1",
        "score": 1.0
      },
      "filter_after_top1": null,
      "filter_before_rank": {
        "id": "keys-public-v1",
        "score": 0.666667
      }
    },
    {
      "case": "no-access",
      "expected": null,
      "rank_only": {
        "id": "retry-v1",
        "score": 1.0
      },
      "filter_after_top1": null,
      "filter_before_rank": null
    },
    {
      "case": "not-yet-effective",
      "expected": null,
      "rank_only": {
        "id": "refund-v2",
        "score": 1.0
      },
      "filter_after_top1": null,
      "filter_before_rank": null
    },
    {
      "case": "no-match",
      "expected": null,
      "rank_only": null,
      "filter_after_top1": null,
      "filter_before_rank": null
    }
  ],
  "summary": {
    "rank_only": {
      "cases": 5,
      "correct": 1,
      "wrong_document": 4,
      "false_abstention": 0,
      "correct_abstention": 1
    },
    "filter_after_top1": {
      "cases": 5,
      "correct": 3,
      "wrong_document": 0,
      "false_abstention": 2,
      "correct_abstention": 3
    },
    "filter_before_rank": {
      "cases": 5,
      "correct": 5,
      "wrong_document": 0,
      "false_abstention": 0,
      "correct_abstention": 3
    }
  },
  "fixture_sha256": "3a63817e7f17dc5cc19df945fc12215f19d68861606d3d9355f2b897ce3877b7",
  "code_sha256": "ed99703248843c4291f5af4545d66f09bafc994ecce227e424783633bd0d71a8"
}

{
  "date": "2026-09-08",
  "host": "fedora",
  "python": "3.14.3",
  "tests": 8,
  "test_exit": 0,
  "reproduces_results_json": true,
  "hashes_verified": true
}

test_each_eligibility_boundary_rejects (test_lab.RetrievalRegressionTests.test_each_eligibility_boundary_rejects) ... ok
test_effective_date_is_inclusive_and_expiry_exclusive (test_lab.RetrievalRegressionTests.test_effective_date_is_inclusive_and_expiry_exclusive) ... ok
test_expected_documents_and_abstentions (test_lab.RetrievalRegressionTests.test_expected_documents_and_abstentions) ... ok
test_lexical_metric_is_not_semantic (test_lab.RetrievalRegressionTests.test_lexical_metric_is_not_semantic) ... ok
test_malformed_permissions_and_dates_fail_closed (test_lab.RetrievalRegressionTests.test_malformed_permissions_and_dates_fail_closed) ... ok
test_missing_metadata_and_unknown_revision_fail_closed (test_lab.RetrievalRegressionTests.test_missing_metadata_and_unknown_revision_fail_closed) ... ok
test_revocation_between_queries_changes_output (test_lab.RetrievalRegressionTests.test_revocation_between_queries_changes_output) ... ok
test_top1_then_filter_loses_valid_results (test_lab.RetrievalRegressionTests.test_top1_then_filter_loses_valid_results) ... ok

----------------------------------------------------------------------
Ran 8 tests in 0.001s

OK

import copy
import json
import unittest
from lab import HERE, eligible, run, similarity


class RetrievalRegressionTests(unittest.TestCase):
    def setUp(self):
        self.fixture = json.loads((HERE / "fixtures.json").read_text())
        self.doc = self.fixture["documents"][1]
        self.case = self.fixture["cases"][0]
        self.active = self.fixture["active_revisions"]

    def test_expected_documents_and_abstentions(self):
        for row in run(self.fixture)["rows"]:
            with self.subTest(case=row["case"]):
                selected = row["filter_before_rank"]
                self.assertEqual(selected["id"] if selected else None, row["expected"])

    def test_top1_then_filter_loses_valid_results(self):
        rows = run(self.fixture)["rows"][:2]
        for row in rows:
            self.assertIsNone(row["filter_after_top1"])
            self.assertIsNotNone(row["filter_before_rank"])

    def test_each_eligibility_boundary_rejects(self):
        mutations = {"tenant": "beta", "revision": 1, "status": "superseded",
                     "groups": ["security"], "valid_from": "2026-09-09",
                     "valid_until": "2026-09-08"}
        for field, value in mutations.items():
            with self.subTest(field=field):
                self.assertFalse(eligible({**self.doc, field: value}, self.case, self.active))
        self.assertTrue(eligible(self.doc, self.case, self.active))

    def test_missing_metadata_and_unknown_revision_fail_closed(self):
        for field in self.doc:
            with self.subTest(field=field):
                doc = copy.deepcopy(self.doc)
                del doc[field]
                self.assertFalse(eligible(doc, self.case, self.active))
        self.assertFalse(eligible(self.doc, self.case, {}))

    def test_malformed_permissions_and_dates_fail_closed(self):
        for field, value in [("groups", []), ("groups", "engineering"),
                             ("groups", [3]), ("valid_from", "yesterday")]:
            self.assertFalse(eligible({**self.doc, field: value}, self.case, self.active))

    def test_effective_date_is_inclusive_and_expiry_exclusive(self):
        case = {**self.case, "as_of": "2026-08-01"}
        self.assertTrue(eligible(self.doc, case, self.active))
        self.assertFalse(eligible({**self.doc, "valid_until": "2026-08-01"}, case, self.active))

    def test_lexical_metric_is_not_semantic(self):
        self.assertEqual(similarity("retry timeout", "timeout retry"), 1.0)
        self.assertEqual(similarity("car", "automobile"), 0.0)
        self.assertEqual(similarity("", ""), 0.0)

    def test_revocation_between_queries_changes_output(self):
        self.assertEqual(run(self.fixture)["rows"][0]["filter_before_rank"]["id"], "retry-v2")
        self.fixture["documents"][1]["groups"] = ["security"]
        self.assertIsNone(run(self.fixture)["rows"][0]["filter_before_rank"])


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