Scope boundary

Owns structural partitioning and retrieval authorization for persistent memory, vector stores, and RAG namespaces by tenant and trust tier. It does not define whether a candidate write is eligible for a trust tier or govern volatile active context.

AID-I-004.001: AID-I-004.002 isolates persistent long-term memory namespaces; AID-I-004.001 isolates volatile session and tenant working context. Persistent partitioning and active-context hygiene have different stores, lifetimes, evidence populations, and failure modes.

AID-M-002.004: AID-I-004.002 enforces persistent namespace and retrieval isolation; AID-M-002.004 defines the provenance, write-eligibility, trust-tier, evidence, and namespace metadata contract. The Model control publishes which state and namespace a write should receive; this control makes that partition and retrieval boundary effective at runtime.

Implementation guidance

• AID-I-004.002-G001: Partition long-term memory by tenant + trust tier; retrieval must consult a central entitlement/policy service and be fully audited.

AID-I-004.002-G001

Security boundary

The retrieval gate must obtain scope and lease state from trusted authorities. A process wall clock is not authorization evidence. Bind the authorization to a signed lease ID/sequence, use authority-provided time, and revalidate immediately before and after every namespace query and again before return. Discard every accumulated hit if the lease expires, authority becomes unavailable, a query times out, or any partition/result binding differs.

Use a bounded lease-aware retrieval gate

# File: memory/retrieval_gate.py
from __future__ import annotations

import hashlib
import json
import math
from dataclasses import dataclass
from typing import Literal, Protocol


@dataclass(frozen=True)
class AuthenticatedPrincipal:
    principal_id: str
    tenant_id: str
    authentication_event_id: str


@dataclass(frozen=True)
class VerifiedRetrievalScope:
    decision_id: str
    principal_id: str
    tenant_id: str
    policy_version: str
    policy_sha256: str
    query_sha256: str
    collection: str
    namespaces: tuple[str, ...]
    result_limit: int
    lease_id: str
    lease_sequence: int
    expires_at_authority_unix: float
    per_query_timeout_seconds: float
    total_timeout_seconds: float
    receipt_sha256: str


@dataclass(frozen=True)
class VerifiedLeaseCheck:
    state: Literal["ACTIVE", "EXPIRED", "REVOKED"]
    lease_id: str
    lease_sequence: int
    authority_time_unix: float
    expires_at_authority_unix: float
    receipt_sha256: str


@dataclass(frozen=True)
class SearchHit:
    hit_id: str
    tenant_id: str
    namespace: str
    score: float
    payload: dict


@dataclass(frozen=True)
class RetrievalResult:
    control_status: Literal["PASS", "FAIL", "INSUFFICIENT_DATA", "ERROR"]
    finding_status: Literal["NO_FINDING", "FINDING", "UNKNOWN"]
    hits: tuple[SearchHit, ...]
    decision_id: str
    final_lease_receipt_sha256: str | None
    audit_receipt_id: str | None
    detail: str


class EntitlementAuthorizer(Protocol):
    def authorize_retrieval(
        self, *, principal: AuthenticatedPrincipal, query_sha256: str,
    ) -> VerifiedRetrievalScope: ...


class RetrievalLeaseAuthority(Protocol):
    def revalidate(
        self, *, scope: VerifiedRetrievalScope, stage: str,
    ) -> VerifiedLeaseCheck: ...


class PartitionedVectorStore(Protocol):
    def search(
        self,
        *,
        collection: str,
        namespace: str,
        tenant_filter: str,
        query_vector: tuple[float, ...],
        limit: int,
        timeout_seconds: float,
    ) -> list[SearchHit]: ...


class AppendOnlyRetrievalAudit(Protocol):
    def append(self, event: dict) -> str: ...
    def read(self, receipt_id: str) -> dict: ...


class RetrievalHalt(Exception):
    def __init__(self, control_status: str, finding_status: str, detail: str):
        super().__init__(detail)
        self.control_status = control_status
        self.finding_status = finding_status
        self.detail = detail


def canonical_digest(value) -> str:
    raw = json.dumps(
        value, sort_keys=True, separators=(",", ":"), ensure_ascii=False,
        allow_nan=False,
    ).encode("utf-8")
    return hashlib.sha256(raw).hexdigest()


def secure_vector_search(
    *,
    vector_store: PartitionedVectorStore,
    authorizer: EntitlementAuthorizer,
    lease_authority: RetrievalLeaseAuthority,
    audit: AppendOnlyRetrievalAudit,
    principal: AuthenticatedPrincipal,
    query_vector: list[float],
) -> RetrievalResult:
    if not all((
        principal.principal_id, principal.tenant_id,
        principal.authentication_event_id,
    )):
        raise PermissionError("authenticated principal is incomplete")
    if (
        not isinstance(query_vector, list)
        or not query_vector
        or any(
            isinstance(value, bool)
            or not isinstance(value, (int, float))
            or not math.isfinite(value)
            for value in query_vector
        )
    ):
        raise ValueError("query vector is empty or non-finite")

    vector = tuple(float(value) for value in query_vector)
    query_sha256 = canonical_digest(list(vector))
    scope = authorizer.authorize_retrieval(
        principal=principal, query_sha256=query_sha256,
    )
    if (
        scope.principal_id != principal.principal_id
        or scope.tenant_id != principal.tenant_id
        or scope.query_sha256 != query_sha256
        or not all((
            scope.decision_id, scope.policy_version, scope.collection,
            scope.lease_id,
        ))
        or len(scope.policy_sha256) != 64
        or len(scope.receipt_sha256) != 64
        or not scope.namespaces
        or tuple(sorted(set(scope.namespaces))) != scope.namespaces
        or any(not namespace for namespace in scope.namespaces)
        or isinstance(scope.result_limit, bool)
        or not isinstance(scope.result_limit, int)
        or scope.result_limit < 1
        or isinstance(scope.lease_sequence, bool)
        or scope.lease_sequence < 1
        or any(
            isinstance(value, bool)
            or not isinstance(value, (int, float))
            or not math.isfinite(value)
            or value <= 0
            for value in (
                scope.expires_at_authority_unix,
                scope.per_query_timeout_seconds,
                scope.total_timeout_seconds,
            )
        )
    ):
        raise PermissionError("verified retrieval scope binding is invalid")

    last_check: VerifiedLeaseCheck | None = None

    def nonpass(
        status: str, finding: str, detail: str,
    ) -> RetrievalResult:
        return RetrievalResult(
            control_status=status,
            finding_status=finding,
            hits=(),
            decision_id=scope.decision_id,
            final_lease_receipt_sha256=(
                None if last_check is None else last_check.receipt_sha256
            ),
            audit_receipt_id=None,
            detail=detail,
        )

    def revalidate(stage: str) -> VerifiedLeaseCheck:
        nonlocal last_check
        try:
            check = lease_authority.revalidate(scope=scope, stage=stage)
        except Exception as exc:
            raise RetrievalHalt(
                "ERROR", "UNKNOWN", f"lease authority unavailable at {stage}",
            ) from exc
        if (
            check.lease_id != scope.lease_id
            or check.lease_sequence != scope.lease_sequence
            or check.expires_at_authority_unix != scope.expires_at_authority_unix
            or len(check.receipt_sha256) != 64
            or not math.isfinite(check.authority_time_unix)
            or not math.isfinite(check.expires_at_authority_unix)
        ):
            raise RetrievalHalt(
                "FAIL", "FINDING", f"lease binding differs at {stage}",
            )
        last_check = check
        if (
            check.state != "ACTIVE"
            or check.authority_time_unix >= check.expires_at_authority_unix
        ):
            raise RetrievalHalt(
                "FAIL", "FINDING", f"lease is not active at {stage}",
            )
        return check

    try:
        first_check = revalidate("after_authorization")
        total_deadline = min(
            first_check.expires_at_authority_unix,
            first_check.authority_time_unix + scope.total_timeout_seconds,
        )
        hits: list[SearchHit] = []
        seen_ids: set[str] = set()
        query_receipts: list[str] = []
        for namespace in scope.namespaces:
            remaining = scope.result_limit - len(hits)
            if remaining == 0:
                break
            before = revalidate(f"before_query:{namespace}")
            remaining_window = min(
                before.expires_at_authority_unix,
                total_deadline,
            ) - before.authority_time_unix
            query_timeout = min(
                scope.per_query_timeout_seconds, remaining_window,
            )
            if not math.isfinite(query_timeout) or query_timeout <= 0:
                raise RetrievalHalt(
                    "FAIL", "FINDING",
                    f"lease or total timeout exhausted before {namespace}",
                )
            try:
                page = vector_store.search(
                    collection=scope.collection,
                    namespace=namespace,
                    tenant_filter=scope.tenant_id,
                    query_vector=vector,
                    limit=remaining,
                    timeout_seconds=query_timeout,
                )
            except TimeoutError as exc:
                raise RetrievalHalt(
                    "FAIL", "FINDING", f"namespace query timed out: {namespace}",
                ) from exc
            except Exception as exc:
                raise RetrievalHalt(
                    "ERROR", "UNKNOWN", f"namespace query failed: {namespace}",
                ) from exc
            after = revalidate(f"after_query:{namespace}")
            if after.authority_time_unix >= total_deadline:
                raise RetrievalHalt(
                    "FAIL", "FINDING",
                    f"total timeout exhausted after {namespace}",
                )
            query_receipts.extend((before.receipt_sha256, after.receipt_sha256))
            if not isinstance(page, list) or len(page) > remaining:
                raise RetrievalHalt(
                    "FAIL", "FINDING",
                    "vector store exceeded the global result budget",
                )
            for hit in page:
                if (
                    not isinstance(hit, SearchHit)
                    or not hit.hit_id
                    or hit.hit_id in seen_ids
                    or hit.tenant_id != scope.tenant_id
                    or hit.namespace != namespace
                    or not math.isfinite(hit.score)
                ):
                    raise RetrievalHalt(
                        "FAIL", "FINDING",
                        "vector-store result escaped the verified partition",
                    )
                seen_ids.add(hit.hit_id)
                hits.append(hit)

        before_audit = revalidate("before_success_audit")
        event = {
            "schema_version": "aidefend.partitioned-retrieval-candidate.v2",
            "decision_id": scope.decision_id,
            "authorization_receipt_sha256": scope.receipt_sha256,
            "authentication_event_id": principal.authentication_event_id,
            "principal_id": principal.principal_id,
            "tenant_id": scope.tenant_id,
            "policy_version": scope.policy_version,
            "policy_sha256": scope.policy_sha256,
            "query_sha256": query_sha256,
            "collection": scope.collection,
            "namespaces": list(scope.namespaces),
            "global_result_limit": scope.result_limit,
            "lease_id": scope.lease_id,
            "lease_sequence": scope.lease_sequence,
            "lease_expires_at_authority_unix": scope.expires_at_authority_unix,
            "query_lease_receipt_sha256s": query_receipts,
            "pre_audit_lease_receipt_sha256": before_audit.receipt_sha256,
            "returned_hit_ids": [hit.hit_id for hit in hits],
            "returned_hit_count": len(hits),
            "outcome": "AUTHORIZED_RESULT_CANDIDATE",
        }
        try:
            audit_receipt_id = audit.append(event)
            if audit.read(audit_receipt_id) != event:
                raise RuntimeError("retrieval audit readback differs")
        except Exception as exc:
            raise RetrievalHalt(
                "ERROR", "UNKNOWN", "retrieval audit unavailable or differs",
            ) from exc
        final_check = revalidate("before_return")
        if final_check.authority_time_unix >= total_deadline:
            raise RetrievalHalt(
                "FAIL", "FINDING", "total timeout exhausted before return",
            )
        return RetrievalResult(
            control_status="PASS",
            finding_status="NO_FINDING",
            hits=tuple(hits),
            decision_id=scope.decision_id,
            final_lease_receipt_sha256=final_check.receipt_sha256,
            audit_receipt_id=audit_receipt_id,
            detail="authorized partitions returned within the active lease",
        )
    except RetrievalHalt as halt:
        return nonpass(halt.control_status, halt.finding_status, halt.detail)

Executable regression fixtures

# File: tests/test_retrieval_gate.py
from types import SimpleNamespace

from memory.retrieval_gate import (
    AuthenticatedPrincipal,
    SearchHit,
    VerifiedLeaseCheck,
    VerifiedRetrievalScope,
    canonical_digest,
    secure_vector_search,
)


class LeaseAuthority:
    def __init__(self, *, expire_at=None, outage_at=None):
        self.calls = []
        self.expire_at = expire_at
        self.outage_at = outage_at

    def revalidate(self, *, scope, stage):
        self.calls.append(stage)
        if stage == self.outage_at:
            raise ConnectionError("authority outage")
        now = 100.0 + len(self.calls)
        state = "EXPIRED" if stage == self.expire_at else "ACTIVE"
        return VerifiedLeaseCheck(
            state, scope.lease_id, scope.lease_sequence, now,
            scope.expires_at_authority_unix, "l" * 64,
        )


class Store:
    def __init__(self):
        self.calls = []

    def search(self, **values):
        self.calls.append(values)
        namespace = values["namespace"]
        return [SearchHit(
            f"hit-{namespace}", "tenant-a", namespace, 0.9, {"safe": True},
        )]


class Audit:
    def __init__(self):
        self.values = {}

    def append(self, event):
        receipt_id = canonical_digest(event)
        self.values[receipt_id] = event
        return receipt_id

    def read(self, receipt_id):
        return self.values[receipt_id]


principal = AuthenticatedPrincipal("principal-a", "tenant-a", "auth-1")
vector = [0.25, 0.75]
scope = VerifiedRetrievalScope(
    "decision-1", "principal-a", "tenant-a", "policy-7", "p" * 64,
    canonical_digest(vector), "memory", ("ns-a", "ns-b"), 2,
    "lease-1", 11, 200.0, 2.0, 10.0, "r" * 64,
)


def run(authority):
    store = Store()
    result = secure_vector_search(
        vector_store=store,
        authorizer=SimpleNamespace(authorize_retrieval=lambda **_: scope),
        lease_authority=authority,
        audit=Audit(),
        principal=principal,
        query_vector=vector,
    )
    return result, store


def test_expiry_after_query_discards_accumulated_hits():
    result, store = run(LeaseAuthority(expire_at="after_query:ns-a"))
    assert store.calls
    assert result.control_status == "FAIL"
    assert result.hits == ()


def test_authority_outage_before_second_namespace_discards_first_page():
    result, _ = run(LeaseAuthority(outage_at="before_query:ns-b"))
    assert (result.control_status, result.finding_status) == ("ERROR", "UNKNOWN")
    assert result.hits == ()


def test_success_revalidates_every_boundary_and_bounds_timeouts():
    authority = LeaseAuthority()
    result, store = run(authority)
    assert (result.control_status, len(result.hits)) == ("PASS", 2)
    assert authority.calls == [
        "after_authorization",
        "before_query:ns-a", "after_query:ns-a",
        "before_query:ns-b", "after_query:ns-b",
        "before_success_audit", "before_return",
    ]
    assert all(0 < call["timeout_seconds"] <= 2.0 for call in store.calls)

Action: The vector adapter must enforce the passed timeout server-side and the authority adapter must verify its own signatures, signer authorization, revocation state, and monotonic lease sequence. Treat expiry, revocation, timeout, authority/audit outage, and stale or mismatched receipts as non-PASS; never return partial or previously accumulated hits on those paths.

Open-source tools

• Qdrant
• Weaviate
• Milvus
• pgvector with PostgreSQL row-level security (tenant and trust partition enforcement)
• Open Policy Agent (OPA)

Commercial / hosted tools

• Pinecone
• Zilliz Cloud
• Qdrant Cloud

Threat mappings

MITRE ATLAS

• AML.T0070 RAG Poisoning
• AML.T0080.000 AI Agent Context Poisoning: Memory
• AML.T0118 Autonomous AI Agent Communication
• AML.T0118.000 Autonomous AI Agent Communication: Communication via Shared Artifacts (partitioned memory namespaces with central retrieval authorization deny cross-agent dead-drop coordination)

MAESTRO

• Compromised RAG Pipelines (L2)
• Data Poisoning (L2)
• Data Leakage (Cross-Layer) (tenant isolation prevents cross-tenant data exposure)

OWASP LLM Top 10 2026

• LLM01:2026 Prompt Injection (trust-tier partitions keep untrusted persistent content outside trusted context)
• LLM02:2026 Sensitive Information Disclosure
• LLM05:2026 Data and Model Poisoning (trust-tier isolation limits poisoned RAG or memory content from trusted retrieval)
• LLM09:2026 Vector and Embedding Weaknesses

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI06:2026 Memory & Context Poisoning
• ASI03:2026 Identity and Privilege Abuse (authorization-gated retrieval prevents unauthorized access)

NIST Adversarial Machine Learning 2025

• NISTAML.015 Indirect Prompt Injection (partitioning limits scope of poisoned content retrieval)
• NISTAML.036 Leaking information from user interactions (tenant partitioning prevents cross-tenant data leakage)

Cisco Integrated AI Security and Safety Framework

• AISubtech-14.1.2 Insufficient Access Controls
• AISubtech-5.1.1 Long-term / Short-term Memory Injection (authorized trust and tenant partitions limit poisoned-memory retrieval across boundaries)
• AISubtech-6.1.1 Knowledge Base Poisoning (authorized trust and tenant partitions limit poisoned-memory retrieval across boundaries)
• AITech-4.2 Context Boundary Attacks (authorized trust and tenant partitions limit poisoned-memory retrieval across boundaries)
• AITech-5.1 Memory System Persistence (authorized trust and tenant partitions limit poisoned-memory retrieval across boundaries)
• AITech-7.2 Memory System Corruption (authorized trust and tenant partitions limit poisoned-memory retrieval across boundaries)
• AITech-8.2 Data Exfiltration / Exposure (retrieval leases and result binding restrict cross-partition data disclosure)

Google Secure AI Framework 2.0 - Risks

• SDD: Sensitive Data Disclosure (tenant partitioning prevents cross-tenant data exposure)
• PIJ: Prompt Injection (partitioning limits scope of poisoned content retrieval)

Databricks AI Security Framework 3.0

• Agents - Core 13.1: Memory Poisoning (partitioning limits memory poisoning to single tenant)
• Agents - Core 13.3: Privilege Compromise (authorization-gated retrieval prevents unauthorized access)
• Agents - Tools MCP Server 13.24: Context Spoofing and Manipulation (partitioning prevents cross-tenant context manipulation)
• Raw Data 1.1: Insufficient access controls (policy-gated retrieval enforces access controls on memory)