Scope boundary

Owns vector namespace rollback, delete-by-ID quarantine, anti-reingestion tombstones, clean index rebuild, candidate validation, and alias cutover evidence after retrieval poisoning or malicious content injection. It does not restore authoritative source records, evict those source records, or provide ongoing production retrieval monitoring.

AID-R-002: AID-R-005 restores the derived vector index and serving alias; AID-R-002 restores the authoritative source records and corpora. This control consumes restored source data and lineage findings but separately proves that tainted vectors are excluded and the rebuilt index is live.

AID-E-003.002: AID-R-005 rebuilds and promotes a clean retrieval index; AID-E-003.002 identifies and evicts the exact poisoned source-record or graph population. A vector rollback or rebuild does not prove that malicious source records can no longer be ingested again, so this control consumes the signed eviction manifest and tombstone population.

AID-D-014: AID-R-005 performs incident-time vector-index recovery; AID-D-014 continuously monitors production RAG content, relevance, and retrieval provenance. Detector findings can trigger recovery, but index rollback and rebuild do not replace ongoing monitoring or finding generation.

Implementation guidance

• AID-R-005-G001: Use immutable write-blocked index versions and a single alias pointer for policy-authorized, atomic cutover or rollback with exact runtime readback.
• AID-R-005-G002: Preserve the exact suspicious vector-point population in immutable storage, publish signed anti-reingestion tombstones, and verify production retrieval exclusion before recovery proceeds.
• AID-R-005-G003: Rebuild a new immutable candidate index from an exact signed source inventory, verified local embedding artifact, authenticated build-policy records, and anti-reingestion tombstones before approval.
• AID-R-005-G004: Validate the complete signed candidate-index query population against clean-utility, incident-attack, provenance, and anti-reingestion expectations before alias promotion; ongoing production monitoring remains AID-D-014.
• AID-R-005-G005: Consume authenticated retrieval-risk findings, execute policy-bound quarantine plus alias rollback with idempotent state transitions, and read the exact alias and index identity back from the serving system.

AID-R-005-G001

Install one fail-closed transport helper for all R-005 methods

The alias, quarantine, rebuild, and response clients below use the same small helper so every HTTPS response is redirect-free, decoded-body bounded, strict UTF-8 JSON with duplicate keys and non-finite numbers rejected. Each method still supplies its own signed origins, limits, CA, client identity, and response schema.

# File: recovery/strict_io.py
from __future__ import annotations

import hashlib
import base64
import os
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey
import json
from pathlib import Path
from typing import Any
from urllib.parse import urlsplit

import requests


def reject_constant(value: str) -> None:
    raise ValueError(f"non-finite JSON number is forbidden: {value}")


def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for name, value in pairs:
        if name in result:
            raise ValueError(f"duplicate JSON key: {name}")
        result[name] = value
    return result


def strict_json(raw: bytes) -> dict[str, Any]:
    value = json.loads(raw.decode("utf-8", errors="strict"), object_pairs_hook=unique_object, parse_constant=reject_constant)
    if not isinstance(value, dict):
        raise ValueError("one JSON object is required")
    return value


def canonical(value: Any) -> bytes:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False).encode("utf-8")


def read_file(path: Path, maximum: int) -> bytes:
    if isinstance(maximum, bool) or not isinstance(maximum, int) or maximum < 1:
        raise ValueError("maximum bytes must be a positive integer")
    with path.open("rb") as source:
        raw = source.read(maximum + 1)
    if len(raw) > maximum:
        raise ValueError(f"{path} exceeds the signed size bound")
    return raw


def public_key(name: str):
    key = serialization.load_pem_public_key(read_file(Path(os.environ[name]), 16384))
    if not isinstance(key, Ed25519PublicKey):
        raise ValueError("approval authority requires an Ed25519 public key")
    return key


def signed_packet(payload: dict, key_env: str) -> dict:
    # Output-receipt signing is a separate role from input approval authority.
    key = serialization.load_pem_private_key(
        read_file(Path(os.environ[key_env]), 16384), password=None)
    if not isinstance(key, Ed25519PrivateKey):
        raise ValueError("receipt signing requires an Ed25519 private key")
    return {"payload": payload, "signature_b64": base64.b64encode(
        key.sign(canonical(payload))).decode("ascii")}


def verified_packet(path: Path, maximum: int, schema: str, key_env: str) -> dict:
    envelope = strict_json(read_file(path, maximum))
    if set(envelope) != {"payload", "signature_b64"} or not isinstance(envelope["payload"], dict):
        raise ValueError("signed envelope schema differs")
    signature = base64.b64decode(envelope["signature_b64"], validate=True)
    public_key(key_env).verify(signature, canonical(envelope["payload"]))
    if envelope["payload"].get("schema") != schema:
        raise ValueError(f"expected {schema}")
    return envelope


def verified_envelope(path: Path, maximum: int, schema: str, key_env: str) -> dict:
    return verified_packet(path, maximum, schema, key_env)["payload"]


def exact_origin(value: object, name: str) -> str:
    if not isinstance(value, str):
        raise ValueError(f"{name} origin must be a string")
    parsed = urlsplit(value)
    if parsed.scheme != "https" or not parsed.hostname or parsed.username is not None or parsed.password is not None or parsed.path not in {"", "/"} or parsed.query or parsed.fragment:
        raise ValueError(f"{name} must be one exact HTTPS origin")
    return f"https://{parsed.netloc.lower()}"


def make_session(*, cert: str, key: str, ca: str, token_file: str | None = None) -> requests.Session:
    session = requests.Session()
    session.trust_env = False
    session.cert = (cert, key)
    session.verify = ca
    session.headers.update({"Accept": "application/json", "Accept-Encoding": "identity"})
    if token_file is not None:
        token = read_file(Path(token_file), 16384).decode("utf-8", errors="strict").strip()
        if not token or any(character.isspace() for character in token):
            raise ValueError("bearer token file is empty or malformed")
        session.headers["Authorization"] = "Bearer " + token
    return session


def request_json(session: requests.Session, method: str, origin: str, path: str, *, timeout: tuple[float, float], maximum: int, body: dict[str, Any] | None = None, headers: dict[str, str] | None = None) -> dict[str, Any]:
    origin = exact_origin(origin, "request")
    if not path.startswith("/") or path.startswith("//"):
        raise ValueError("request path must be origin-relative")
    request_headers = dict(headers or {})
    raw = None
    if body is not None:
        raw = canonical(body)
        request_headers["Content-Type"] = "application/json"
    with session.request(method, origin + path, data=raw, headers=request_headers, timeout=timeout, allow_redirects=False, stream=True) as response:
        if 300 <= response.status_code < 400 or response.headers.get("Location"):
            raise RuntimeError("redirects are forbidden")
        if response.headers.get("Content-Encoding", "identity").lower() != "identity":
            raise RuntimeError("unexpected response content encoding")
        if response.headers.get("Content-Type", "").split(";", 1)[0].strip().lower() != "application/json":
            raise RuntimeError("response is not application/json")
        chunks: list[bytes] = []
        observed = 0
        for chunk in response.iter_content(chunk_size=65536):
            observed += len(chunk)
            if observed > maximum:
                raise RuntimeError("decoded response exceeds the signed size bound")
            chunks.append(chunk)
        response.raise_for_status()
    return strict_json(b"".join(chunks))


def exact_ids(value: object, name: str, maximum: int) -> list[str]:
    if not isinstance(value, list) or not value or len(value) > maximum or any(not isinstance(item, str) or not item for item in value) or value != sorted(set(value)):
        raise ValueError(f"{name} must be a sorted, unique, bounded exact-ID population")
    return value

Perform an exact compare-and-swap alias change

# File: recovery/atomic_vector_alias.py
from __future__ import annotations

import argparse
import hashlib
import json
import math
import os
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.parse import quote

from recovery.strict_io import canonical, exact_origin, signed_packet, verified_packet, make_session, request_json, verified_envelope

SAFE_NAME = re.compile(r"^[a-z0-9][a-z0-9._-]{0,254}$")
INPUT_LIMIT = int(os.environ["R005_ALIAS_MAXIMUM_INPUT_BYTES"])
RESPONSE_LIMIT = int(os.environ["R005_ALIAS_MAXIMUM_RESPONSE_BYTES"])
CONNECT_TIMEOUT = float(os.environ["R005_ALIAS_CONNECT_TIMEOUT_SECONDS"])
READ_TIMEOUT = float(os.environ["R005_ALIAS_READ_TIMEOUT_SECONDS"])
if INPUT_LIMIT < 1 or RESPONSE_LIMIT < 1 or any(not math.isfinite(value) or value <= 0 for value in (CONNECT_TIMEOUT, READ_TIMEOUT)):
    raise RuntimeError("signed alias runtime profile is invalid")
TIMEOUT = (CONNECT_TIMEOUT, READ_TIMEOUT)


def require_digest(value: object, name: str) -> str:
    if not isinstance(value, str) or not re.fullmatch(r"[0-9a-f]{64}", value) or set(value) == {"0"}:
        raise ValueError(f"{name} must be a nonzero SHA-256 digest")
    return value


def load_decision(path: Path) -> dict[str, Any]:
    decision = verified_envelope(path, INPUT_LIMIT, "aidefend.r005.alias-decision.v1", "VECTOR_DECISION_VERIFY_KEY_FILE")
    fields = {"schema", "tenant_id", "application_id", "decision_id", "alias", "expected_current_index", "target_index", "target_manifest_sha256", "policy_digest", "opensearch_origin", "evidence_origin", "expires_at"}
    if set(decision) != fields:
        raise ValueError("alias decision fields differ")
    for name in ("tenant_id", "application_id", "decision_id", "alias", "expected_current_index", "target_index"):
        if not isinstance(decision[name], str) or not SAFE_NAME.fullmatch(decision[name]):
            raise ValueError(f"unsafe {name}")
    require_digest(decision["target_manifest_sha256"], "target manifest")
    require_digest(decision["policy_digest"], "policy")
    expiry = datetime.fromisoformat(decision["expires_at"].replace("Z", "+00:00"))
    if expiry.tzinfo is None or datetime.now(timezone.utc) > expiry.astimezone(timezone.utc):
        raise RuntimeError("alias decision is expired")
    if decision["target_index"] == decision["expected_current_index"]:
        raise ValueError("target must differ from current index")
    for name, environment in (("opensearch_origin", "OPENSEARCH_ORIGIN"), ("evidence_origin", "EVIDENCE_ORIGIN")):
        if exact_origin(decision[name], name) != exact_origin(os.environ[environment], environment):
            raise RuntimeError(f"{name} is not the locally allowlisted origin")
    if exact_origin(decision["opensearch_origin"], "opensearch") == exact_origin(decision["evidence_origin"], "evidence"):
        raise RuntimeError("control plane and evidence store must use separate origins")
    return decision


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--decision", type=Path, required=True)
    args = parser.parse_args()
    decision = load_decision(args.decision)
    search_credentials = (os.environ["OPENSEARCH_CLIENT_CERT"], os.environ["OPENSEARCH_CLIENT_KEY"])
    evidence_credentials = (os.environ["EVIDENCE_CLIENT_CERT"], os.environ["EVIDENCE_CLIENT_KEY"])
    if search_credentials == evidence_credentials:
        raise RuntimeError("OpenSearch and evidence identities must differ")
    search = make_session(cert=search_credentials[0], key=search_credentials[1], ca=os.environ["OPENSEARCH_CA_FILE"])
    evidence = make_session(cert=evidence_credentials[0], key=evidence_credentials[1], ca=os.environ["EVIDENCE_CA_FILE"])
    origin = decision["opensearch_origin"]
    alias_path = "/_alias/" + quote(decision["alias"], safe="")

    def target() -> str:
        state = request_json(search, "GET", origin, alias_path, timeout=TIMEOUT, maximum=RESPONSE_LIMIT)
        targets = sorted(state)
        if len(targets) != 1 or not isinstance(state[targets[0]], dict):
            raise RuntimeError(f"alias must resolve to exactly one index: {targets}")
        aliases = state[targets[0]].get("aliases")
        if not isinstance(aliases, dict) or set(aliases) != {decision["alias"]}:
            raise RuntimeError("alias readback population differs")
        return targets[0]

    before = target()
    if before != decision["expected_current_index"]:
        raise RuntimeError("alias compare-and-swap precondition failed")
    index_path = "/" + quote(decision["target_index"], safe="")
    settings = request_json(search, "GET", origin, index_path + "/_settings?flat_settings=true", timeout=TIMEOUT, maximum=RESPONSE_LIMIT)
    mappings = request_json(search, "GET", origin, index_path + "/_mapping", timeout=TIMEOUT, maximum=RESPONSE_LIMIT)
    if set(settings) != {decision["target_index"]} or set(mappings) != {decision["target_index"]}:
        raise RuntimeError("target settings/mapping response population differs")
    target_settings = settings[decision["target_index"]].get("settings")
    metadata = mappings[decision["target_index"]].get("mappings", {}).get("_meta", {})
    if not isinstance(target_settings, dict) or target_settings.get("index.blocks.write") != "true":
        raise RuntimeError("target index is not write-blocked")
    if metadata.get("aidefend_manifest_sha256") != decision["target_manifest_sha256"] or metadata.get("tenant_id") != decision["tenant_id"] or metadata.get("application_id") != decision["application_id"]:
        raise RuntimeError("target manifest or scope binding differs")
    update = request_json(search, "POST", origin, "/_aliases", timeout=TIMEOUT, maximum=RESPONSE_LIMIT, body={"actions": [{"remove": {"index": before, "alias": decision["alias"]}}, {"add": {"index": decision["target_index"], "alias": decision["alias"], "is_write_index": False}}]})
    if update.get("acknowledged") is not True:
        raise RuntimeError("OpenSearch did not acknowledge the atomic alias operation")
    after = target()
    if after != decision["target_index"]:
        raise RuntimeError("alias readback differs from target")
    payload = {"schema": "aidefend.r005.alias-operation.v1", "outcome": "PASS", "decision_id": decision["decision_id"], "tenant_id": decision["tenant_id"], "application_id": decision["application_id"], "alias": decision["alias"], "before_index": before, "after_index": after, "target_manifest_sha256": decision["target_manifest_sha256"], "decision_sha256": hashlib.sha256(canonical(decision)).hexdigest(), "completed_at": datetime.now(timezone.utc).isoformat()}
    envelope = signed_packet(payload, "VECTOR_RESULT_SIGNING_KEY_FILE")
    body_digest = hashlib.sha256(canonical(envelope)).hexdigest()
    receipt = request_json(evidence, "POST", decision["evidence_origin"], "/v1/vector-alias-operations", timeout=TIMEOUT, maximum=RESPONSE_LIMIT, body=envelope, headers={"Idempotency-Key": decision["decision_id"]})
    if set(receipt) != {"receipt_digest"} or receipt["receipt_digest"] != body_digest:
        raise RuntimeError("evidence-store readback digest differs")
    print(json.dumps({"outcome": "PASS", "receipt_digest": body_digest}, sort_keys=True))


if __name__ == "__main__":
    main()

The signed decision is the complete authorization population for one alias transition. Test a wrong origin, redirect, oversized or duplicate-key response, non-finite response, two alias targets, changed target metadata, reused credentials, stale decision, and receipt mismatch. A separate read-only identity repeats the alias/settings/mapping GETs and signs the same before/after and manifest binding; an acknowledgement without that exact readback is not a pass.

Separate authority verification from execution receipts

Use independently provisioned Ed25519 public keys for policy, scope, source inventory, model manifest, detector findings and human approval inputs. The executor cannot access those issuers' private keys. Receipt/tombstone signing keys belong to the named output producer and are not accepted as approval keys. Pin keys through the verified deployment profile; canonical signatures cover the entire payload, while the existing tenant, application, digest, freshness and idempotency checks remain mandatory. Install cryptography through the approved dependency lock. An independent read-only verifier checks signatures and serving-state readback; an executor-signed receipt alone does not prove authorized approval or successful recovery. Replay payload/key tampering, wrong scope, expired approval and replayed decisions.

AID-R-005-G002

Verify the exact signed quarantine population

Consume a signed finding scope and quarantine policy with exact, sorted point IDs. The policy binds separate Qdrant, Object-Lock, tombstone-control, and retrieval-probe origins plus signed size, count, retry, and deadline limits. The identities for those boundaries must be different. This method preserves every named point before publishing exclusion state; it never silently drops a missing point.

# File: recovery/quarantine_chunks.py
from __future__ import annotations

import argparse
import base64
import hashlib
import json
import math
import os
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.parse import quote

import boto3
from botocore.config import Config

from recovery.strict_io import canonical, exact_ids, exact_origin, signed_packet, verified_packet, make_session, read_file, request_json, verified_envelope

SAFE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,254}$")
INPUT_LIMIT = int(os.environ["R005_QUARANTINE_MAXIMUM_INPUT_BYTES"])
RESPONSE_LIMIT = int(os.environ["R005_QUARANTINE_MAXIMUM_RESPONSE_BYTES"])
CONNECT_TIMEOUT = float(os.environ["R005_QUARANTINE_CONNECT_TIMEOUT_SECONDS"])
READ_TIMEOUT = float(os.environ["R005_QUARANTINE_READ_TIMEOUT_SECONDS"])
if INPUT_LIMIT < 1 or RESPONSE_LIMIT < 1 or any(not math.isfinite(value) or value <= 0 for value in (CONNECT_TIMEOUT, READ_TIMEOUT)):
    raise RuntimeError("signed quarantine runtime profile is invalid")
TIMEOUT = (CONNECT_TIMEOUT, READ_TIMEOUT)
SCOPE_FIELDS = {"schema", "tenant_id", "application_id", "incident_id", "source_index", "serving_alias", "index_manifest_sha256", "chunk_ids", "reason"}
POLICY_FIELDS = {"schema", "tenant_id", "application_id", "maximum_chunk_count", "maximum_artifact_bytes", "qdrant_origin", "quarantine_control_origin", "retrieval_probe_origin", "s3_origin", "s3_region", "quarantine_bucket", "kms_key_id", "retain_until", "maximum_sdk_attempts"}


def digest(value: object, name: str) -> str:
    if not isinstance(value, str) or not re.fullmatch(r"[0-9a-f]{64}", value) or set(value) == {"0"}:
        raise ValueError(f"{name} must be a nonzero SHA-256 digest")
    return value


def bounded_positive(value: object, name: str) -> int:
    if isinstance(value, bool) or not isinstance(value, int) or value < 1:
        raise ValueError(f"{name} must be a positive integer")
    return value


def qdrant_points(scope: dict[str, Any], policy: dict[str, Any]) -> list[dict[str, Any]]:
    maximum = bounded_positive(policy["maximum_chunk_count"], "maximum chunk count")
    ids = exact_ids(scope["chunk_ids"], "chunk_ids", maximum)
    session = make_session(cert=os.environ["QDRANT_CLIENT_CERT"], key=os.environ["QDRANT_CLIENT_KEY"], ca=os.environ["QDRANT_CA_FILE"])
    api_key = read_file(Path(os.environ["QDRANT_API_KEY_FILE"]), 16384).decode("utf-8", errors="strict").strip()
    if not api_key or any(character.isspace() for character in api_key):
        raise ValueError("Qdrant API key file is malformed")
    session.headers["api-key"] = api_key
    response = request_json(session, "POST", policy["qdrant_origin"], "/collections/" + quote(scope["source_index"], safe="") + "/points", timeout=TIMEOUT, maximum=RESPONSE_LIMIT, body={"ids": ids, "with_payload": True, "with_vector": True})
    if response.get("status") != "ok" or not isinstance(response.get("result"), list):
        raise RuntimeError("Qdrant retrieval response schema differs")
    by_id: dict[str, dict[str, Any]] = {}
    for point in response["result"]:
        if not isinstance(point, dict) or not isinstance(point.get("id"), str) or point["id"] not in ids or point["id"] in by_id or "payload" not in point or "vector" not in point:
            raise RuntimeError("Qdrant returned an unknown, duplicate, or incomplete point")
        by_id[point["id"]] = point
    if set(by_id) != set(ids):
        raise RuntimeError(f"source point population differs: missing={sorted(set(ids) - set(by_id))}")
    records = []
    for point_id in ids:
        payload = by_id[point_id]["payload"]
        vector = by_id[point_id]["vector"]
        if not isinstance(payload, dict) or not isinstance(vector, (list, dict)):
            raise RuntimeError(f"point {point_id} payload/vector type differs")
        # This text-ingestion path hashes the exact UTF-8 text represented by
        # the point, not its metadata wrapper or a mutable source URI.
        text = payload.get("text")
        if not isinstance(text, str) or not text:
            raise RuntimeError("source content identity unavailable; keep quarantine incomplete")
        source_content_sha256 = hashlib.sha256(text.encode("utf-8")).hexdigest()
        records.append({"point_id": point_id, "source_content_sha256": source_content_sha256, "payload": payload, "payload_sha256": hashlib.sha256(canonical(payload)).hexdigest(), "vector": vector, "vector_sha256": hashlib.sha256(canonical(vector)).hexdigest()})
    return records


def write_worm(scope: dict[str, Any], policy: dict[str, Any], records: list[dict[str, Any]]) -> dict[str, str]:
    artifact = {"schema": "aidefend.r005.vector-quarantine-artifact.v1", "tenant_id": scope["tenant_id"], "application_id": scope["application_id"], "incident_id": scope["incident_id"], "source_index": scope["source_index"], "index_manifest_sha256": scope["index_manifest_sha256"], "reason": scope["reason"], "records": records, "created_at": datetime.now(timezone.utc).isoformat()}
    body = canonical(artifact)
    if len(body) > bounded_positive(policy["maximum_artifact_bytes"], "maximum artifact bytes"):
        raise RuntimeError("quarantine artifact exceeds the signed size bound")
    artifact_digest = hashlib.sha256(body).hexdigest()
    object_key = "/".join((scope["tenant_id"], scope["incident_id"], artifact_digest + ".json"))
    checksum = base64.b64encode(hashlib.sha256(body).digest()).decode("ascii")
    retain_until = datetime.fromisoformat(policy["retain_until"].replace("Z", "+00:00"))
    if retain_until.tzinfo is None or retain_until.astimezone(timezone.utc) <= datetime.now(timezone.utc):
        raise ValueError("retention deadline must be future and timezone-aware")
    client = boto3.client("s3", region_name=policy["s3_region"], endpoint_url=policy["s3_origin"], config=Config(connect_timeout=CONNECT_TIMEOUT, read_timeout=READ_TIMEOUT, retries={"mode": "standard", "max_attempts": bounded_positive(policy["maximum_sdk_attempts"], "maximum SDK attempts")}, s3={"addressing_style": "path"}))
    put = client.put_object(Bucket=policy["quarantine_bucket"], Key=object_key, Body=body, ContentType="application/json", ServerSideEncryption="aws:kms", SSEKMSKeyId=policy["kms_key_id"], ObjectLockMode="COMPLIANCE", ObjectLockRetainUntilDate=retain_until, ChecksumSHA256=checksum, IfNoneMatch="*")
    version = put.get("VersionId")
    if not isinstance(version, str) or not version:
        raise RuntimeError("immutable S3 version ID is missing")
    head = client.head_object(Bucket=policy["quarantine_bucket"], Key=object_key, VersionId=version, ChecksumMode="ENABLED")
    if head.get("ChecksumSHA256") != checksum or head.get("ObjectLockMode") != "COMPLIANCE" or head.get("ContentLength") != len(body) or head.get("VersionId") != version:
        raise RuntimeError("quarantine object checksum, size, version, or retention readback differs")
    return {"bucket": policy["quarantine_bucket"], "key": object_key, "version_id": version, "artifact_sha256": artifact_digest, "checksum_sha256_base64": checksum}


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--scope", type=Path, required=True)
    parser.add_argument("--policy", type=Path, required=True)
    args = parser.parse_args()
    scope = verified_envelope(args.scope, INPUT_LIMIT, "aidefend.r005.quarantine-scope.v1", "VECTOR_SCOPE_VERIFY_KEY_FILE")
    policy = verified_envelope(args.policy, INPUT_LIMIT, "aidefend.r005.quarantine-policy.v1", "RECOVERY_POLICY_VERIFY_KEY_FILE")
    if set(scope) != SCOPE_FIELDS or set(policy) != POLICY_FIELDS:
        raise ValueError("quarantine scope or policy fields differ")
    for name in ("tenant_id", "application_id", "incident_id", "source_index", "serving_alias"):
        if not isinstance(scope[name], str) or not SAFE.fullmatch(scope[name]):
            raise ValueError(f"unsafe {name}")
    if (scope["tenant_id"], scope["application_id"]) != (policy["tenant_id"], policy["application_id"]):
        raise RuntimeError("scope-to-policy binding differs")
    digest(scope["index_manifest_sha256"], "index manifest")
    origins = {name: exact_origin(policy[name], name) for name in ("qdrant_origin", "quarantine_control_origin", "retrieval_probe_origin", "s3_origin")}
    for name, environment in (("qdrant_origin", "QDRANT_ORIGIN"), ("quarantine_control_origin", "QUARANTINE_CONTROL_ORIGIN"), ("retrieval_probe_origin", "RETRIEVAL_PROBE_ORIGIN"), ("s3_origin", "QUARANTINE_S3_ORIGIN")):
        if origins[name] != exact_origin(os.environ[environment], environment):
            raise RuntimeError(f"{name} is not locally allowlisted")
    if len(set(origins.values())) != len(origins):
        raise RuntimeError("quarantine trust boundaries must use separate origins")
    records = qdrant_points(scope, policy)
    artifact = write_worm(scope, policy, records)
    point_ids = [record["point_id"] for record in records]
    tombstone_payload = {"schema": "aidefend.r005.vector-tombstones.v1", "tenant_id": scope["tenant_id"], "application_id": scope["application_id"], "incident_id": scope["incident_id"], "serving_alias": scope["serving_alias"], "source_index": scope["source_index"], "index_manifest_sha256": scope["index_manifest_sha256"], "point_ids": point_ids, "point_content_digests": [{"point_id": record["point_id"], "source_content_sha256": record["source_content_sha256"], "payload_sha256": record["payload_sha256"], "vector_sha256": record["vector_sha256"]} for record in records], "quarantine_artifact_sha256": artifact["artifact_sha256"], "policy_sha256": hashlib.sha256(canonical(policy)).hexdigest(), "created_at": datetime.now(timezone.utc).isoformat()}
    envelope = signed_packet(tombstone_payload, "VECTOR_TOMBSTONE_SIGNING_KEY_FILE")
    tombstone_digest = hashlib.sha256(canonical(envelope)).hexdigest()
    control = make_session(cert=os.environ["QUARANTINE_CLIENT_CERT"], key=os.environ["QUARANTINE_CLIENT_KEY"], ca=os.environ["QUARANTINE_CA_FILE"])
    probe = make_session(cert=os.environ["PROBE_CLIENT_CERT"], key=os.environ["PROBE_CLIENT_KEY"], ca=os.environ["PROBE_CA_FILE"])
    if (control.cert, control.verify) == (probe.cert, probe.verify):
        raise RuntimeError("tombstone-control and retrieval-probe credentials/trust must differ")
    receipt = request_json(control, "POST", policy["quarantine_control_origin"], "/v1/vector-tombstone-sets", timeout=TIMEOUT, maximum=RESPONSE_LIMIT, body=envelope, headers={"Idempotency-Key": tombstone_digest})
    if set(receipt) != {"tombstone_set_digest", "effective_point_ids"} or receipt["tombstone_set_digest"] != tombstone_digest or receipt["effective_point_ids"] != point_ids:
        raise RuntimeError("effective tombstone population readback differs")
    exclusion = request_json(probe, "POST", policy["retrieval_probe_origin"], "/v1/security/tombstone-probe", timeout=TIMEOUT, maximum=RESPONSE_LIMIT, body={"tenant_id": scope["tenant_id"], "application_id": scope["application_id"], "serving_alias": scope["serving_alias"], "point_ids": point_ids, "incident_id": scope["incident_id"]})
    if set(exclusion) != {"serving_alias", "excluded_point_ids", "retrievable_point_ids"} or exclusion["serving_alias"] != scope["serving_alias"] or exclusion["excluded_point_ids"] != point_ids or exclusion["retrievable_point_ids"] != []:
        raise RuntimeError("production exclusion readback differs")
    result = {"outcome": "PASS", "incident_id": scope["incident_id"], "population_count": len(point_ids), "population_sha256": hashlib.sha256(canonical(point_ids)).hexdigest(), "quarantine_artifact": artifact, "tombstone_set_digest": tombstone_digest, "probe_sha256": hashlib.sha256(canonical(exclusion)).hexdigest()}
    print(json.dumps(result, sort_keys=True))


if __name__ == "__main__":
    main()

Exercise wrong and duplicate point IDs, missing points, malformed vectors, non-finite values, cross-tenant scope, expired retention, redirect, decompression/size overflow, incorrect S3 version/checksum/lock, partial tombstone activation, and a retrievable quarantined point. Preserve the exact Qdrant response digest, Object-Lock version, effective tombstone population, and independent probe result. Any absent point or unavailable boundary is ERROR or INSUFFICIENT_DATA, never a successful quarantine.

Content identity: This runnable text-ingestion path requires each source point to retain its actual UTF-8 text in payload.text. The source_content_sha256 hashes those bytes, excluding point ID, URI and metadata. If an older collection lacks that text, stop with INSUFFICIENT_DATA and recover it through its independently verified immutable source manifest before rebuilding; never substitute the full payload hash or trust an unverified content_sha256 field. Non-text ingestion needs its own explicit source-byte contract.

Separate authority verification from execution receipts

Use independently provisioned Ed25519 public keys for policy, scope, source inventory, model manifest, detector findings and human approval inputs. The executor cannot access those issuers' private keys. Receipt/tombstone signing keys belong to the named output producer and are not accepted as approval keys. Pin keys through the verified deployment profile; canonical signatures cover the entire payload, while the existing tenant, application, digest, freshness and idempotency checks remain mandatory. Install cryptography through the approved dependency lock. An independent read-only verifier checks signatures and serving-state readback; an executor-signed receipt alone does not prove authorized approval or successful recovery. Replay payload/key tampering, wrong scope, expired approval and replayed decisions.

AID-R-005-G003

Build only from the complete signed source population

Use four separately authenticated inputs: source inventory, build policy, local embedding-model file manifest, and effective tombstone receipt. The policy binds one staging root, one new candidate collection, exact HTTPS scanner/Qdrant/evidence origins, distinct identities, model and source byte limits, batch limits, and command/network deadlines. A missing, extra, rejected, or tombstoned source aborts the build.

# File: recovery/rebuild_candidate_index.py
from __future__ import annotations

import argparse
import base64
import hashlib
import json
import math
import os
import re
import subprocess
import struct
import tempfile
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath
from typing import Any
from urllib.parse import quote

from sentence_transformers import SentenceTransformer

from recovery.strict_io import signed_packet, verified_packet, canonical, exact_origin, make_session, read_file, request_json, verified_envelope

SAFE_NAME = re.compile(r"^[a-z0-9][a-z0-9._-]{0,254}$")
HEX = re.compile(r"^[0-9a-f]{64}$")
INPUT_LIMIT = int(os.environ["R005_REBUILD_MAXIMUM_INPUT_BYTES"])
RESPONSE_LIMIT = int(os.environ["R005_REBUILD_MAXIMUM_RESPONSE_BYTES"])
CONNECT_TIMEOUT = float(os.environ["R005_REBUILD_CONNECT_TIMEOUT_SECONDS"])
READ_TIMEOUT = float(os.environ["R005_REBUILD_READ_TIMEOUT_SECONDS"])
COMMAND_TIMEOUT = float(os.environ["R005_REBUILD_COMMAND_TIMEOUT_SECONDS"])
if INPUT_LIMIT < 1 or RESPONSE_LIMIT < 1 or any(not math.isfinite(value) or value <= 0 for value in (CONNECT_TIMEOUT, READ_TIMEOUT, COMMAND_TIMEOUT)):
    raise RuntimeError("signed candidate-build runtime profile is invalid")
TIMEOUT = (CONNECT_TIMEOUT, READ_TIMEOUT)
INVENTORY_FIELDS = {"schema", "tenant_id", "application_id", "documents"}
DOCUMENT_FIELDS = {"point_id", "source_uri", "staged_relative_path", "content_sha256"}
POLICY_FIELDS = {"schema", "tenant_id", "application_id", "candidate_collection", "staging_root", "embedding_model_root", "embedding_model_manifest_sha256", "embedding_dimension", "distance_metric", "maximum_document_count", "maximum_source_bytes", "maximum_model_file_bytes", "maximum_model_file_count", "maximum_request_bytes", "upsert_batch_size", "content_policy_digest", "tombstone_set_digest", "scanner_origin", "qdrant_origin", "evidence_origin", "builder_kms_key_uri", "builder_public_key"}
MODEL_FIELDS = {"schema", "tenant_id", "application_id", "files"}
MODEL_FILE_FIELDS = {"path", "size_bytes", "sha256"}
TOMBSTONE_FIELDS = {"schema", "tenant_id", "application_id", "incident_id", "serving_alias", "source_index", "index_manifest_sha256", "point_ids", "point_content_digests", "quarantine_artifact_sha256", "policy_sha256", "created_at"}


def digest(value: object, name: str) -> str:
    if not isinstance(value, str) or HEX.fullmatch(value) is None or set(value) == {"0"}:
        raise ValueError(f"{name} must be a nonzero SHA-256 digest")
    return value


def positive_int(value: object, name: str) -> int:
    if isinstance(value, bool) or not isinstance(value, int) or value < 1:
        raise ValueError(f"{name} must be a positive integer")
    return value


def vector_sha256(vector: object, dimension: int) -> str:
    if not isinstance(vector, list) or len(vector) != dimension:
        raise ValueError("vector dimension differs")
    digestor = hashlib.sha256()
    for item in vector:
        if isinstance(item, bool) or not isinstance(item, (int, float)) or not math.isfinite(float(item)):
            raise ValueError("vector contains a non-finite or non-numeric coordinate")
        digestor.update(struct.pack("<f", float(item)))
    return digestor.hexdigest()


def safe_child(root: Path, relative: object, name: str) -> Path:
    if not isinstance(relative, str) or not relative:
        raise ValueError(f"{name} path is required")
    pure = PurePosixPath(relative)
    if pure.is_absolute() or ".." in pure.parts or "." in pure.parts:
        raise ValueError(f"unsafe {name} relative path")
    resolved_root = root.resolve(strict=True)
    candidate = resolved_root.joinpath(*pure.parts).resolve(strict=True)
    if candidate == resolved_root or resolved_root not in candidate.parents or candidate.is_symlink() or not candidate.is_file():
        raise ValueError(f"{name} escapes its signed root or is not a regular file")
    return candidate


def verify_model(root: Path, manifest: dict[str, Any], policy: dict[str, Any]) -> str:
    if set(manifest) != MODEL_FIELDS or manifest["schema"] != "aidefend.r005.embedding-model-files.v1":
        raise ValueError("embedding-model manifest schema differs")
    files = manifest["files"]
    maximum_count = positive_int(policy["maximum_model_file_count"], "maximum model file count")
    if not isinstance(files, list) or not files or len(files) > maximum_count:
        raise ValueError("embedding-model file population is invalid")
    expected: list[str] = []
    for entry in files:
        if not isinstance(entry, dict) or set(entry) != MODEL_FILE_FIELDS:
            raise ValueError("embedding-model file entry differs")
        path = safe_child(root, entry["path"], "model")
        relative = path.relative_to(root.resolve(strict=True)).as_posix()
        if relative in expected:
            raise ValueError("duplicate embedding-model path")
        expected.append(relative)
        raw = read_file(path, positive_int(policy["maximum_model_file_bytes"], "maximum model file bytes"))
        if len(raw) != positive_int(entry["size_bytes"], "model file size") or hashlib.sha256(raw).hexdigest() != digest(entry["sha256"], "model file"):
            raise RuntimeError(f"embedding-model file differs: {relative}")
    actual = sorted(path.relative_to(root.resolve(strict=True)).as_posix() for path in root.resolve(strict=True).rglob("*") if path.is_file() and not path.is_symlink())
    if sorted(expected) != actual:
        raise RuntimeError("embedding-model file population differs")
    result = hashlib.sha256(canonical(manifest)).hexdigest()
    if result != digest(policy["embedding_model_manifest_sha256"], "embedding-model manifest"):
        raise RuntimeError("embedding-model manifest is not policy-approved")
    return result


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--inventory", type=Path, required=True)
    parser.add_argument("--policy", type=Path, required=True)
    parser.add_argument("--model-manifest", type=Path, required=True)
    parser.add_argument("--tombstones", type=Path, required=True)
    parser.add_argument("--manifest-output", type=Path, default=Path("recovery/candidate-manifest.json"))
    args = parser.parse_args()
    inventory = verified_envelope(args.inventory, INPUT_LIMIT, "aidefend.r005.source-inventory.v1", "SOURCE_INVENTORY_VERIFY_KEY_FILE")
    policy = verified_envelope(args.policy, INPUT_LIMIT, "aidefend.r005.build-policy.v1", "RECOVERY_POLICY_VERIFY_KEY_FILE")
    model_manifest = verified_envelope(args.model_manifest, INPUT_LIMIT, "aidefend.r005.embedding-model-files.v1", "MODEL_MANIFEST_VERIFY_KEY_FILE")
    tombstone_envelope = verified_packet(args.tombstones, INPUT_LIMIT, "aidefend.r005.vector-tombstones.v1", "VECTOR_TOMBSTONE_VERIFY_KEY_FILE")
    tombstones = tombstone_envelope["payload"]
    if set(inventory) != INVENTORY_FIELDS or set(policy) != POLICY_FIELDS or set(tombstones) != TOMBSTONE_FIELDS:
        raise ValueError("candidate-build input schema differs")
    scope = (policy["tenant_id"], policy["application_id"])
    if any((item.get("tenant_id"), item.get("application_id")) != scope for item in (inventory, model_manifest, tombstones)):
        raise RuntimeError("candidate-build input scope differs")
    candidate = policy["candidate_collection"]
    if not isinstance(candidate, str) or SAFE_NAME.fullmatch(candidate) is None:
        raise ValueError("unsafe candidate collection")
    if hashlib.sha256(canonical(tombstone_envelope)).hexdigest() != digest(policy["tombstone_set_digest"], "tombstone set"):
        raise RuntimeError("effective tombstone receipt differs from policy")
    origins = {name: exact_origin(policy[name], name) for name in ("scanner_origin", "qdrant_origin", "evidence_origin")}
    for name, environment in (("scanner_origin", "CONTENT_SCANNER_ORIGIN"), ("qdrant_origin", "QDRANT_ORIGIN"), ("evidence_origin", "EVIDENCE_ORIGIN")):
        if origins[name] != exact_origin(os.environ[environment], environment):
            raise RuntimeError(f"{name} is not locally allowlisted")
    if len(set(origins.values())) != 3:
        raise RuntimeError("scanner, vector store, and evidence origins must differ")

    documents = inventory["documents"]
    maximum_documents = positive_int(policy["maximum_document_count"], "maximum document count")
    if not isinstance(documents, list) or not documents or len(documents) > maximum_documents:
        raise ValueError("source document population is invalid")
    ids: list[str] = []
    for document in documents:
        if not isinstance(document, dict) or set(document) != DOCUMENT_FIELDS or not isinstance(document["point_id"], str) or SAFE_NAME.fullmatch(document["point_id"]) is None or not isinstance(document["source_uri"], str) or not document["source_uri"]:
            raise ValueError("source document schema differs")
        digest(document["content_sha256"], "source content")
        ids.append(document["point_id"])
    if ids != sorted(set(ids)):
        raise ValueError("source point IDs must be sorted and unique")
    if tombstones["point_ids"] != sorted(set(tombstones["point_ids"])):
        raise ValueError("effective tombstone IDs are not an exact sorted population")
    tombstone_entries = tombstones["point_content_digests"]
    if not isinstance(tombstone_entries, list) or len(tombstone_entries) != len(tombstones["point_ids"]):
        raise ValueError("tombstone digest population differs")
    tombstoned_content: set[str] = set()
    tombstone_entry_ids: list[str] = []
    for entry in tombstone_entries:
        if not isinstance(entry, dict) or set(entry) != {"point_id", "source_content_sha256", "payload_sha256", "vector_sha256"}:
            raise ValueError("tombstone digest entry schema differs")
        tombstone_entry_ids.append(entry["point_id"])
        tombstoned_content.add(digest(entry["source_content_sha256"], "tombstoned source content"))
        digest(entry["payload_sha256"], "tombstoned payload")
        digest(entry["vector_sha256"], "tombstoned vector")
    if tombstone_entry_ids != tombstones["point_ids"]:
        raise ValueError("tombstone ID/digest populations differ")

    model_root = Path(policy["embedding_model_root"])
    model_digest = verify_model(model_root, model_manifest, policy)
    encoder = SentenceTransformer(str(model_root.resolve(strict=True)), local_files_only=True, trust_remote_code=False)
    dimension = positive_int(policy["embedding_dimension"], "embedding dimension")
    if encoder.get_sentence_embedding_dimension() != dimension or policy["distance_metric"] not in {"Cosine", "Dot", "Euclid", "Manhattan"}:
        raise RuntimeError("embedding dimension or distance differs from signed policy")
    scanner = make_session(cert=os.environ["SCANNER_CLIENT_CERT"], key=os.environ["SCANNER_CLIENT_KEY"], ca=os.environ["SCANNER_CA_FILE"])
    qdrant = make_session(cert=os.environ["QDRANT_CLIENT_CERT"], key=os.environ["QDRANT_CLIENT_KEY"], ca=os.environ["QDRANT_CA_FILE"])
    qdrant_api_key = read_file(Path(os.environ["QDRANT_API_KEY_FILE"]), 16384).decode("utf-8", errors="strict").strip()
    if not qdrant_api_key or any(character.isspace() for character in qdrant_api_key):
        raise ValueError("Qdrant API key file is malformed")
    qdrant.headers["api-key"] = qdrant_api_key
    evidence = make_session(cert=os.environ["EVIDENCE_CLIENT_CERT"], key=os.environ["EVIDENCE_CLIENT_KEY"], ca=os.environ["EVIDENCE_CA_FILE"])
    if len({scanner.cert, qdrant.cert, evidence.cert}) != 3:
        raise RuntimeError("scanner, Qdrant, and evidence identities must differ")
    collection_path = "/collections/" + quote(candidate, safe="")
    created = request_json(qdrant, "PUT", policy["qdrant_origin"], collection_path, timeout=TIMEOUT, maximum=RESPONSE_LIMIT, body={"vectors": {"size": dimension, "distance": policy["distance_metric"]}, "on_disk_payload": True})
    if created.get("status") != "ok" or created.get("result") is not True:
        raise RuntimeError("Qdrant must create a new, previously absent candidate collection")
    empty = request_json(qdrant, "POST", policy["qdrant_origin"], collection_path + "/points/count", timeout=TIMEOUT, maximum=RESPONSE_LIMIT, body={"exact": True})
    if empty.get("status") != "ok" or (empty.get("result") or {}).get("count") != 0:
        raise RuntimeError("new candidate collection is not empty")

    staging_root = Path(policy["staging_root"])
    inventory_digest = hashlib.sha256(canonical(inventory)).hexdigest()
    points: list[dict[str, Any]] = []
    manifest_rows: list[dict[str, Any]] = []
    for document in documents:
        content = read_file(safe_child(staging_root, document["staged_relative_path"], "source"), positive_int(policy["maximum_source_bytes"], "maximum source bytes"))
        content_digest = hashlib.sha256(content).hexdigest()
        if (content_digest != document["content_sha256"] or content_digest in tombstoned_content
                or document["point_id"] in tombstones["point_ids"]):
            raise RuntimeError(f"source changed or is tombstoned: {document['point_id']}")
        scan_body = {"tenant_id": policy["tenant_id"], "application_id": policy["application_id"], "source_uri": document["source_uri"], "content_sha256": content_digest, "content_base64": base64.b64encode(content).decode("ascii"), "policy_digest": digest(policy["content_policy_digest"], "content policy")}
        if len(canonical(scan_body)) > positive_int(policy["maximum_request_bytes"], "maximum request bytes"):
            raise RuntimeError("scanner request exceeds the signed request bound")
        scan = request_json(scanner, "POST", policy["scanner_origin"], "/v1/source-scans", timeout=TIMEOUT, maximum=RESPONSE_LIMIT, body=scan_body)
        if set(scan) != {"outcome", "content_sha256", "policy_digest", "receipt_digest"} or scan["outcome"] != "PASS" or scan["content_sha256"] != content_digest or scan["policy_digest"] != policy["content_policy_digest"]:
            raise RuntimeError(f"source scan rejected or differed: {document['point_id']}")
        vector = encoder.encode(content.decode("utf-8", errors="strict"), normalize_embeddings=True).tolist()
        vector_digest = vector_sha256(vector, dimension)
        payload = {"text": content.decode("utf-8", errors="strict"), "source_uri": document["source_uri"], "content_sha256": content_digest, "source_inventory_sha256": inventory_digest, "content_scan_receipt": scan["receipt_digest"], "tombstone_set_digest": policy["tombstone_set_digest"], "candidate_collection": candidate}
        points.append({"id": document["point_id"], "vector": vector, "payload": payload})
        manifest_rows.append({"point_id": document["point_id"], "vector_sha256": vector_digest, **{k: v for k, v in payload.items() if k != "text"}})
    batch_size = positive_int(policy["upsert_batch_size"], "upsert batch size")
    for offset in range(0, len(points), batch_size):
        body = {"points": points[offset:offset + batch_size]}
        if len(canonical(body)) > positive_int(policy["maximum_request_bytes"], "maximum request bytes"):
            raise RuntimeError("Qdrant upsert batch exceeds signed request bound")
        upsert = request_json(qdrant, "PUT", policy["qdrant_origin"], collection_path + "/points?wait=true&ordering=strong", timeout=TIMEOUT, maximum=RESPONSE_LIMIT, body=body)
        if upsert.get("status") != "ok" or (upsert.get("result") or {}).get("status") != "completed":
            raise RuntimeError("Qdrant upsert did not complete")

    observed: dict[str, dict[str, Any]] = {}
    next_offset: object | None = None
    page_count = 0
    while True:
        body: dict[str, Any] = {"limit": min(batch_size, maximum_documents), "with_payload": True, "with_vector": True}
        if next_offset is not None:
            body["offset"] = next_offset
        page = request_json(qdrant, "POST", policy["qdrant_origin"], collection_path + "/points/scroll", timeout=TIMEOUT, maximum=RESPONSE_LIMIT, body=body)
        result = page.get("result")
        if page.get("status") != "ok" or not isinstance(result, dict) or not isinstance(result.get("points"), list):
            raise RuntimeError("Qdrant complete-population scroll schema differs")
        rows = result["points"]
        for point in rows:
            point_id = point.get("id") if isinstance(point, dict) else None
            if not isinstance(point_id, str) or point_id in observed or not isinstance(point.get("payload"), dict):
                raise RuntimeError("candidate scroll contains an unknown-form or duplicate point")
            observed[point_id] = {"payload": point["payload"], "vector_sha256": vector_sha256(point.get("vector"), dimension)}
            if len(observed) > maximum_documents:
                raise RuntimeError("candidate scroll exceeds the signed population bound")
        following = result.get("next_page_offset")
        page_count += 1
        if following is None:
            break
        if not rows or following == next_offset or page_count > maximum_documents:
            raise RuntimeError("candidate scroll did not make bounded progress")
        next_offset = following
    exact_count = request_json(qdrant, "POST", policy["qdrant_origin"], collection_path + "/points/count", timeout=TIMEOUT, maximum=RESPONSE_LIMIT, body={"exact": True})
    if exact_count.get("status") != "ok" or (exact_count.get("result") or {}).get("count") != len(observed):
        raise RuntimeError("Qdrant exact count differs from the complete scroll")
    expected = {point["id"]: {"payload": point["payload"], "vector_sha256": vector_sha256(point["vector"], dimension)} for point in points}
    if set(observed) != set(ids) or observed != expected:
        raise RuntimeError("candidate contains missing, extra, payload-drifted, or vector-drifted points")

    manifest = {"schema": "aidefend.r005.candidate-manifest.v1", "tenant_id": policy["tenant_id"], "application_id": policy["application_id"], "candidate_collection": candidate, "built_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), "source_inventory_sha256": inventory_digest, "embedding_model_manifest_sha256": model_digest, "tombstone_set_digest": policy["tombstone_set_digest"], "build_policy_sha256": hashlib.sha256(canonical(policy)).hexdigest(), "point_count": len(points), "point_population_sha256": hashlib.sha256(canonical(ids)).hexdigest(), "points": manifest_rows}
    with tempfile.TemporaryDirectory(prefix="r005-build-") as directory:
        snapshot_root = Path(directory)
        os.chmod(snapshot_root, 0o700)
        manifest_path = snapshot_root / "candidate-manifest.json"
        bundle_path = snapshot_root / "candidate-manifest.json.sigstore.json"
        manifest_bytes = canonical(manifest)
        flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
        if hasattr(os, "O_NOFOLLOW"):
            flags |= os.O_NOFOLLOW
        descriptor = os.open(manifest_path, flags, 0o400)
        with os.fdopen(descriptor, "wb") as target:
            target.write(manifest_bytes)
            target.flush()
            os.fsync(target.fileno())
        subprocess.run(["cosign", "sign-blob", "--yes", "--key", policy["builder_kms_key_uri"], "--bundle", str(bundle_path), str(manifest_path)], check=True, capture_output=True, text=True, timeout=COMMAND_TIMEOUT)
        os.chmod(bundle_path, 0o400)
        bundle_bytes = read_file(bundle_path, INPUT_LIMIT)
        if not bundle_bytes:
            raise RuntimeError("Cosign did not produce a bundle")
        subprocess.run(["cosign", "verify-blob", "--key", policy["builder_public_key"], "--bundle", str(bundle_path), str(manifest_path)], check=True, capture_output=True, text=True, timeout=COMMAND_TIMEOUT)
        if read_file(manifest_path, len(manifest_bytes)) != manifest_bytes:
            raise RuntimeError("candidate manifest snapshot changed after signing")
        if read_file(bundle_path, INPUT_LIMIT) != bundle_bytes:
            raise RuntimeError("candidate Sigstore bundle snapshot changed after verification")
        bundle = json.loads(bundle_bytes.decode("utf-8", errors="strict"))
        if not isinstance(bundle, dict):
            raise RuntimeError("Cosign bundle must be one JSON object")
    envelope = {"manifest": manifest, "cosign_bundle": bundle, "public_key": policy["builder_public_key"]}
    envelope_digest = hashlib.sha256(canonical(envelope)).hexdigest()
    receipt = request_json(evidence, "POST", policy["evidence_origin"], "/v1/vector-candidate-manifests", timeout=TIMEOUT, maximum=RESPONSE_LIMIT, body=envelope, headers={"Idempotency-Key": envelope_digest})
    if set(receipt) != {"receipt_digest", "candidate_collection", "point_population_sha256"} or receipt["receipt_digest"] != envelope_digest or receipt["candidate_collection"] != candidate or receipt["point_population_sha256"] != manifest["point_population_sha256"]:
        raise RuntimeError("candidate evidence readback differs")

    output_path = args.manifest_output
    if SAFE_NAME.fullmatch(output_path.name) is None or output_path.name.endswith(".sigstore.json"):
        raise ValueError("manifest output must name a safe payload path, not a Sigstore bundle")
    output_parent = output_path.parent.resolve(strict=True)
    if not output_parent.is_dir():
        raise ValueError("manifest output parent must be an existing directory")
    output_path = output_parent / output_path.name
    output_bundle_path = Path(str(output_path) + ".sigstore.json")
    published: list[Path] = []
    try:
        with tempfile.TemporaryDirectory(prefix=".r005-publish-", dir=output_parent) as publish_directory:
            publish_root = Path(publish_directory)
            os.chmod(publish_root, 0o700)
            staged_manifest = publish_root / output_path.name
            staged_bundle = publish_root / output_bundle_path.name
            output_flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
            if hasattr(os, "O_NOFOLLOW"):
                output_flags |= os.O_NOFOLLOW
            for path, raw in ((staged_manifest, manifest_bytes), (staged_bundle, bundle_bytes)):
                descriptor = os.open(path, output_flags, 0o400)
                with os.fdopen(descriptor, "wb") as target:
                    target.write(raw)
                    target.flush()
                    os.fsync(target.fileno())
                os.chmod(path, 0o400)
            subprocess.run(["cosign", "verify-blob", "--key", policy["builder_public_key"], "--bundle", str(staged_bundle), str(staged_manifest)], check=True, capture_output=True, text=True, timeout=COMMAND_TIMEOUT)
            if read_file(staged_manifest, len(manifest_bytes)) != manifest_bytes or read_file(staged_bundle, INPUT_LIMIT) != bundle_bytes:
                raise RuntimeError("staged candidate payload/bundle pair differs from the verified snapshots")
            # Publish the bundle first; the no-clobber payload link is the atomic visibility/commit point.
            os.link(staged_bundle, output_bundle_path, follow_symlinks=False)
            published.append(output_bundle_path)
            os.link(staged_manifest, output_path, follow_symlinks=False)
            published.append(output_path)
            subprocess.run(["cosign", "verify-blob", "--key", policy["builder_public_key"], "--bundle", str(output_bundle_path), str(output_path)], check=True, capture_output=True, text=True, timeout=COMMAND_TIMEOUT)
            if read_file(output_path, len(manifest_bytes)) != manifest_bytes:
                raise RuntimeError("published candidate manifest differs from the verified snapshot")
            if read_file(output_bundle_path, INPUT_LIMIT) != bundle_bytes:
                raise RuntimeError("published candidate Sigstore bundle differs from the verified snapshot")
    except Exception:
        for path in reversed(published):
            path.unlink(missing_ok=True)
        raise
    print(json.dumps({"outcome": "PASS", "candidate_collection": candidate, "point_count": len(points), "receipt_digest": envelope_digest, "manifest_path": str(output_path), "bundle_path": str(output_bundle_path)}, sort_keys=True))


if __name__ == "__main__":
    main()

Test missing/extra/duplicate source IDs, path escape or symlink, oversized bytes, digest drift, non-finite embeddings, tombstoned content, scanner rejection, reused collection, partial upsert, wrong Qdrant origin, redirect, response overflow, model-file drift, bundle verification failure, and evidence mismatch. A separate builder verifier uses a complete paginated scroll plus an exact count, rejects unknown IDs, and recomputes payload, float32-vector, population, and manifest digests before the alias guidance can consume the receipt.

Pass the original authenticated G002 vector-tombstones.v1 envelope to --tombstones; its canonical envelope SHA-256 must match the live effective tombstone receipt pinned in the signed build policy. Test original IDs with changed text, renamed IDs with identical text, wrong tenant/application, missing source content identity, and a clean new ID/text. A changed scanner verdict does not bypass either tombstone check.

Separate authority verification from execution receipts

Use independently provisioned Ed25519 public keys for policy, scope, source inventory, model manifest, detector findings and human approval inputs. The executor cannot access those issuers' private keys. Receipt/tombstone signing keys belong to the named output producer and are not accepted as approval keys. Pin keys through the verified deployment profile; canonical signatures cover the entire payload, while the existing tenant, application, digest, freshness and idempotency checks remain mandatory. Install cryptography through the approved dependency lock. An independent read-only verifier checks signatures and serving-state readback; an executor-signed receipt alone does not prove authorized approval or successful recovery. Replay payload/key tampering, wrong scope, expired approval and replayed decisions.

AID-R-005-G004

Pre-promotion boundary

Validate only the newly built candidate index. The recovery authority signs the exact case population and validity window; the candidate builder and quarantine authority separately sign the candidate manifest and tombstones. A fixed, root-owned bootstrap key authenticates the runtime profile, trust paths, mTLS identity, query endpoint, byte limits, deadlines, and safety ceilings.

Sign the exact validation population

{
  "schema_version": "aidefend.r005.candidate-validation.v2",
  "control": "AID-R-005",
  "tenant_id": "tenant-acme",
  "application_id": "benefits-search",
  "incident_id": "INC-EXAMPLE",
  "issued_at": "2026-07-20T18:00:00Z",
  "expires_at": "2026-07-20T19:00:00Z",
  "maximum_candidate_age_seconds": 3600,
  "maximum_future_skew_seconds": 30,
  "candidate_index": "rag-v43-candidate",
  "candidate_manifest_sha256": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
  "tombstone_set_sha256": "123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef0",
  "query_endpoint": "https://candidate-query.internal.example/v1/query",
  "cases": [
    {"case_id":"attack-001","case_class":"incident_attack","query":"incident reproducer text","top_k":10,"required_chunk_ids":[],"forbidden_chunk_ids":["poisoned-chunk-88421"]},
    {"case_id":"clean-001","case_class":"clean_utility","query":"approved leave policy","top_k":10,"required_chunk_ids":["trusted-leave-policy-v7"],"forbidden_chunk_ids":[]}
  ]
}

Replay and publish create-once evidence

set -euo pipefail
umask 077
python recovery/validate_candidate_index.py \
  --validation-manifest recovery/candidate-validation.json \
  --validation-bundle recovery/candidate-validation.json.sigstore.json \
  --candidate-manifest recovery/candidate-manifest.json \
  --candidate-bundle recovery/candidate-manifest.json.sigstore.json \
  --tombstones recovery/vector-tombstones.json \
  --tombstone-bundle recovery/vector-tombstones.json.sigstore.json \
  --output evidence/candidate-validation-result.json

# File: recovery/validate_candidate_index.py
from __future__ import annotations

import argparse
import hashlib
import json
import math
import os
import stat
import subprocess
import tempfile
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
from urllib.parse import urlsplit

import requests

PROFILE = Path("/etc/aidefend/r005/candidate-validation-profile.json")
PROFILE_BUNDLE = Path(str(PROFILE) + ".sigstore.json")
BOOTSTRAP_KEY = Path("/etc/aidefend/r005/bootstrap-root.pub")
BOOTSTRAP_VERIFICATION_DEADLINE_CEILING_SECONDS = 30.0
# Compiled process-safety ceilings bound bootstrap verification before authenticated policy is available.
BOOTSTRAP_CAPTURE_CEILING = 1024 * 1024
SHA = set("0123456789abcdef")
PROFILE_FIELDS = {"schema","version","query_endpoint","connect_timeout_seconds","read_timeout_seconds","signature_timeout_seconds","maximum_top_k","maximum_case_count","maximum_query_characters","maximum_response_bytes","maximum_signed_document_bytes","maximum_allowed_candidate_age_seconds","maximum_allowed_future_skew_seconds","client_cert","client_key","ca_file","validator_kms_key_uri","trust"}
TRUST_ROLES = {"recovery_authority","candidate_builder","quarantine_authority","candidate_validator"}
TRUST_FIELDS = {"path","sha256"}
VALIDATION_FIELDS = {"schema_version","control","tenant_id","application_id","incident_id","issued_at","expires_at","maximum_candidate_age_seconds","maximum_future_skew_seconds","candidate_index","candidate_manifest_sha256","tombstone_set_sha256","query_endpoint","cases"}
CANDIDATE_FIELDS = {"schema","tenant_id","application_id","candidate_collection","built_at","source_inventory_sha256","embedding_model_manifest_sha256","tombstone_set_digest","build_policy_sha256","point_count","point_population_sha256","points"}
POINT_FIELDS = {"point_id","vector_sha256","source_uri","content_sha256","source_inventory_sha256","content_scan_receipt","tombstone_set_digest","candidate_collection"}
CASE_FIELDS = {"case_id","case_class","query","top_k","required_chunk_ids","forbidden_chunk_ids"}
RESPONSE_FIELDS = {"candidate_index","candidate_manifest_sha256","case_id","chunk_ids"}

def strict_object(raw: bytes, label: str) -> dict[str, Any]:
    def pairs(items: list[tuple[str, Any]]) -> dict[str, Any]:
        value: dict[str, Any] = {}
        for key, item in items:
            if key in value: raise ValueError(f"{label}: duplicate key {key}")
            value[key] = item
        return value
    def constant(value: str) -> None: raise ValueError(f"{label}: non-finite {value}")
    value = json.loads(raw.decode("utf-8", "strict"), object_pairs_hook=pairs, parse_constant=constant)
    if not isinstance(value, dict): raise ValueError(f"{label} must be one object")
    return value

def digest(value: object, label: str) -> str:
    if not isinstance(value, str) or len(value) != 64 or set(value) - SHA or set(value) == {"0"}: raise ValueError(f"{label} must be a nonzero SHA-256")
    return value

def number(value: object, label: str, *, integer: bool = False, zero: bool = False) -> float | int:
    if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(float(value)) or value < 0 or (value == 0 and not zero) or (integer and not isinstance(value, int)): raise ValueError(f"invalid {label}")
    return value

def trusted_file(path: Path, label: str) -> bytes:
    info = path.lstat()
    if path.is_symlink() or not stat.S_ISREG(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o022: raise RuntimeError(f"{label} is not a fixed root-owned regular file")
    return path.read_bytes()

def limited(path: Path, maximum: int) -> bytes:
    with path.open("rb") as handle: raw = handle.read(maximum + 1)
    if not raw or len(raw) > maximum: raise ValueError(f"{path} is empty or exceeds the authenticated bound")
    return raw

def private(path: Path, raw: bytes) -> None:
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)
    fd = os.open(path, flags, 0o400)
    with os.fdopen(fd, "wb") as handle:
        handle.write(raw); handle.flush(); os.fsync(handle.fileno())

def verified(path: Path, bundle: Path, key: Path, *, timeout: float, maximum: int, label: str, trusted_key: bool = True) -> bytes:
    key_raw = trusted_file(key, f"{label} key") if trusted_key else limited(key, maximum)
    raw, bundle_raw = limited(path, maximum), limited(bundle, maximum)
    with tempfile.TemporaryDirectory(prefix="r005-verify-") as directory:
        root = Path(directory); os.chmod(root, 0o700)
        payload_copy, bundle_copy, key_copy = root / "payload", root / "payload.sigstore.json", root / "key.pub"
        private(payload_copy, raw); private(bundle_copy, bundle_raw); private(key_copy, key_raw)
        subprocess.run(["cosign","verify-blob","--key",str(key_copy),"--bundle",str(bundle_copy),str(payload_copy)], check=True, capture_output=True, text=True, timeout=timeout)
        if limited(payload_copy, maximum) != raw or limited(bundle_copy, maximum) != bundle_raw or limited(key_copy, maximum) != key_raw: raise RuntimeError(f"{label} verified snapshot changed")
    return raw

def load_profile() -> tuple[dict[str, Any], str]:
    raw = verified(PROFILE, PROFILE_BUNDLE, BOOTSTRAP_KEY, timeout=BOOTSTRAP_VERIFICATION_DEADLINE_CEILING_SECONDS, maximum=BOOTSTRAP_CAPTURE_CEILING, label="runtime profile")
    profile = strict_object(raw, "runtime profile")
    if set(profile) != PROFILE_FIELDS or profile.get("schema") != "aidefend.r005.candidate-validation-profile.v1" or not isinstance(profile["version"], str) or not profile["version"]: raise ValueError("runtime profile schema differs")
    for field in ("connect_timeout_seconds","read_timeout_seconds","signature_timeout_seconds"): number(profile[field], field)
    for field in ("maximum_top_k","maximum_case_count","maximum_query_characters","maximum_response_bytes","maximum_signed_document_bytes","maximum_allowed_candidate_age_seconds","maximum_allowed_future_skew_seconds"): number(profile[field], field, integer=True, zero=field.endswith("future_skew_seconds"))
    if profile["signature_timeout_seconds"] > 120 or profile["maximum_signed_document_bytes"] > 16 * 1024 * 1024: raise ValueError("runtime profile exceeds compiled safety ceilings")
    if not isinstance(profile["trust"], dict) or set(profile["trust"]) != TRUST_ROLES: raise ValueError("runtime trust roles differ")
    for role, entry in profile["trust"].items():
        if not isinstance(entry, dict) or set(entry) != TRUST_FIELDS: raise ValueError(f"{role} trust entry differs")
        raw_key = trusted_file(Path(entry["path"]), role)
        if hashlib.sha256(raw_key).hexdigest() != digest(entry["sha256"], f"{role} key digest"): raise RuntimeError(f"{role} key digest differs")
    for field in ("client_cert","client_key","ca_file"): trusted_file(Path(profile[field]), field)
    return profile, hashlib.sha256(raw).hexdigest()

def utc(value: object, label: str) -> datetime:
    if not isinstance(value, str): raise ValueError(f"{label} must be RFC 3339")
    try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError as error: raise ValueError(f"{label} must be RFC 3339") from error
    if parsed.tzinfo is None: raise ValueError(f"{label} lacks timezone")
    return parsed.astimezone(timezone.utc)

def exact_ids(value: object, label: str) -> list[str]:
    if not isinstance(value, list) or value != sorted(set(value)) or any(not isinstance(item, str) or not item or "*" in item for item in value): raise ValueError(f"{label} must be a sorted unique exact-ID list")
    return value

def parse_candidate(raw: bytes, validation: dict[str, Any], now: datetime) -> tuple[dict[str, Any], set[str]]:
    value = strict_object(raw, "candidate manifest")
    if set(value) != CANDIDATE_FIELDS or value.get("schema") != "aidefend.r005.candidate-manifest.v1": raise ValueError("candidate manifest schema differs")
    if (value["tenant_id"], value["application_id"], value["candidate_collection"]) != (validation["tenant_id"], validation["application_id"], validation["candidate_index"]): raise ValueError("candidate scope differs")
    rows = value["points"]
    count = value["point_count"]
    if isinstance(count, bool) or not isinstance(count, int) or count < 1 or not isinstance(rows, list) or len(rows) != count: raise ValueError("candidate point count differs")
    ids: list[str] = []
    for row in rows:
        if not isinstance(row, dict) or set(row) != POINT_FIELDS: raise ValueError("candidate point schema differs")
        point_id = row["point_id"]
        if not isinstance(point_id, str) or not point_id or len(point_id) > 255 or any(ord(char) < 32 for char in point_id): raise ValueError("candidate point ID is invalid")
        ids.append(point_id)
        for field in ("vector_sha256","content_sha256","source_inventory_sha256","content_scan_receipt","tombstone_set_digest"): digest(row[field], f"point {field}")
        if row["candidate_collection"] != value["candidate_collection"] or row["source_inventory_sha256"] != value["source_inventory_sha256"] or row["tombstone_set_digest"] != value["tombstone_set_digest"]: raise ValueError("candidate point binding differs")
    if ids != sorted(set(ids)) or hashlib.sha256(json.dumps(ids, sort_keys=True, separators=(",", ":")).encode()).hexdigest() != value["point_population_sha256"]: raise ValueError("candidate complete point population differs")
    built = utc(value["built_at"], "candidate built_at")
    age = (now - built).total_seconds()
    if age > validation["maximum_candidate_age_seconds"] or age < -validation["maximum_future_skew_seconds"]: raise ValueError("candidate manifest is stale or future-dated")
    return value, set(ids)

def parse_tombstones(raw: bytes, validation: dict[str, Any], candidate: dict[str, Any]) -> set[str]:
    envelope = strict_object(raw, "tombstone envelope")
    if set(envelope) != {"payload","signature_b64"} or not isinstance(envelope["payload"], dict): raise ValueError("tombstone envelope schema differs")
    payload = envelope["payload"]
    if payload.get("schema") != "aidefend.r005.vector-tombstones.v1" or (payload.get("tenant_id"), payload.get("application_id")) != (validation["tenant_id"], validation["application_id"]): raise ValueError("tombstone scope differs")
    logical_digest = hashlib.sha256(json.dumps(envelope, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False).encode("utf-8")).hexdigest()
    if logical_digest != candidate["tombstone_set_digest"]: raise ValueError("logical tombstone digest differs from candidate")
    ids = exact_ids(payload.get("point_ids"), "tombstone point_ids")
    entries = payload.get("point_content_digests")
    if not isinstance(entries, list) or [e.get("point_id") for e in entries if isinstance(e, dict)] != ids:
        raise ValueError("tombstone content population differs")
    denied_content = {digest(e.get("source_content_sha256"), "tombstone source content") for e in entries}
    if any(row["point_id"] in set(ids) for row in candidate["points"]):
        raise ValueError("candidate reintroduced a tombstoned point ID")
    if any(row["content_sha256"] in denied_content for row in candidate["points"]):
        raise ValueError("candidate reintroduced tombstoned content under a renamed ID")
    return set(ids)

def bounded_response(response: requests.Response, maximum: int) -> dict[str, Any]:
    if response.is_redirect or response.headers.get("Location") or response.headers.get("Content-Encoding", "identity").lower() != "identity": raise RuntimeError("redirected or compressed candidate response is forbidden")
    if response.headers.get("Content-Type", "").split(";",1)[0].strip().lower() != "application/json": raise RuntimeError("candidate response type differs")
    response.raise_for_status(); chunks: list[bytes] = []; total = 0
    for chunk in response.iter_content(65536):
        total += len(chunk)
        if total > maximum: raise RuntimeError("candidate response exceeds authenticated bound")
        chunks.append(chunk)
    value = strict_object(b"".join(chunks), "candidate response")
    if set(value) != RESPONSE_FIELDS: raise RuntimeError("candidate response schema differs")
    return value

def publish(output: Path, value: dict[str, Any], profile: dict[str, Any], profile_sha: str) -> str:
    bundle = Path(str(output) + ".sigstore.json")
    if output.exists() or bundle.exists(): raise FileExistsError("validation result or bundle already exists")
    if not output.parent.is_dir() or output.parent.is_symlink(): raise ValueError("output parent must be an existing real directory")
    raw = (json.dumps(value, sort_keys=True, separators=(",",":"), ensure_ascii=False, allow_nan=False) + "\n").encode()
    validator = profile["trust"]["candidate_validator"]
    with tempfile.TemporaryDirectory(prefix=".r005-publish-", dir=output.parent) as directory:
        root = Path(directory); os.chmod(root, 0o700)
        payload_copy, bundle_copy = root / "result.json", root / "result.json.sigstore.json"
        private(payload_copy, raw)
        subprocess.run(["cosign","sign-blob","--yes","--key",profile["validator_kms_key_uri"],"--bundle",str(bundle_copy),str(payload_copy)], check=True, capture_output=True, text=True, timeout=profile["signature_timeout_seconds"])
        key = Path(validator["path"])
        verified(payload_copy, bundle_copy, key, timeout=profile["signature_timeout_seconds"], maximum=profile["maximum_signed_document_bytes"], label="validation result")
        os.link(bundle_copy, bundle, follow_symlinks=False)
        os.link(payload_copy, output, follow_symlinks=False)  # payload is the commit marker
    return hashlib.sha256(raw).hexdigest()

def main() -> None:
    parser = argparse.ArgumentParser()
    for name in ("validation-manifest","validation-bundle","candidate-manifest","candidate-bundle","tombstones","tombstone-bundle"): parser.add_argument("--" + name, type=Path, required=True)
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args(); profile, profile_sha = load_profile(); maximum = profile["maximum_signed_document_bytes"]
    trust = profile["trust"]
    validation_raw = verified(args.validation_manifest, args.validation_bundle, Path(trust["recovery_authority"]["path"]), timeout=profile["signature_timeout_seconds"], maximum=maximum, label="validation policy")
    candidate_raw = verified(args.candidate_manifest, args.candidate_bundle, Path(trust["candidate_builder"]["path"]), timeout=profile["signature_timeout_seconds"], maximum=maximum, label="candidate manifest")
    tombstone_raw = verified(args.tombstones, args.tombstone_bundle, Path(trust["quarantine_authority"]["path"]), timeout=profile["signature_timeout_seconds"], maximum=maximum, label="tombstones")
    validation = strict_object(validation_raw, "validation policy")
    if set(validation) != VALIDATION_FIELDS or validation.get("schema_version") != "aidefend.r005.candidate-validation.v2" or validation.get("control") != "AID-R-005": raise ValueError("validation policy schema differs")
    for field in ("tenant_id","application_id","incident_id","candidate_index"):
        if not isinstance(validation[field], str) or not validation[field] or len(validation[field]) > 255 or any(ord(char) < 32 for char in validation[field]): raise ValueError(f"invalid {field}")
    now = datetime.now(timezone.utc)
    future = number(validation["maximum_future_skew_seconds"], "maximum_future_skew_seconds", zero=True)
    age = number(validation["maximum_candidate_age_seconds"], "maximum_candidate_age_seconds")
    if future > profile["maximum_allowed_future_skew_seconds"] or age > profile["maximum_allowed_candidate_age_seconds"]: raise ValueError("validation time bounds exceed authenticated runtime ceilings")
    issued, expires = utc(validation["issued_at"], "issued_at"), utc(validation["expires_at"], "expires_at")
    if issued > now + timedelta(seconds=future) or expires <= now or expires <= issued: raise ValueError("validation policy is future-issued, expired, or inverted")
    if hashlib.sha256(candidate_raw).hexdigest() != digest(validation["candidate_manifest_sha256"], "candidate manifest") or hashlib.sha256(tombstone_raw).hexdigest() != digest(validation["tombstone_set_sha256"], "tombstones"): raise ValueError("validation input digest differs")
    candidate, allowed_ids = parse_candidate(candidate_raw, validation, now)
    tombstones = parse_tombstones(tombstone_raw, validation, candidate)
    if validation["query_endpoint"] != profile["query_endpoint"]: raise ValueError("query endpoint differs from authenticated runtime profile")
    endpoint = urlsplit(validation["query_endpoint"])
    if endpoint.scheme != "https" or endpoint.hostname is None or endpoint.port not in (None,443) or endpoint.username or endpoint.password or endpoint.query or endpoint.fragment: raise ValueError("query endpoint is not canonical HTTPS")
    cases = validation["cases"]
    if not isinstance(cases, list) or not cases or len(cases) > profile["maximum_case_count"]: raise ValueError("validation population is empty or exceeds authenticated bound")
    case_ids = [row.get("case_id") if isinstance(row,dict) else None for row in cases]
    if case_ids != sorted(set(case_ids)) or any(not isinstance(item,str) or not item for item in case_ids): raise ValueError("case IDs are not an exact sorted population")
    if {row.get("case_class") for row in cases} != {"clean_utility","incident_attack"}: raise ValueError("both validation classes are required")
    results = []
    with requests.Session() as session:
        session.trust_env = False
        for case in cases:
            if not isinstance(case,dict) or set(case) != CASE_FIELDS or case["case_class"] not in {"clean_utility","incident_attack"}: raise ValueError("case schema differs")
            if not isinstance(case["query"],str) or not case["query"] or len(case["query"]) > profile["maximum_query_characters"]: raise ValueError("query is invalid")
            number(case["top_k"], "top_k", integer=True)
            if case["top_k"] > profile["maximum_top_k"]: raise ValueError("top_k exceeds authenticated bound")
            required = set(exact_ids(case["required_chunk_ids"], "required IDs")); forbidden = set(exact_ids(case["forbidden_chunk_ids"], "forbidden IDs")) | tombstones
            if not required <= allowed_ids: raise ValueError("required ID is absent from the signed candidate population")
            if not required and not forbidden: raise ValueError("case has no assertion")
            with session.post(validation["query_endpoint"], headers={"Accept":"application/json","Accept-Encoding":"identity"}, json={"candidate_index":validation["candidate_index"],"candidate_manifest_sha256":validation["candidate_manifest_sha256"],"case_id":case["case_id"],"query":case["query"],"top_k":case["top_k"]}, cert=(profile["client_cert"],profile["client_key"]), verify=profile["ca_file"], timeout=(profile["connect_timeout_seconds"],profile["read_timeout_seconds"]), allow_redirects=False, stream=True) as response:
                observed = bounded_response(response, profile["maximum_response_bytes"])
            if (observed["candidate_index"],observed["candidate_manifest_sha256"],observed["case_id"]) != (validation["candidate_index"],validation["candidate_manifest_sha256"],case["case_id"]): raise RuntimeError("query response binding differs")
            returned_list = exact_ids(observed["chunk_ids"], "returned chunk IDs")
            if len(returned_list) > case["top_k"]: raise RuntimeError("query returned more than top_k")
            returned = set(returned_list)
            unknown, missing, prohibited = returned - allowed_ids, required - returned, forbidden & returned
            if unknown or missing or prohibited: raise RuntimeError(f"case {case['case_id']} failed: unknown={sorted(unknown)}, missing={sorted(missing)}, prohibited={sorted(prohibited)}")
            results.append({"case_id":case["case_id"],"case_class":case["case_class"],"returned_chunk_ids":returned_list,"result":"PASS"})
    output = {"schema_version":"aidefend.r005.candidate-validation-result.v2","outcome":"PASS","tenant_id":validation["tenant_id"],"application_id":validation["application_id"],"incident_id":validation["incident_id"],"candidate_index":validation["candidate_index"],"candidate_manifest_sha256":validation["candidate_manifest_sha256"],"candidate_point_population_sha256":candidate["point_population_sha256"],"tombstone_set_sha256":validation["tombstone_set_sha256"],"validation_policy_sha256":hashlib.sha256(validation_raw).hexdigest(),"runtime_profile_version":profile["version"],"runtime_profile_sha256":profile_sha,"evaluated_at":now.isoformat().replace("+00:00","Z"),"complete_case_count":len(cases),"results":results}
    print(json.dumps({"outcome":"PASS","result_sha256":publish(args.output,output,profile,profile_sha)},sort_keys=True))

if __name__ == "__main__": main()

Independent replay and promotion decision

Run the same immutable inputs under a separate validator identity. Promotion receives only two independently signed result payload/bundle pairs that bind the policy validity window, candidate build time, complete candidate point population, tombstone digest, exact case population, runtime profile, and every returned ID. A stale or future-dated input, missing or unknown candidate ID, required ID outside the manifest, tombstoned hit, redirect, response overflow, timeout, signature failure, or partial population prevents PASS. Runtime retrieval monitoring after promotion remains AID-D-014.

Separate authority verification from execution receipts

Use independently provisioned Ed25519 public keys for policy, scope, source inventory, model manifest, detector findings and human approval inputs. The executor cannot access those issuers' private keys. Receipt/tombstone signing keys belong to the named output producer and are not accepted as approval keys. Pin keys through the verified deployment profile; canonical signatures cover the entire payload, while the existing tenant, application, digest, freshness and idempotency checks remain mandatory. Install cryptography through the approved dependency lock. An independent read-only verifier checks signatures and serving-state readback; an executor-signed receipt alone does not prove authorized approval or successful recovery. Replay payload/key tampering, wrong scope, expired approval and replayed decisions.

AID-R-005-G005

Bind the finding, response policy, and exact recovery boundaries

The finding names the complete sorted point population, active index, alias, detector release/policy digest, event time, and risk score. The separately signed response policy binds the approved last-known-good index and manifest, exact recovery-control, vector-readback, and evidence origins, different credentials/trust stores, response threshold, freshness, approval mode, and operation deadline. Detection scores content; this method only orchestrates the existing quarantine and alias mechanisms and verifies their exact readback.

# File: recovery/retrieval_response.py
from __future__ import annotations

import argparse
import hashlib
import json
import math
import os
import re
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.parse import quote

from recovery.strict_io import canonical, exact_ids, exact_origin, signed_packet, verified_packet, make_session, request_json, verified_envelope

SAFE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,254}$")
HEX = re.compile(r"^[0-9a-f]{64}$")
INPUT_LIMIT = int(os.environ["R005_RESPONSE_MAXIMUM_INPUT_BYTES"])
RESPONSE_LIMIT = int(os.environ["R005_RESPONSE_MAXIMUM_RESPONSE_BYTES"])
CONNECT_TIMEOUT = float(os.environ["R005_RESPONSE_CONNECT_TIMEOUT_SECONDS"])
READ_TIMEOUT = float(os.environ["R005_RESPONSE_READ_TIMEOUT_SECONDS"])
POLL_INTERVAL = float(os.environ["R005_RESPONSE_POLL_INTERVAL_SECONDS"])
FUTURE_SKEW = float(os.environ["R005_RESPONSE_MAXIMUM_FUTURE_SKEW_SECONDS"])
if INPUT_LIMIT < 1 or RESPONSE_LIMIT < 1 or any(not math.isfinite(value) or value <= 0 for value in (CONNECT_TIMEOUT, READ_TIMEOUT, POLL_INTERVAL)) or not math.isfinite(FUTURE_SKEW) or FUTURE_SKEW < 0:
    raise RuntimeError("signed retrieval-response runtime profile is invalid")
TIMEOUT = (CONNECT_TIMEOUT, READ_TIMEOUT)
FINDING_FIELDS = {"schema", "event_id", "tenant_id", "application_id", "serving_alias", "active_index", "active_index_manifest_sha256", "chunk_ids", "risk_score", "detector_release", "detector_policy_digest", "observed_at"}
POLICY_FIELDS = {"schema", "tenant_id", "application_id", "maximum_chunk_count", "maximum_event_age_seconds", "response_threshold", "automation_mode", "required_approval_roles", "last_known_good_index", "last_known_good_manifest_sha256", "operation_timeout_seconds", "recovery_control_origin", "vector_readback_origin", "evidence_origin"}
APPROVAL_FIELDS = {"schema", "event_id", "target_index", "target_manifest_sha256", "approver_roles", "expires_at"}


def digest(value: object, name: str) -> str:
    if not isinstance(value, str) or HEX.fullmatch(value) is None or set(value) == {"0"}:
        raise ValueError(f"{name} must be a nonzero SHA-256 digest")
    return value


def finite(value: object, name: str, *, minimum: float | None = None) -> float:
    if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or (minimum is not None and value < minimum):
        raise ValueError(f"{name} is invalid")
    return float(value)


def timestamp(value: object, name: str) -> datetime:
    if not isinstance(value, str):
        raise ValueError(f"{name} timestamp is required")
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None:
        raise ValueError(f"{name} must include a timezone")
    return parsed.astimezone(timezone.utc)


def poll_operation(session, origin: str, operation_id: str, action: str, deadline: float) -> dict[str, Any]:
    if not SAFE.fullmatch(operation_id):
        raise RuntimeError("unsafe operation ID")
    path = "/v1/operations/" + quote(operation_id, safe="")
    expected_fields = ({"operation_id", "state", "action", "tenant_id", "application_id", "incident_id", "receipt_digest", "point_ids", "source_index"} if action == "quarantine" else {"operation_id", "state", "action", "tenant_id", "application_id", "incident_id", "receipt_digest", "alias", "before_index", "target_index", "target_manifest_sha256", "quarantine_receipt_digest"})
    while time.monotonic() < deadline:
        result = request_json(session, "GET", origin, path, timeout=TIMEOUT, maximum=RESPONSE_LIMIT)
        if set(result) != expected_fields or result["operation_id"] != operation_id or result["action"] != action:
            raise RuntimeError("operation readback schema or identity differs")
        if result["state"] == "succeeded":
            return result
        if result["state"] in {"failed", "cancelled"}:
            raise RuntimeError(f"{action} operation ended in {result['state']}")
        if result["state"] not in {"pending", "running"}:
            raise RuntimeError("unknown operation state")
        time.sleep(POLL_INTERVAL)
    raise TimeoutError(f"{action} operation exceeded the signed deadline")


def start_operation(session, origin: str, path: str, body: dict[str, Any], idempotency_key: str, action: str, deadline: float) -> dict[str, Any]:
    started = request_json(session, "POST", origin, path, timeout=TIMEOUT, maximum=RESPONSE_LIMIT, body=body, headers={"Idempotency-Key": idempotency_key})
    if set(started) != {"operation_id"} or not isinstance(started["operation_id"], str):
        raise RuntimeError(f"{action} start response differs")
    return poll_operation(session, origin, started["operation_id"], action, deadline)


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--finding", type=Path, required=True)
    parser.add_argument("--policy", type=Path, required=True)
    parser.add_argument("--approval", type=Path)
    args = parser.parse_args()
    finding = verified_envelope(args.finding, INPUT_LIMIT, "aidefend.r005.retrieval-finding.v1", "RETRIEVAL_FINDING_VERIFY_KEY_FILE")
    policy = verified_envelope(args.policy, INPUT_LIMIT, "aidefend.r005.response-policy.v1", "RECOVERY_POLICY_VERIFY_KEY_FILE")
    approval = verified_envelope(args.approval, INPUT_LIMIT, "aidefend.r005.response-approval.v1", "RECOVERY_APPROVAL_VERIFY_KEY_FILE") if args.approval else None
    if set(finding) != FINDING_FIELDS or set(policy) != POLICY_FIELDS or (approval is not None and set(approval) != APPROVAL_FIELDS):
        raise ValueError("finding, policy, or approval schema differs")
    for name in ("event_id", "tenant_id", "application_id", "serving_alias", "active_index"):
        if not isinstance(finding[name], str) or SAFE.fullmatch(finding[name]) is None:
            raise ValueError(f"unsafe finding {name}")
    if (finding["tenant_id"], finding["application_id"]) != (policy["tenant_id"], policy["application_id"]):
        raise RuntimeError("finding-to-policy scope differs")
    digest(finding["active_index_manifest_sha256"], "active index manifest")
    digest(finding["detector_policy_digest"], "detector policy")
    digest(policy["last_known_good_manifest_sha256"], "last-known-good manifest")
    maximum_chunks = int(finite(policy["maximum_chunk_count"], "maximum chunk count", minimum=1))
    if isinstance(policy["maximum_chunk_count"], bool) or policy["maximum_chunk_count"] != maximum_chunks:
        raise ValueError("maximum chunk count must be an integer")
    point_ids = exact_ids(finding["chunk_ids"], "finding chunk_ids", maximum_chunks)
    risk = finite(finding["risk_score"], "finding risk score")
    threshold = finite(policy["response_threshold"], "response threshold")
    age = (datetime.now(timezone.utc) - timestamp(finding["observed_at"], "observed_at")).total_seconds()
    if age < -FUTURE_SKEW:
        raise RuntimeError("finding timestamp is in the future")
    if age > finite(policy["maximum_event_age_seconds"], "maximum event age", minimum=0):
        outcome = {"outcome": "INSUFFICIENT_DATA", "reason": "finding_stale"}
    elif risk < threshold:
        outcome = {"outcome": "NOT_APPLICABLE", "reason": "policy_threshold_not_crossed"}
    else:
        target = policy["last_known_good_index"]
        if not isinstance(target, str) or SAFE.fullmatch(target) is None or target == finding["active_index"]:
            outcome = {"outcome": "INSUFFICIENT_DATA", "reason": "valid_rollback_target_missing"}
        else:
            roles = policy["required_approval_roles"]
            if not isinstance(roles, list) or roles != sorted(set(roles)) or any(not isinstance(role, str) or not role for role in roles):
                raise ValueError("required approval roles are not an exact population")
            if policy["automation_mode"] == "approval_required":
                if approval is None:
                    outcome = {"outcome": "INSUFFICIENT_DATA", "reason": "required_approval_missing"}
                else:
                    if approval["event_id"] != finding["event_id"] or approval["target_index"] != target or approval["target_manifest_sha256"] != policy["last_known_good_manifest_sha256"] or timestamp(approval["expires_at"], "approval expiry") <= datetime.now(timezone.utc):
                        raise RuntimeError("approval binding or freshness differs")
                    approver_roles = approval["approver_roles"]
                    if not isinstance(approver_roles, list) or approver_roles != sorted(set(approver_roles)) or not set(roles).issubset(approver_roles):
                        outcome = {"outcome": "INSUFFICIENT_DATA", "reason": "required_approval_roles_missing"}
                    else:
                        outcome = None
            elif policy["automation_mode"] == "automatic":
                outcome = None
            else:
                raise ValueError("unknown automation mode")
            if outcome is None:
                origins = {name: exact_origin(policy[name], name) for name in ("recovery_control_origin", "vector_readback_origin", "evidence_origin")}
                for name, environment in (("recovery_control_origin", "RECOVERY_CONTROL_ORIGIN"), ("vector_readback_origin", "VECTOR_READBACK_ORIGIN"), ("evidence_origin", "EVIDENCE_ORIGIN")):
                    if origins[name] != exact_origin(os.environ[environment], environment):
                        raise RuntimeError(f"{name} is not locally allowlisted")
                if len(set(origins.values())) != 3:
                    raise RuntimeError("recovery, readback, and evidence origins must differ")
                recovery = make_session(cert=os.environ["RECOVERY_CONTROL_CLIENT_CERT"], key=os.environ["RECOVERY_CONTROL_CLIENT_KEY"], ca=os.environ["RECOVERY_CONTROL_CA_FILE"], token_file=os.environ["RECOVERY_CONTROL_TOKEN_FILE"])
                readback_session = make_session(cert=os.environ["VECTOR_READBACK_CLIENT_CERT"], key=os.environ["VECTOR_READBACK_CLIENT_KEY"], ca=os.environ["VECTOR_READBACK_CA_FILE"], token_file=os.environ["VECTOR_READBACK_TOKEN_FILE"])
                evidence = make_session(cert=os.environ["EVIDENCE_CLIENT_CERT"], key=os.environ["EVIDENCE_CLIENT_KEY"], ca=os.environ["EVIDENCE_CA_FILE"])
                if len({recovery.cert, readback_session.cert, evidence.cert}) != 3:
                    raise RuntimeError("recovery, readback, and evidence identities must differ")
                operation_seconds = finite(policy["operation_timeout_seconds"], "operation timeout", minimum=0.001)
                deadline = time.monotonic() + operation_seconds
                finding_digest = hashlib.sha256(canonical(finding)).hexdigest()
                quarantine = start_operation(recovery, policy["recovery_control_origin"], "/v1/vector-quarantine-jobs", {"tenant_id": finding["tenant_id"], "application_id": finding["application_id"], "source_index": finding["active_index"], "point_ids": point_ids, "incident_id": finding["event_id"], "expected_finding_digest": finding_digest}, finding["event_id"] + ":quarantine", "quarantine", deadline)
                if quarantine["tenant_id"] != finding["tenant_id"] or quarantine["application_id"] != finding["application_id"] or quarantine["incident_id"] != finding["event_id"] or quarantine["source_index"] != finding["active_index"] or quarantine["point_ids"] != point_ids:
                    raise RuntimeError("quarantine operation population readback differs")
                rollback = start_operation(recovery, policy["recovery_control_origin"], "/v1/vector-alias-rollbacks", {"tenant_id": finding["tenant_id"], "application_id": finding["application_id"], "alias": finding["serving_alias"], "expected_current_index": finding["active_index"], "target_index": target, "target_manifest_sha256": policy["last_known_good_manifest_sha256"], "incident_id": finding["event_id"], "quarantine_receipt_digest": quarantine["receipt_digest"]}, finding["event_id"] + ":rollback", "rollback", deadline)
                if rollback["alias"] != finding["serving_alias"] or rollback["before_index"] != finding["active_index"] or rollback["target_index"] != target or rollback["target_manifest_sha256"] != policy["last_known_good_manifest_sha256"] or rollback["quarantine_receipt_digest"] != quarantine["receipt_digest"]:
                    raise RuntimeError("alias rollback readback differs")
                effective = request_json(readback_session, "GET", policy["vector_readback_origin"], "/v1/aliases/" + quote(finding["serving_alias"], safe=""), timeout=TIMEOUT, maximum=RESPONSE_LIMIT)
                if set(effective) != {"alias", "target_index", "index_manifest_sha256", "quarantined_point_ids"} or effective["alias"] != finding["serving_alias"] or effective["target_index"] != target or effective["index_manifest_sha256"] != policy["last_known_good_manifest_sha256"] or effective["quarantined_point_ids"] != point_ids:
                    outcome = {"outcome": "FAIL", "reason": "direct_readback_differs", "effective_state_sha256": hashlib.sha256(canonical(effective)).hexdigest()}
                else:
                    outcome = {"outcome": "PASS", "reason": "quarantine_and_rollback_verified", "event_id": finding["event_id"], "quarantine_receipt_digest": quarantine["receipt_digest"], "rollback_receipt_digest": rollback["receipt_digest"], "effective_state_sha256": hashlib.sha256(canonical(effective)).hexdigest(), "target_index": target, "point_population_sha256": hashlib.sha256(canonical(point_ids)).hexdigest()}
    payload = {"schema": "aidefend.r005.retrieval-response-result.v1", "finding_event_id": finding["event_id"], "finding_sha256": hashlib.sha256(canonical(finding)).hexdigest(), "response_policy_sha256": hashlib.sha256(canonical(policy)).hexdigest(), "completed_at": datetime.now(timezone.utc).isoformat(), **outcome}
    envelope = signed_packet(payload, "RECOVERY_RESULT_SIGNING_KEY_FILE")
    result_digest = hashlib.sha256(canonical(envelope)).hexdigest()
    evidence_session = locals().get("evidence") or make_session(cert=os.environ["EVIDENCE_CLIENT_CERT"], key=os.environ["EVIDENCE_CLIENT_KEY"], ca=os.environ["EVIDENCE_CA_FILE"])
    evidence_origin = exact_origin(policy["evidence_origin"], "evidence")
    if evidence_origin != exact_origin(os.environ["EVIDENCE_ORIGIN"], "allowlisted evidence"):
        raise RuntimeError("evidence origin is not locally allowlisted")
    receipt = request_json(evidence_session, "POST", evidence_origin, "/v1/vector-recovery-results", timeout=TIMEOUT, maximum=RESPONSE_LIMIT, body=envelope, headers={"Idempotency-Key": result_digest})
    if set(receipt) != {"receipt_digest", "outcome"} or receipt["receipt_digest"] != result_digest or receipt["outcome"] != outcome["outcome"]:
        raise RuntimeError("evidence-store result readback differs")
    print(json.dumps({"outcome": outcome["outcome"], "receipt_digest": result_digest}, sort_keys=True))
    return 0 if outcome["outcome"] in {"PASS", "NOT_APPLICABLE"} else 2


if __name__ == "__main__":
    raise SystemExit(main())

Test a duplicate or missing point, NaN score, stale/future finding, wrong origin, redirect, response overflow, unquoted operation/alias identifier, approval mismatch, partial quarantine, alias manifest mismatch, credential reuse, and evidence receipt mismatch. The recovery and readback tokens are read into their own Python sessions and never appear in subprocess arguments. PASS requires the exact point population, alias, target index, target-manifest digest, quarantine receipt, rollback receipt, and direct serving readback to agree.

Separate authority verification from execution receipts

Use independently provisioned Ed25519 public keys for policy, scope, source inventory, model manifest, detector findings and human approval inputs. The executor cannot access those issuers' private keys. Receipt/tombstone signing keys belong to the named output producer and are not accepted as approval keys. Pin keys through the verified deployment profile; canonical signatures cover the entire payload, while the existing tenant, application, digest, freshness and idempotency checks remain mandatory. Install cryptography through the approved dependency lock. An independent read-only verifier checks signatures and serving-state readback; an executor-signed receipt alone does not prove authorized approval or successful recovery. Replay payload/key tampering, wrong scope, expired approval and replayed decisions.

Open-source tools

• Qdrant
• Milvus
• FAISS
• OpenSearch (Apache-2.0)
• Elasticsearch 8.16+ (AGPL-3.0 option)
• GX Core
• Open Policy Agent (OPA)
• Cosign
• GnuPG
• sha256sum (GNU Coreutils)

Commercial / hosted tools

• Pinecone
• Weaviate Shared Cloud
• Weaviate Dedicated Cloud
• Amazon OpenSearch Service
• Amazon S3 Object Lock

Threat mappings

MITRE ATLAS

• AML.T0099 AI Agent Tool Data Poisoning
• AML.T0059 Erode Dataset Integrity
• AML.T0070 RAG Poisoning
• AML.T0071 False RAG Entry Injection
• AML.T0051 LLM Prompt Injection (rollback removes injected content from vector index)
• AML.T0080.000 AI Agent Context Poisoning: Memory (rollback restores poisoned retrieval memory context)
• AML.T0066 Retrieval Content Crafting

MAESTRO

• Data Poisoning (L2)
• Data Tampering (L2)
• Compromised RAG Pipelines (L2)

OWASP LLM Top 10 2026

• LLM01:2026 Prompt Injection (rollback removes indirect prompt injections embedded in a RAG index)
• LLM05:2026 Data and Model Poisoning
• LLM09:2026 Vector and Embedding Weaknesses

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI01:2026 Agent Goal Hijack (rollback removes RAG content that could redirect agent goals)
• ASI06:2026 Memory & Context Poisoning
• ASI08:2026 Cascading Failures (rapid rollback prevents poisoned RAG from cascading across agents)

NIST Adversarial Machine Learning 2025

• NISTAML.015 Indirect Prompt Injection
• NISTAML.027 Misaligned Outputs (rollback removes RAG content causing misaligned LLM outputs)

Cisco Integrated AI Security and Safety Framework

• AISubtech-6.1.1 Knowledge Base Poisoning
• AITech-1.2 Indirect Prompt Injection (Removes the identified indirect-injection chunks from retrieval; other input channels remain separate)
• AITech-7.3 Data Source Abuse and Manipulation

Google Secure AI Framework 2.0 - Risks

• PIJ: Prompt Injection (rollback removes indirect prompt injections embedded in RAG index)
• IMO: Insecure Model Output (rollback removes RAG content causing insecure model outputs)

Databricks AI Security Framework 3.0

• Agents - Core 13.1: Memory Poisoning (rollback removes poisoned RAG content from agent retrieval)
• Agents - Core 13.5: Cascading Hallucination Attacks
• Agents - Tools MCP Server 13.24: Context Spoofing and Manipulation (rollback removes spoofed/manipulated context from retrieval index)