Scope boundary

Owns semantic relevance anomaly findings by independently re-scoring the complete live query-to-chunk top-k with a pinned model separate from the retriever's primary score.

AID-D-014.001: AID-D-014.002 detects off-topic or unexpectedly weak query-to-chunk relevance; AID-D-014.001 detects malicious instructions and payloads inside retrieved content. Relevance and content safety are independent properties.

AID-D-014.003: AID-D-014.002 evaluates semantic fit between the query and each chunk; AID-D-014.003 evaluates source identity, ordered population completeness, diversity, concentration, duplicates, and registry membership. A highly relevant result can still come from an unapproved or over-concentrated source.

Implementation guidance

• AID-D-014.002-G001: Score query-to-chunk semantic relevance and log low-relevance retrieval anomalies.

AID-D-014.002-G001

Pin the runtime inputs

Measure relevance over the complete signed top-k for every query rather than relying on a wrapper's default activation or truncation behavior. Install Cosign at /usr/local/bin/cosign and keep the bootstrap public key at /etc/aidefend/d014/trust/runtime-profile-root.pub. The key, profile, bundle, and every parent directory must be root-owned and non-writable by group or other users. The deployment-only provisioner below computes lowercase SHA-256 values from the exact deployed role-key bytes and refuses to overwrite an activated profile. Supply its version and validity interval through controlled release automation, then sign the exact output with the offline bootstrap role. An evaluator caller cannot select keys, paths, evaluation time, limits, or the result directory.

from __future__ import annotations

# File: detection/provision_d014002_g001_relevance_profile.py
from datetime import datetime, timezone
import hashlib
import json
from pathlib import Path
import re
import stat
import sys

from detection.d014_verified_artifacts import (
    BOOTSTRAP_DIRECTORY, HARD_KEY_CAPTURE_CEILING, read_stable, write_exclusive,
)


CONTROL = "AID-D-014.002-G001"
PROFILE_NAME = "d014002-g001-relevance"
ARTIFACTS = {
    "snapshot": {
        "payload": "/var/lib/aidefend/d014/input/d014002-g001-relevance-snapshot.jsonl",
        "bundle": "/var/lib/aidefend/d014/input/d014002-g001-relevance-snapshot.jsonl.sigstore.json"
    },
    "receipt": {
        "payload": "/var/lib/aidefend/d014/input/d014002-g001-relevance-receipt.json",
        "bundle": "/var/lib/aidefend/d014/input/d014002-g001-relevance-receipt.json.sigstore.json"
    },
    "policy": {
        "payload": "/etc/aidefend/d014/input/d014002-g001-relevance-policy.json",
        "bundle": "/etc/aidefend/d014/input/d014002-g001-relevance-policy.json.sigstore.json"
    },
    "model": {
        "payload": "/var/lib/aidefend/d014/input/d014002-g001-relevance-model.tar",
        "bundle": "/var/lib/aidefend/d014/input/d014002-g001-relevance-model.tar.sigstore.json"
    }
}
KEY_PATHS = {
    "snapshot": "/etc/aidefend/d014/trust/d014002-g001-relevance-snapshot.pub",
    "receipt": "/etc/aidefend/d014/trust/d014002-g001-relevance-receipt.pub",
    "policy": "/etc/aidefend/d014/trust/d014002-g001-relevance-policy.pub",
    "model": "/etc/aidefend/d014/trust/d014002-g001-relevance-model.pub",
    "result": "/etc/aidefend/d014/trust/d014002-g001-relevance-result.pub"
}
RESULT_BASE = "/var/lib/aidefend/d014/results/d014002-g001-relevance"
PROFILE_VERSION = re.compile(r"^[A-Za-z0-9._-]{1,128}$")


def parse_time(value: str) -> datetime:
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.utcoffset() is None:
        raise ValueError("profile validity timestamp lacks timezone")
    return parsed.astimezone(timezone.utc)


def utc_text(value: datetime) -> str:
    return value.isoformat(timespec="seconds").replace("+00:00", "Z")


def main() -> int:
    if len(sys.argv) != 4 or not PROFILE_VERSION.fullmatch(sys.argv[1]):
        raise SystemExit(
            "usage: provision_profile PROFILE_VERSION VALID_FROM VALID_UNTIL"
        )
    valid_from = parse_time(sys.argv[2])
    valid_until = parse_time(sys.argv[3])
    if valid_from >= valid_until:
        raise ValueError("profile validity interval is empty or reversed")
    directory = BOOTSTRAP_DIRECTORY.lstat()
    if (
        not stat.S_ISDIR(directory.st_mode)
        or BOOTSTRAP_DIRECTORY.is_symlink()
        or directory.st_uid != 0
        or directory.st_mode & 0o022
    ):
        raise PermissionError("profile directory is not root-owned and non-writable")
    keys = {}
    for role, value in KEY_PATHS.items():
        raw = read_stable(
            Path(value), maximum_bytes=HARD_KEY_CAPTURE_CEILING, trust_root=True,
        )
        keys[role] = {"path": value, "sha256": hashlib.sha256(raw).hexdigest()}
    if len({spec["sha256"] for spec in keys.values()}) != len(keys):
        raise ValueError("verification roles do not use distinct public keys")
    profile = {
        "schema_version": "aidefend.d014-runtime-profile.v1",
        "profile_version": sys.argv[1],
        "control": CONTROL,
        "valid_from": utc_text(valid_from),
        "valid_until": utc_text(valid_until),
        "artifacts": ARTIFACTS,
        "verification_keys": keys,
        "result_base": RESULT_BASE,
    }
    output = BOOTSTRAP_DIRECTORY / (PROFILE_NAME + ".json")
    raw = json.dumps(
        profile, sort_keys=True, indent=2, allow_nan=False,
    ).encode("utf-8") + b"\n"
    write_exclusive(output, raw, mode=0o640)
    print(output)
    return 0


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

Each JSONL line binds a query ID/class/text to one ranked chunk ID/text. The receipt binds exact bytes, query and pair counts, asset, completion time, and completeness. The policy binds top-k by query class, model tar/revision, exact labels, relevance label, sigmoid or softmax transform, class thresholds, sequence/window limits, and freshness.

Capture and verify exact bytes

The shared module below opens every input with no-follow and stable-descriptor checks, copies payload, bundle, and pinned key bytes into a private directory, and invokes cosign verify-blob --bundle only on those copies. Its fixed hard ceilings protect the evaluator process; method thresholds, freshness windows, and population limits remain signed policy values. The signed model path uses a deterministic uncompressed tar, two streaming inspection passes, regular files only, and safetensors; it never calls extractall or enables remote model code.

from __future__ import annotations

# File: detection/d014_verified_artifacts.py

from contextlib import contextmanager
from dataclasses import dataclass
import ctypes
from datetime import datetime, timezone
import errno
import hashlib
import hmac
from io import BytesIO
import json
import os
from pathlib import Path, PurePosixPath
import re
import shutil
import stat
import subprocess
import tarfile
import tempfile


COSIGN = "/usr/local/bin/cosign"
BOOTSTRAP_DIRECTORY = Path("/etc/aidefend/d014/profiles")
BOOTSTRAP_KEY = Path("/etc/aidefend/d014/trust/runtime-profile-root.pub")
# These are non-policy process-safety ceilings. Signed policies and callers
# may choose lower limits but cannot raise these values.
COSIGN_DEADLINE_CEILING_SECONDS = 15.0
HARD_ARTIFACT_CAPTURE_CEILING = 256 * 1024 * 1024
HARD_METADATA_CAPTURE_CEILING = 8 * 1024 * 1024
HARD_KEY_CAPTURE_CEILING = 1024 * 1024
HARD_MODEL_MEMBER_COUNT = 4096
HARD_MODEL_MEMBER_BYTES = HARD_ARTIFACT_CAPTURE_CEILING
HARD_MODEL_TOTAL_BYTES = HARD_ARTIFACT_CAPTURE_CEILING
HARD_MODEL_NAME_CAPTURE_CEILING = 512
SHA256 = re.compile(r"^[0-9a-f]{64}$")
PROFILE_FIELDS = {
    "schema_version", "profile_version", "control", "artifacts",
    "verification_keys", "result_base", "valid_from", "valid_until",
}
ARTIFACT_FIELDS = {"payload", "bundle"}
KEY_FIELDS = {"path", "sha256"}
AT_FDCWD = -100
RENAME_NOREPLACE = 1


@dataclass(frozen=True)
class Captured:
    raw: bytes
    bundle: bytes
    key: bytes

    def evidence(self) -> dict:
        return {
            "payload_sha256": hashlib.sha256(self.raw).hexdigest(),
            "bundle_sha256": hashlib.sha256(self.bundle).hexdigest(),
            "verification_key_sha256": hashlib.sha256(self.key).hexdigest(),
        }


def reject_constant(value: str):
    raise ValueError("non-finite JSON value is forbidden: " + value)


def unique_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("duplicate JSON key: " + key)
        result[key] = value
    return result


def strict_json_object(raw: bytes, label: str) -> dict:
    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(label + " must be a JSON object")
    return value


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


def read_stable(
    path: Path, *, maximum_bytes: int, allow_empty: bool = False,
    trust_root: bool = False,
) -> bytes:
    if trust_root:
        current = Path(path.anchor)
        for part in path.parts[1:-1]:
            current /= part
            metadata = current.lstat()
            if (
                not stat.S_ISDIR(metadata.st_mode)
                or current.is_symlink()
                or metadata.st_uid != 0
                or metadata.st_mode & 0o022
            ):
                raise PermissionError(
                    "trusted directory is not root-owned and non-writable: "
                    + str(current)
                )
    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0)
    descriptor = os.open(path, flags)
    try:
        before = os.fstat(descriptor)
        if (
            not stat.S_ISREG(before.st_mode)
            or before.st_size < 0
            or before.st_size > maximum_bytes
            or (before.st_size == 0 and not allow_empty)
        ):
            raise ValueError("input is not a bounded regular file: " + str(path))
        if trust_root and (before.st_uid != 0 or before.st_mode & 0o022):
            raise PermissionError(
                "trusted file must be root-owned and non-writable: " + str(path)
            )
        chunks = []
        total = 0
        while True:
            block = os.read(descriptor, min(1024 * 1024, maximum_bytes + 1 - total))
            if not block:
                break
            chunks.append(block)
            total += len(block)
            if total > maximum_bytes:
                raise ValueError("input exceeds its hard byte ceiling: " + str(path))
        after = os.fstat(descriptor)
    finally:
        os.close(descriptor)
    identity = lambda item: (
        item.st_dev, item.st_ino, item.st_size,
        item.st_mtime_ns, item.st_ctime_ns,
    )
    raw = b"".join(chunks)
    if identity(before) != identity(after) or len(raw) != before.st_size:
        raise RuntimeError("input changed during capture: " + str(path))
    return raw


def write_exclusive(path: Path, raw: bytes, mode: int = 0o600) -> None:
    descriptor = os.open(
        path,
        os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_CLOEXEC", 0),
        mode,
    )
    with os.fdopen(descriptor, "wb") as stream:
        stream.write(raw)
        stream.flush()
        os.fsync(stream.fileno())


def verify_bytes(label: str, raw: bytes, bundle: bytes, key: bytes) -> None:
    with tempfile.TemporaryDirectory(prefix="aidefend-d014-verify-") as name:
        root = Path(name)
        os.chmod(root, 0o700)
        payload = root / "payload"
        signature = root / "payload.sigstore.json"
        verify_key = root / "verify.pub"
        for path, value in ((payload, raw), (signature, bundle), (verify_key, key)):
            write_exclusive(path, value)
        subprocess.run(
            [
                COSIGN, "verify-blob", "--key", str(verify_key),
                "--bundle", str(signature), str(payload),
            ],
            check=True, capture_output=True, text=True,
            timeout=COSIGN_DEADLINE_CEILING_SECONDS,
        )
        for path, expected in ((payload, raw), (signature, bundle), (verify_key, key)):
            actual = read_stable(
                path, maximum_bytes=max(len(expected), 1),
                allow_empty=(path == payload and not expected),
            )
            if not hmac.compare_digest(actual, expected):
                raise RuntimeError(label + " private verification snapshot changed")


def capture_explicit(
    label: str, payload: Path, bundle: Path, key_path: Path,
    *, maximum_bytes: int, allow_empty: bool = False,
    trust_payload_paths: bool = False,
) -> Captured:
    raw = read_stable(
        payload, maximum_bytes=maximum_bytes, allow_empty=allow_empty,
        trust_root=trust_payload_paths,
    )
    bundle_raw = read_stable(
        bundle, maximum_bytes=HARD_METADATA_CAPTURE_CEILING,
        trust_root=trust_payload_paths,
    )
    key = read_stable(
        key_path, maximum_bytes=HARD_KEY_CAPTURE_CEILING, trust_root=True
    )
    verify_bytes(label, raw, bundle_raw, key)
    return Captured(raw, bundle_raw, key)


def canonical_absolute(value: object, label: str, roots: tuple[str, ...]) -> Path:
    if not isinstance(value, str) or not value or "\\" in value:
        raise ValueError(label + " is not a canonical absolute POSIX path")
    pure = PurePosixPath(value)
    if not pure.is_absolute() or ".." in pure.parts or str(pure) != value:
        raise ValueError(label + " is not a canonical absolute POSIX path")
    if not any(value.startswith(root + "/") for root in roots):
        raise ValueError(label + " is outside its fixed deployment roots")
    return Path(value)


def load_profile(
    expected_control: str, artifact_roles: tuple[str, ...], *,
    profile_name: str, evaluated_at: datetime,
) -> tuple[dict, Captured]:
    if not re.fullmatch(r"[a-z0-9-]+", profile_name):
        raise ValueError("fixed runtime-profile name is invalid")
    profile_path = BOOTSTRAP_DIRECTORY / (profile_name + ".json")
    profile_bundle = BOOTSTRAP_DIRECTORY / (profile_name + ".sigstore.json")
    bootstrap = capture_explicit(
        "D014 runtime profile", profile_path, profile_bundle,
        BOOTSTRAP_KEY, maximum_bytes=HARD_METADATA_CAPTURE_CEILING,
        trust_payload_paths=True,
    )
    profile = strict_json_object(bootstrap.raw, "D014 runtime profile")
    roles = set(artifact_roles)
    if (
        set(profile) != PROFILE_FIELDS
        or profile.get("schema_version") != "aidefend.d014-runtime-profile.v1"
        or profile.get("control") != expected_control
        or not isinstance(profile.get("profile_version"), str)
        or not profile["profile_version"]
        or not isinstance(profile.get("artifacts"), dict)
        or set(profile["artifacts"]) != roles
        or not isinstance(profile.get("verification_keys"), dict)
        or set(profile["verification_keys"]) != roles | {"result"}
    ):
        raise ValueError("signed D014 runtime profile schema or scope differs")
    if evaluated_at.utcoffset() is None:
        raise ValueError("runtime-profile evaluation time lacks timezone")
    valid_from = datetime.fromisoformat(
        str(profile["valid_from"]).replace("Z", "+00:00")
    )
    valid_until = datetime.fromisoformat(
        str(profile["valid_until"]).replace("Z", "+00:00")
    )
    if (
        valid_from.utcoffset() is None or valid_until.utcoffset() is None
        or valid_from.astimezone(timezone.utc) >= valid_until.astimezone(timezone.utc)
        or not (
            valid_from.astimezone(timezone.utc)
            <= evaluated_at.astimezone(timezone.utc)
            < valid_until.astimezone(timezone.utc)
        )
    ):
        raise ValueError("signed D014 runtime profile is outside its validity interval")
    for role, artifact in profile["artifacts"].items():
        if not isinstance(artifact, dict) or set(artifact) != ARTIFACT_FIELDS:
            raise ValueError("artifact path contract differs for role " + role)
        for field in ARTIFACT_FIELDS:
            canonical_absolute(
                artifact[field], role + "." + field,
                ("/etc/aidefend/d014", "/var/lib/aidefend/d014"),
            )
    key_digests = []
    for role, spec in profile["verification_keys"].items():
        if (
            not isinstance(spec, dict)
            or set(spec) != KEY_FIELDS
            or not SHA256.fullmatch(str(spec.get("sha256")))
        ):
            raise ValueError("verification-key contract differs for role " + role)
        canonical_absolute(
            spec["path"], role + ".key",
            ("/etc/aidefend/d014/trust",),
        )
        key_digests.append(spec["sha256"])
    bootstrap_key_sha = bootstrap.evidence()["verification_key_sha256"]
    if len(key_digests) != len(set(key_digests)) or bootstrap_key_sha in key_digests:
        raise ValueError("D014 trust roles are not key-separated")
    canonical_absolute(
        profile["result_base"], "result_base",
        ("/var/lib/aidefend/d014/results",),
    )
    return profile, bootstrap


def capture_role(
    profile: dict, role: str, *, maximum_bytes: int,
    allow_empty: bool = False,
) -> Captured:
    artifact = profile["artifacts"][role]
    key_spec = profile["verification_keys"][role]
    captured = capture_explicit(
        role,
        Path(artifact["payload"]), Path(artifact["bundle"]),
        Path(key_spec["path"]), maximum_bytes=maximum_bytes,
        allow_empty=allow_empty,
    )
    if not hmac.compare_digest(
        captured.evidence()["verification_key_sha256"], key_spec["sha256"]
    ):
        raise ValueError("captured verification key differs for role " + role)
    return captured


def safe_member_name(value: str) -> PurePosixPath:
    if (
        not value or "\\" in value
        or len(value.encode("utf-8")) > HARD_MODEL_NAME_CAPTURE_CEILING
    ):
        raise ValueError("model archive member name is invalid")
    path = PurePosixPath(value)
    if path.is_absolute() or ".." in path.parts or str(path) != value:
        raise ValueError("model archive member escapes its extraction root")
    return path


@contextmanager
def safe_model_root(model_tar_raw: bytes):
    """Preinspect and manually copy an uncompressed, signed model tar."""
    with tempfile.TemporaryDirectory(prefix="aidefend-d014-model-") as name:
        root = Path(name)
        os.chmod(root, 0o700)
        manifest = []
        with tarfile.open(fileobj=BytesIO(model_tar_raw), mode="r|") as archive:
            seen = set()
            regular = set()
            total = 0
            for member in archive:
                if len(manifest) >= HARD_MODEL_MEMBER_COUNT:
                    raise ValueError("model archive member count exceeds its hard ceiling")
                path = safe_member_name(member.name)
                folded = str(path).casefold()
                if folded in seen:
                    raise ValueError("duplicate model archive member")
                seen.add(folded)
                if not (member.isdir() or member.isreg()) or member.issym() or member.islnk():
                    raise ValueError("links and special model archive members are forbidden")
                if member.isreg() and member.mode & 0o111:
                    raise ValueError("executable model archive member is forbidden")
                if member.isreg():
                    if member.size < 0 or member.size > HARD_MODEL_MEMBER_BYTES:
                        raise ValueError("model member exceeds its hard byte ceiling")
                    total += member.size
                    if total > HARD_MODEL_TOTAL_BYTES:
                        raise ValueError("model archive exceeds its hard total-byte ceiling")
                    suffix = path.suffix.lower()
                    if suffix not in {".json", ".safetensors", ".txt", ".model", ".vocab", ".merges"}:
                        raise ValueError("unsafe or unsupported model member type: " + suffix)
                    regular.add(str(path))
                manifest.append((
                    member.name, member.size, member.mode,
                    member.isdir(), member.isreg(),
                ))
            if not manifest:
                raise ValueError("model archive is empty")
            for filename in regular:
                parts = PurePosixPath(filename).parts
                for index in range(1, len(parts)):
                    if str(PurePosixPath(*parts[:index])) in regular:
                        raise ValueError("model archive file shadows a parent directory")
            if "config.json" not in regular:
                raise ValueError("model archive lacks config.json")
            if not any(name.endswith(".safetensors") for name in regular):
                raise ValueError("model archive lacks safetensors weights")
            tokenizer_names = {
                "tokenizer.json", "tokenizer.model", "vocab.json", "vocab.txt",
                "sentencepiece.bpe.model", "merges.txt",
            }
            if not regular.intersection(tokenizer_names):
                raise ValueError("model archive lacks a supported tokenizer asset")
        with tarfile.open(fileobj=BytesIO(model_tar_raw), mode="r|") as archive:
            position = -1
            for position, member in enumerate(archive):
                observed = (
                    member.name, member.size, member.mode,
                    member.isdir(), member.isreg(),
                )
                if position >= len(manifest) or observed != manifest[position]:
                    raise RuntimeError("model archive changed between inspection passes")
                relative = safe_member_name(member.name)
                target = root.joinpath(*relative.parts)
                if member.isdir():
                    target.mkdir(parents=True, exist_ok=True, mode=0o700)
                    continue
                target.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
                stream = archive.extractfile(member)
                if stream is None:
                    raise ValueError("regular model member cannot be read")
                raw = stream.read(member.size + 1)
                if len(raw) != member.size:
                    raise ValueError("model member size differs during extraction")
                write_exclusive(target, raw, mode=0o400)
            if position + 1 != len(manifest):
                raise RuntimeError("model archive member count changed between passes")
        yield root


def fsync_directory(path: Path) -> None:
    descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
    try:
        os.fsync(descriptor)
    finally:
        os.close(descriptor)


def fsync_regular_file(path: Path) -> None:
    descriptor = os.open(
        path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
        | getattr(os, "O_CLOEXEC", 0),
    )
    try:
        metadata = os.fstat(descriptor)
        if not stat.S_ISREG(metadata.st_mode):
            raise ValueError("publication member is not a regular file")
        os.fsync(descriptor)
    finally:
        os.close(descriptor)


def assert_private_result_base(path: Path) -> None:
    current = Path(path.anchor)
    for part in path.parts[1:]:
        current /= part
        metadata = current.lstat()
        is_result_base = current == path
        expected_uid = os.geteuid() if is_result_base else 0
        if (
            not stat.S_ISDIR(metadata.st_mode)
            or current.is_symlink()
            or metadata.st_uid != expected_uid
            or metadata.st_mode & 0o022
        ):
            raise PermissionError(
                "result directory chain is not privileged and non-writable: "
                + str(current)
            )


def rename_noreplace(source: Path, target: Path) -> None:
    libc = ctypes.CDLL(None, use_errno=True)
    function = getattr(libc, "renameat2", None)
    if function is None:
        raise OSError(errno.ENOSYS, "renameat2 is required for no-replace publication")
    function.argtypes = [
        ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p,
        ctypes.c_uint,
    ]
    function.restype = ctypes.c_int
    result = function(
        AT_FDCWD, os.fsencode(source), AT_FDCWD, os.fsencode(target),
        RENAME_NOREPLACE,
    )
    if result != 0:
        number = ctypes.get_errno()
        raise OSError(number, os.strerror(number), str(target))


def publish_result(
    value: dict, profile: dict, input_key_sha256s: list[str],
    *, signing_key_uri: str,
) -> tuple[Path, dict]:
    if not signing_key_uri.startswith("env://D014"):
        raise ValueError("result signing-key URI is not a fixed D014 secret name")
    result_base = Path(profile["result_base"])
    assert_private_result_base(result_base)
    key_spec = profile["verification_keys"]["result"]
    result_key = read_stable(
        Path(key_spec["path"]), maximum_bytes=HARD_KEY_CAPTURE_CEILING,
        trust_root=True,
    )
    result_key_sha = hashlib.sha256(result_key).hexdigest()
    if (
        not hmac.compare_digest(result_key_sha, key_spec["sha256"])
        or result_key_sha in set(input_key_sha256s)
    ):
        raise ValueError("result verification key is unpinned or not role-separated")
    published = {**value, "result_verification_key_sha256": result_key_sha}
    raw = canonical(published) + b"\n"
    if len(raw) > HARD_METADATA_CAPTURE_CEILING:
        raise ValueError("result exceeds its hard byte ceiling")
    result_sha = hashlib.sha256(raw).hexdigest()
    target = result_base / result_sha
    if target.exists() or target.is_symlink():
        raise FileExistsError("content-addressed result already exists: " + result_sha)
    staging = Path(tempfile.mkdtemp(prefix="." + result_sha + ".", dir=result_base))
    os.chmod(staging, 0o700)
    try:
        result_path = staging / "result.json"
        bundle_path = staging / "result.sigstore.json"
        key_path = staging / "result.pub"
        write_exclusive(result_path, raw)
        write_exclusive(key_path, result_key)
        subprocess.run(
            [
                COSIGN, "sign-blob", "--yes", "--key", signing_key_uri,
                "--bundle", str(bundle_path), str(result_path),
            ],
            check=True, timeout=COSIGN_DEADLINE_CEILING_SECONDS,
        )
        fsync_regular_file(bundle_path)
        bundle_raw = read_stable(
            bundle_path, maximum_bytes=HARD_METADATA_CAPTURE_CEILING
        )
        if not hmac.compare_digest(
            read_stable(result_path, maximum_bytes=HARD_METADATA_CAPTURE_CEILING), raw
        ):
            raise RuntimeError("private result changed before verification")
        verify_bytes("D014 result", raw, bundle_raw, result_key)
        committed = canonical({
            "schema_version": "aidefend.signed-result-publication.v1",
            "payload_sha256": result_sha,
            "bundle_sha256": hashlib.sha256(bundle_raw).hexdigest(),
            "verification_key_sha256": result_key_sha,
        }) + b"\n"
        write_exclusive(staging / "COMMITTED.json", committed)
        expected = {
            "result.json": raw,
            "result.sigstore.json": bundle_raw,
            "result.pub": result_key,
            "COMMITTED.json": committed,
        }
        if {path.name for path in staging.iterdir()} != set(expected):
            raise RuntimeError("staged result directory contents differ")
        for path in staging.iterdir():
            os.chmod(path, 0o440)
            fsync_regular_file(path)
        os.chmod(staging, 0o550)
        fsync_directory(staging)
        rename_noreplace(staging, target)
        fsync_directory(result_base)
        staging = None
        if {path.name for path in target.iterdir()} != set(expected):
            raise RuntimeError("published result directory contents differ")
        readback = {
            name: read_stable(path, maximum_bytes=HARD_METADATA_CAPTURE_CEILING)
            for name, path in ((name, target / name) for name in expected)
        }
        for name, wanted in expected.items():
            if not hmac.compare_digest(readback[name], wanted):
                raise RuntimeError("published result readback differs: " + name)
        verify_bytes(
            "published D014 result", readback["result.json"],
            readback["result.sigstore.json"], readback["result.pub"],
        )
        return target, published
    finally:
        if staging is not None and staging.exists():
            os.chmod(staging, 0o700)
            for path in staging.iterdir():
                os.chmod(path, 0o600)
            shutil.rmtree(staging)

Evaluate the complete population

The evaluator verifies ranks 1..k for every query, tokenizes query and chunk without truncation, repeats the complete query in every pair, and windows the complete chunk with no uncovered tail. It explicitly selects the policy-named relevance logit after the bound transform and takes the minimum relevance across all chunk windows, so a relevant prefix cannot hide an unrelated tail. Equality with the minimum threshold passes; a lower window score is a finding.

from __future__ import annotations

# File: detection/evaluate_retrieval_relevance.py

from collections import defaultdict
from datetime import datetime, timedelta, timezone
import hashlib
import hmac
from io import BytesIO
import math

from detection.d014_verified_artifacts import (
    HARD_ARTIFACT_CAPTURE_CEILING, HARD_METADATA_CAPTURE_CEILING,
    SHA256, canonical, capture_role, load_profile, publish_result,
    safe_model_root, strict_json_object,
)


CONTROL = "AID-D-014.002-G001"
PROFILE_NAME = "d014002-g001-relevance"
ROLES = ("snapshot", "receipt", "policy", "model")
SIGNING_KEY_URI = "env://D014002_G001_RESULT_SIGNING_KEY"
HARD_PAIR_COUNT = 500_000
HARD_LINE_CAPTURE_CEILING = 16 * 1024 * 1024
HARD_QUERY_TOKEN_COUNT = 32_000
HARD_CHUNK_TOKEN_COUNT = 1_000_000
HARD_TOTAL_WINDOW_COUNT = 1_000_000
HARD_SEQUENCE_TOKEN_COUNT = 32_768
HARD_FINDING_SAMPLE_COUNT = 1_000
POLICY_FIELDS = {
    "schema_version", "policy_version", "applicable_query_classes",
    "expected_top_k_by_query_class", "minimum_relevance_by_query_class",
    "model_sha256", "model_revision", "id2label", "relevance_label",
    "score_transform", "max_sequence_tokens", "window_overlap_tokens",
    "maximum_query_tokens", "maximum_chunk_tokens", "maximum_total_windows",
    "maximum_pairs", "maximum_snapshot_age_seconds",
    "maximum_future_skew_seconds", "retention_policy_sha256",
}
RECEIPT_FIELDS = {
    "schema_version", "subject", "snapshot_completed_at", "emitted_at",
    "query_count", "pair_count", "collection_complete", "snapshot_sha256",
}
RECORD_FIELDS = {
    "query_id", "query_class", "query_text", "chunk_id", "chunk_text", "rank",
}
EXIT_CODES = {
    "PASS": 0, "FAIL": 1, "INSUFFICIENT_DATA": 2,
    "NOT_APPLICABLE": 0, "ERROR": 3,
}


def parse_time(value: object) -> datetime:
    if not isinstance(value, str):
        raise ValueError("timestamp is not text")
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.utcoffset() is None:
        raise ValueError("timestamp lacks timezone")
    return parsed.astimezone(timezone.utc)


def integer(value: object, label: str, *, positive: bool = False) -> int:
    if isinstance(value, bool) or not isinstance(value, int):
        raise ValueError(label + " must be an integer")
    if value < (1 if positive else 0):
        raise ValueError(label + " is outside its allowed range")
    return value


def unit_interval(value: object, label: str) -> float:
    if (
        isinstance(value, bool)
        or not isinstance(value, (int, float))
        or not math.isfinite(float(value))
        or not 0 <= float(value) <= 1
    ):
        raise ValueError(label + " must be finite and in [0,1]")
    return float(value)


def sorted_strings(value: object, label: str) -> list[str]:
    if (
        not isinstance(value, list) or not value
        or any(not isinstance(item, str) or not item for item in value)
        or value != sorted(set(value))
    ):
        raise ValueError(label + " must be a sorted unique string list")
    return value


def snapshot_records(raw: bytes, maximum_pairs: int) -> list[dict]:
    if raw and not raw.endswith(b"\n"):
        raise ValueError("relevance JSONL must end with a newline")
    stream = BytesIO(raw)
    records = []
    while True:
        line = stream.readline(HARD_LINE_CAPTURE_CEILING + 1)
        if not line:
            break
        if len(line) > HARD_LINE_CAPTURE_CEILING:
            raise ValueError("relevance pair exceeds its hard line ceiling")
        if not line.strip():
            raise ValueError("relevance JSONL contains a blank record")
        record = strict_json_object(line, "retrieval relevance pair")
        if (
            set(record) != RECORD_FIELDS
            or any(
                not isinstance(record[field], str) or not record[field]
                for field in RECORD_FIELDS - {"rank"}
            )
        ):
            raise ValueError("relevance pair schema differs")
        integer(record["rank"], "rank", positive=True)
        records.append(record)
        if len(records) > maximum_pairs:
            raise ValueError("relevance pair population exceeds policy")
    if len({(item["query_id"], item["rank"]) for item in records}) != len(records):
        raise ValueError("duplicate query rank in relevance snapshot")
    if len({(item["query_id"], item["chunk_id"]) for item in records}) != len(records):
        raise ValueError("duplicate query-chunk pair in relevance snapshot")
    return records


def load_inputs(profile: dict, bootstrap, evaluated_at: datetime) -> dict:
    captures = {
        "snapshot": capture_role(
            profile, "snapshot", maximum_bytes=HARD_ARTIFACT_CAPTURE_CEILING,
            allow_empty=True,
        ),
        "receipt": capture_role(
            profile, "receipt", maximum_bytes=HARD_METADATA_CAPTURE_CEILING,
        ),
        "policy": capture_role(
            profile, "policy", maximum_bytes=HARD_METADATA_CAPTURE_CEILING,
        ),
        "model": capture_role(
            profile, "model", maximum_bytes=HARD_ARTIFACT_CAPTURE_CEILING,
        ),
    }
    policy = strict_json_object(captures["policy"].raw, "relevance policy")
    receipt = strict_json_object(captures["receipt"].raw, "relevance receipt")
    if (
        set(policy) != POLICY_FIELDS
        or policy.get("schema_version") != "aidefend.d014002-relevance-policy.v2"
        or any(
            not isinstance(policy.get(field), str) or not policy[field]
            for field in ("policy_version", "model_revision", "relevance_label")
        )
        or policy.get("score_transform") not in {"sigmoid", "softmax"}
        or not SHA256.fullmatch(str(policy.get("model_sha256")))
        or not SHA256.fullmatch(str(policy.get("retention_policy_sha256")))
    ):
        raise ValueError("relevance policy schema or identity differs")
    classes = sorted_strings(policy["applicable_query_classes"], "query classes")
    expected_top_k = policy["expected_top_k_by_query_class"]
    minimum_relevance = policy["minimum_relevance_by_query_class"]
    if (
        not isinstance(expected_top_k, dict) or set(expected_top_k) != set(classes)
        or not isinstance(minimum_relevance, dict) or set(minimum_relevance) != set(classes)
    ):
        raise ValueError("query-class policy mappings are incomplete")
    for query_class in classes:
        integer(expected_top_k[query_class], "top-k for " + query_class, positive=True)
        unit_interval(minimum_relevance[query_class], "relevance for " + query_class)
    id2label = policy["id2label"]
    indexes = [str(index) for index in range(len(id2label))] if isinstance(id2label, dict) else []
    if (
        not indexes or set(id2label) != set(indexes)
        or any(not isinstance(id2label[index], str) or not id2label[index] for index in indexes)
        or len(set(id2label.values())) != len(id2label)
        or policy["relevance_label"] not in set(id2label.values())
        or (policy["score_transform"] == "softmax" and len(id2label) < 2)
    ):
        raise ValueError("relevance label semantics differ")
    for field, ceiling in (
        ("max_sequence_tokens", HARD_SEQUENCE_TOKEN_COUNT),
        ("maximum_query_tokens", HARD_QUERY_TOKEN_COUNT),
        ("maximum_chunk_tokens", HARD_CHUNK_TOKEN_COUNT),
        ("maximum_total_windows", HARD_TOTAL_WINDOW_COUNT),
        ("maximum_pairs", HARD_PAIR_COUNT),
    ):
        if integer(policy[field], field, positive=True) > ceiling:
            raise ValueError(field + " exceeds its process-safety ceiling")
    overlap = integer(policy["window_overlap_tokens"], "window_overlap_tokens")
    if overlap >= policy["max_sequence_tokens"]:
        raise ValueError("window overlap leaves no forward progress")
    integer(policy["maximum_snapshot_age_seconds"], "maximum_snapshot_age_seconds", positive=True)
    integer(policy["maximum_future_skew_seconds"], "maximum_future_skew_seconds")
    if not hmac.compare_digest(
        captures["model"].evidence()["payload_sha256"], policy["model_sha256"]
    ):
        raise ValueError("relevance policy does not bind the model tar")

    if (
        set(receipt) != RECEIPT_FIELDS
        or receipt.get("schema_version") != "aidefend.d014002-relevance-receipt.v2"
        or set(receipt.get("subject", {})) != {"asset_id", "asset_version"}
        or any(
            not isinstance(receipt["subject"].get(field), str)
            or not receipt["subject"][field]
            for field in ("asset_id", "asset_version")
        )
        or not isinstance(receipt.get("collection_complete"), bool)
        or not SHA256.fullmatch(str(receipt.get("snapshot_sha256")))
    ):
        raise ValueError("relevance receipt schema or subject differs")
    query_count = integer(receipt["query_count"], "query_count")
    pair_count = integer(receipt["pair_count"], "pair_count")
    if pair_count > policy["maximum_pairs"]:
        raise ValueError("signed pair count exceeds policy")
    if not hmac.compare_digest(
        captures["snapshot"].evidence()["payload_sha256"], receipt["snapshot_sha256"]
    ):
        raise ValueError("relevance receipt does not bind the exact snapshot")
    records = snapshot_records(captures["snapshot"].raw, policy["maximum_pairs"])
    if len(records) != pair_count:
        raise ValueError("relevance pair count differs from receipt")
    grouped = defaultdict(list)
    for record in records:
        if record["query_class"] not in set(classes):
            raise ValueError("query class is outside signed policy")
        grouped[record["query_id"]].append(record)
    if len(grouped) != query_count:
        raise ValueError("query population differs from receipt")
    for query_id, items in grouped.items():
        query_classes = {item["query_class"] for item in items}
        query_texts = {item["query_text"] for item in items}
        if len(query_classes) != 1 or len(query_texts) != 1:
            raise ValueError("query identity has inconsistent class or text")
        query_class = items[0]["query_class"]
        expected = policy["expected_top_k_by_query_class"][query_class]
        if sorted(item["rank"] for item in items) != list(range(1, expected + 1)):
            raise ValueError("query does not contain the complete signed top-k")
    completed = parse_time(receipt["snapshot_completed_at"])
    emitted = parse_time(receipt["emitted_at"])
    if emitted < completed:
        raise ValueError("receipt predates relevance snapshot completion")
    if emitted > evaluated_at + timedelta(seconds=policy["maximum_future_skew_seconds"]):
        raise ValueError("receipt is future-dated beyond policy skew")
    age = max(0.0, (evaluated_at - completed).total_seconds())
    if age > policy["maximum_snapshot_age_seconds"]:
        readiness, reason = "INSUFFICIENT_DATA", "signed relevance snapshot is stale"
    elif receipt["collection_complete"] is not True:
        readiness, reason = "INSUFFICIENT_DATA", "authenticated top-k collection is incomplete"
    else:
        readiness, reason = "READY", None
    return {
        "status": readiness, "reason": reason, "profile": profile,
        "bootstrap": bootstrap, "captures": captures, "policy": policy,
        "receipt": receipt, "records": records, "snapshot_age_seconds": age,
    }


def window_starts(token_count: int, capacity: int, overlap: int) -> list[int]:
    if token_count <= capacity:
        return [0]
    step = capacity - overlap
    final_start = token_count - capacity
    starts = list(range(0, final_start + 1, step))
    if starts[-1] != final_start:
        starts.append(final_start)
    return starts


def scan(records: list[dict], model_raw: bytes, policy: dict) -> dict:
    import torch
    from transformers import AutoModelForSequenceClassification, AutoTokenizer

    labels = [policy["id2label"][str(index)] for index in range(len(policy["id2label"]))]
    relevance_index = labels.index(policy["relevance_label"])
    finding_count = 0
    finding_sample = []
    unmeasured_pairs = 0
    scored_pairs = 0
    scored_windows = 0
    digest = hashlib.sha256()
    with safe_model_root(model_raw) as model_root:
        tokenizer = AutoTokenizer.from_pretrained(
            str(model_root), local_files_only=True, trust_remote_code=False,
        )
        model = AutoModelForSequenceClassification.from_pretrained(
            str(model_root), local_files_only=True, trust_remote_code=False,
            use_safetensors=True,
        )
        model.eval()
        model_labels = {str(key): str(value) for key, value in dict(model.config.id2label).items()}
        if model_labels != policy["id2label"] or model.config.num_labels != len(labels):
            raise ValueError("loaded relevance labels differ from signed policy")
        limits = [
            value for value in (
                getattr(model.config, "max_position_embeddings", None),
                getattr(tokenizer, "model_max_length", None),
            ) if isinstance(value, int) and not isinstance(value, bool) and 0 < value < 1_000_000
        ]
        if limits and policy["max_sequence_tokens"] > min(limits):
            raise ValueError("signed sequence limit exceeds the loaded model limit")
        special = tokenizer.num_special_tokens_to_add(pair=True)
        for record in records:
            query_ids = tokenizer.encode(
                record["query_text"], add_special_tokens=False, truncation=False,
            )
            chunk_ids = tokenizer.encode(
                record["chunk_text"], add_special_tokens=False, truncation=False,
            )
            capacity = policy["max_sequence_tokens"] - special - len(query_ids)
            overlap = policy["window_overlap_tokens"]
            if (
                len(query_ids) > policy["maximum_query_tokens"]
                or len(chunk_ids) > policy["maximum_chunk_tokens"]
                or capacity <= 0 or overlap >= capacity
            ):
                unmeasured_pairs += 1
                continue
            starts = window_starts(len(chunk_ids), capacity, overlap)
            count = len(starts)
            if scored_windows + count > policy["maximum_total_windows"]:
                unmeasured_pairs += 1
                continue
            step = capacity - overlap
            minimum = 1.0
            for start in starts:
                piece = chunk_ids[start:start + capacity]
                encoded = tokenizer.prepare_for_model(
                    query_ids, piece, add_special_tokens=True, truncation=False,
                    return_attention_mask=True, return_tensors="pt",
                )
                if int(encoded["input_ids"].shape[-1]) > policy["max_sequence_tokens"]:
                    raise ValueError("prepared relevance window exceeds policy")
                batch = {name: value.to(model.device) for name, value in encoded.items()}
                with torch.inference_mode():
                    logits = model(**batch).logits
                if tuple(logits.shape) != (1, len(labels)) or not torch.isfinite(logits).all():
                    raise ValueError("relevance logits shape or values differ")
                scores = (
                    torch.sigmoid(logits)[0] if policy["score_transform"] == "sigmoid"
                    else torch.softmax(logits, dim=-1)[0]
                ).detach().cpu().tolist()
                if len(scores) != len(labels) or any(
                    not isinstance(value, (int, float))
                    or not math.isfinite(float(value))
                    or not 0 <= float(value) <= 1
                    for value in scores
                ):
                    raise ValueError("relevance score domain differs")
                minimum = min(minimum, float(scores[relevance_index]))
                scored_windows += 1
            threshold = float(
                policy["minimum_relevance_by_query_class"][record["query_class"]]
            )
            scored = {
                "query_id": record["query_id"], "chunk_id": record["chunk_id"],
                "window_count": count, "minimum_relevance": minimum,
                "threshold": threshold,
            }
            digest.update(canonical(scored) + b"\n")
            if minimum < threshold:
                finding_count += 1
                if len(finding_sample) < HARD_FINDING_SAMPLE_COUNT:
                    finding_sample.append(scored)
            scored_pairs += 1
    return {
        "scored_pairs": scored_pairs, "scored_windows": scored_windows,
        "unmeasured_pairs": unmeasured_pairs, "finding_count": finding_count,
        "finding_sample": finding_sample,
        "score_records_sha256": digest.hexdigest(),
    }


def evaluate(profile: dict, bootstrap, evaluated_at: datetime) -> tuple[dict, list[str]]:
    inputs = load_inputs(profile, bootstrap, evaluated_at)
    receipt, policy = inputs["receipt"], inputs["policy"]
    evidence = {name: item.evidence() for name, item in inputs["captures"].items()}
    base = {
        "schema_version": "aidefend.d014002-relevance-result.v3",
        "control": CONTROL, "subject": receipt["subject"],
        "policy_version": policy["policy_version"],
        "model_revision": policy["model_revision"],
        "profile_version": profile["profile_version"],
        "profile_sha256": bootstrap.evidence()["payload_sha256"],
        "snapshot_age_seconds": inputs["snapshot_age_seconds"],
        "retention_policy_sha256": policy["retention_policy_sha256"],
        "input_evidence": evidence,
        "evaluated_at": evaluated_at.isoformat().replace("+00:00", "Z"),
    }
    keys = [bootstrap.evidence()["verification_key_sha256"]] + [
        item.evidence()["verification_key_sha256"] for item in inputs["captures"].values()
    ]
    if inputs["status"] != "READY":
        return {**base, "status": inputs["status"], "reason": inputs["reason"], "metrics": None}, keys
    if not inputs["records"]:
        return {**base, "status": "INSUFFICIENT_DATA", "reason": "no complete top-k query occurred", "metrics": None}, keys
    metrics = scan(inputs["records"], inputs["captures"]["model"].raw, policy)
    if metrics["finding_count"]:
        status, reason = "FAIL", "one or more complete chunk windows fell below signed relevance"
    elif metrics["unmeasured_pairs"]:
        status, reason = "INSUFFICIENT_DATA", "one or more query-chunk pairs could not be fully windowed"
    elif metrics["scored_pairs"] != receipt["pair_count"]:
        raise ValueError("relevance pair population does not reconcile")
    else:
        status, reason = "PASS", "every complete chunk window met signed relevance"
    return {**base, "status": status, "reason": reason, "metrics": metrics}, keys


def main() -> int:
    evaluated_at = datetime.now(timezone.utc)
    profile, bootstrap = load_profile(
        CONTROL, ROLES, profile_name=PROFILE_NAME, evaluated_at=evaluated_at,
    )
    reserved = [bootstrap.evidence()["verification_key_sha256"]] + [
        spec["sha256"] for role, spec in profile["verification_keys"].items()
        if role != "result"
    ]
    try:
        result, observed = evaluate(profile, bootstrap, evaluated_at)
        reserved = list(set(reserved + observed))
    except Exception as error:
        result = {
            "schema_version": "aidefend.d014002-relevance-result.v3",
            "control": CONTROL, "profile_version": profile["profile_version"],
            "profile_sha256": bootstrap.evidence()["payload_sha256"],
            "status": "ERROR",
            "reason": "input verification, relevance evaluation, or result preparation failed",
            "error_class": type(error).__name__,
            "evaluated_at": evaluated_at.isoformat().replace("+00:00", "Z"),
        }
    target, published = publish_result(
        result, profile, reserved, signing_key_uri=SIGNING_KEY_URI,
    )
    print(str(target))
    return EXIT_CODES[published["status"]]


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

Run and retain the result

Use Python 3.11 or newer with torch, transformers, safetensors, and Cosign. Provision the result directory for the dedicated evaluator identity, sign each exact input with its separately authorized role key, then sign the profile with the offline bootstrap role:

install -d -m 0755 -o root -g root /var/lib/aidefend/d014 /var/lib/aidefend/d014/results
install -d -m 0750 -o aidefend-d014 -g aidefend-d014 /var/lib/aidefend/d014/results/d014002-g001-relevance
cosign sign-blob --yes --key env://D014_SNAPSHOT_SIGNING_KEY --bundle /var/lib/aidefend/d014/input/d014002-g001-relevance-snapshot.jsonl.sigstore.json /var/lib/aidefend/d014/input/d014002-g001-relevance-snapshot.jsonl
cosign sign-blob --yes --key env://D014_RECEIPT_SIGNING_KEY --bundle /var/lib/aidefend/d014/input/d014002-g001-relevance-receipt.json.sigstore.json /var/lib/aidefend/d014/input/d014002-g001-relevance-receipt.json
cosign sign-blob --yes --key env://D014_POLICY_SIGNING_KEY --bundle /etc/aidefend/d014/input/d014002-g001-relevance-policy.json.sigstore.json /etc/aidefend/d014/input/d014002-g001-relevance-policy.json
cosign sign-blob --yes --key env://D014_MODEL_SIGNING_KEY --bundle /var/lib/aidefend/d014/input/d014002-g001-relevance-model.tar.sigstore.json /var/lib/aidefend/d014/input/d014002-g001-relevance-model.tar
: "${D014_PROFILE_VERSION:?set a deployment-controlled profile version}"
: "${D014_PROFILE_VALID_FROM:?set an RFC 3339 validity start}"
: "${D014_PROFILE_VALID_UNTIL:?set an RFC 3339 validity end}"
python -m detection.provision_d014002_g001_relevance_profile "${D014_PROFILE_VERSION}" "${D014_PROFILE_VALID_FROM}" "${D014_PROFILE_VALID_UNTIL}"
cosign sign-blob --yes --key env://D014_RUNTIME_PROFILE_SIGNING_KEY --bundle /etc/aidefend/d014/profiles/d014002-g001-relevance.sigstore.json /etc/aidefend/d014/profiles/d014002-g001-relevance.json
python -m detection.evaluate_retrieval_relevance

A complete population whose every window meets its class threshold is PASS; a low-relevance pair is FAIL; stale, incomplete, empty, overlong, or otherwise not-fully-windowed populations are INSUFFICIENT_DATA; model-label, integrity, schema, or publication faults are ERROR. The evaluator obtains UTC time once inside main(). It signs a canonical v3 result, verifies it against the profile-pinned result key, commits it under its content digest with no-replace semantics, and reads every file back before returning. Replicate the committed directory and exact signed inputs to append-only or WORM storage under a separate retention identity; replay from those bytes rather than reopening mutable producer paths.

Open-source tools

• sentence-transformers
• FAISS
• Weaviate
• Cosign

Commercial / hosted tools

• Pinecone
• Weaviate Cloud
• Cohere Rerank API
• Jina AI Reranker API
• Voyage AI Rerank API

Threat mappings

MITRE ATLAS

• AML.T0070 RAG Poisoning (independent query-chunk scoring detects off-topic retrieval manipulation; on-topic falsehoods remain outside this detector)
• AML.T0071 False RAG Entry Injection (independent query-chunk scoring detects off-topic retrieval manipulation; on-topic falsehoods remain outside this detector)
• AML.T0066 Retrieval Content Crafting (independent query-chunk scoring detects off-topic retrieval manipulation; on-topic falsehoods remain outside this detector)

MAESTRO

• Compromised RAG Pipelines (L2) (independent query-chunk scoring detects off-topic retrieval manipulation)

OWASP LLM Top 10 2026

• LLM09:2026 Vector and Embedding Weaknesses (independent query-chunk scoring detects off-topic retrieval manipulation; on-topic falsehoods remain outside this detector)

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI06:2026 Memory & Context Poisoning (independent query-chunk scoring detects off-topic retrieval manipulation; on-topic falsehoods remain outside this detector)

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AISubtech-6.1.1 Knowledge Base Poisoning (independent query-chunk scoring detects off-topic poisoned retrievals; on-topic falsehoods remain outside scope)

Google Secure AI Framework 2.0 - Risks

• N/A

Databricks AI Security Framework 3.0

• Agents - Tools MCP Server 13.24: Context Spoofing and Manipulation (independent query-chunk scoring detects off-topic retrieval manipulation; on-topic falsehoods remain outside this detector)