AID-D-001.003: Vector-Space Anomaly Detection

Version 1.20261015 · Actionable sub-technique

Detect anomalous text inputs at a declared, bounded AI application ingress by measuring their distance from a trusted, versioned embedding baseline. A concrete deployment is a narrow task assistant where an attacker submits instructions to replace the assigned task or elicit a prohibited response. Use statistical novelty as an investigative signal only when an independent replay on that application demonstrates useful detection of those attacks, including adaptive prompts near normal clusters, within policy limits for benign alerts and latency.

Bind each finding to the exact input, application surface, embedding model, baseline and policy generation, and reconcile the complete independently observed ingress population by ID. A normal-distance prompt can still be malicious; a novel legitimate task can be an outlier. An absent text-ingress architecture is NOT_APPLICABLE. Missing attack evaluation or omitted applicable traffic is INSUFFICIENT_DATA; a measured breached detection or utility requirement is FAIL, and processing errors remain ERROR. A successful arithmetic or clustering run alone cannot establish PASS.

Pillars: model, app

Lifecycle: operation

Scope boundary

Owns profile-qualified vector novelty findings at the declared AI text ingress, using independently authenticated attack/benign/adaptive replay and complete runtime population reconciliation. It reports findings for investigation; it does not identify attack intent, authorize an action, block a request, or claim protection for unevaluated surfaces. Benign traffic cannot be inferred from the largest current cluster. Broader languages, modalities, retrieval sources and threat families require their own measured profile.

AID-D-001.001: AID-D-001.003 detects statistical novelty in a governed vector space; AID-D-001.001 detects content, intent, injection, jailbreak, and obfuscation signals in the prompt and recovered views. A vector outlier need not be malicious, and a malicious prompt may remain near a normal embedding cluster. The vector profile must demonstrate its own useful attack triage without relying on findings from the content detector; combined evidence must not earn the same detection credit twice.

Warning

Anomaly Is Not Attack Intent

Use this method for a bounded application with representative trusted traffic and independently demonstrated security value. Novel benign inputs can trigger alerts, and adaptive malicious instructions can remain close to normal embeddings. Do not promote a detector from a percentile threshold alone, use an unavailable clean counterpart as a runtime oracle, automatically absorb live traffic into the trusted baseline, or count no cluster finding as full coverage. Keep noise, unscored records, and unsupported inputs visible; retain content analysis and execution authorization where applicable. Broader attack protection is unproven until measured in that exact deployment.

Implementation guidance

AID-D-001.003-G001: Fit and independently replay a trusted task-specific embedding baseline, qualify attack/benign/adaptive detection against signed policy, and score every applicable runtime prompt against the same generation.
AID-D-001.003-G002: Detect separated prompt campaigns against the trusted baseline rather than the largest current traffic cluster, preserving exact window coverage and individual noise scoring.

AID-D-001.003-G001

Fit and independently replay a trusted task-specific embedding baseline, qualify attack/benign/adaptive detection against signed policy, and score every applicable runtime prompt against the same generation.

Select a security-qualified task surface

Declare the application, supported text surface, languages, normal task intents, attacker query access, and prohibited instruction-override effects. For example, a customer-support assistant must not replace its assigned task with an attacker instruction to reveal protected context. Use a quarantined clone to independently observe that effect for attack cases; an unusual topic or embedding distance is not an attack label. Include legitimate topic changes, instructions quoted as data, task-preserving paraphrases, multilingual benign controls, and adaptive attacks crafted to stay near normal vectors. Keep the held-out attack/benign population separate from baseline fitting and threshold tuning. Derive acceptance rates, sample minima and latency limits from the signed environment policy, and admit this detector only for the measured profile.

Bind the detector to a complete local model tree

Embedding distance is meaningful only when baseline construction and online scoring use the same signed policy and exact Sentence Transformers directory. The policy lists every model file and its SHA-256 digest, the trusted-prompt corpus digest, clustering parameters, quantile method, and the minimum population. Resolve all artifact paths, verification keys, and timeouts only from a root-owned service definition. The code copies verified model bytes to a private read-only snapshot before loading; a single weight-file hash is not a valid Sentence Transformers model identity.

Build and publish the baseline

from __future__ import annotations

# File: detection/prompt_embedding_baseline.py

import hashlib
import hmac
import json
import math
import os
import posixpath
import re
import subprocess
import tempfile
from pathlib import Path, PurePosixPath

import numpy as np
from sentence_transformers import SentenceTransformer
from sklearn.cluster import KMeans


SHA256 = re.compile(r"^[0-9a-f]{64}$")
POLICY_FIELDS = {
    "schema_version", "policy_version", "baseline_id", "embedding_model_dir",
    "model_files", "trusted_prompt_corpus_sha256", "cluster_count",
    "minimum_trusted_prompts", "minimum_samples_per_cluster",
    "threshold_quantile", "quantile_method", "kmeans_n_init", "random_seed",
}


def write_private(path: Path, payload: bytes) -> None:
    descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(descriptor, "wb") as stream:
        stream.write(payload)
        stream.flush()
        os.fsync(stream.fileno())
    os.chmod(path, 0o400)


def strict_json(raw: bytes) -> object:
    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

    return json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_object,
        parse_constant=lambda token: (_ for _ in ()).throw(
            ValueError("non-finite JSON constant: " + token)
        ),
    )


def canonical_json_bytes(value: object) -> bytes:
    return (
        json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
        + "\n"
    ).encode("utf-8")


def capture_signed_bytes(
    payload_path: Path,
    bundle_path: Path,
    key_reference: str,
    timeout_seconds: float,
) -> bytes:
    if not math.isfinite(timeout_seconds) or timeout_seconds <= 0:
        raise ValueError("signature-verification timeout is invalid")
    payload_path = Path(payload_path)
    bundle_path = Path(bundle_path)
    if payload_path == bundle_path:
        raise ValueError("payload and bundle paths collide")
    payload_raw = payload_path.read_bytes()
    bundle_raw = bundle_path.read_bytes()
    remote_key = bool(re.match(
        r"^[A-Za-z][A-Za-z0-9+.-]*://", key_reference
    )) or key_reference.startswith("pkcs11:")
    key_raw = None if remote_key else (
        key_reference.encode("utf-8")
        if key_reference.startswith("-----BEGIN ")
        else Path(key_reference).read_bytes()
    )
    with tempfile.TemporaryDirectory(prefix="aidefend-d001003-signed-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_snapshot = root / "payload"
        bundle_snapshot = root / "payload.bundle"
        write_private(payload_snapshot, payload_raw)
        write_private(bundle_snapshot, bundle_raw)
        captured = [(payload_snapshot, payload_raw), (bundle_snapshot, bundle_raw)]
        key_argument = key_reference
        if key_raw is not None:
            key_snapshot = root / "verify-key.pub"
            write_private(key_snapshot, key_raw)
            captured.append((key_snapshot, key_raw))
            key_argument = str(key_snapshot)
        try:
            subprocess.run(
                [
                    "cosign", "verify-blob", "--key", key_argument,
                    "--bundle", str(bundle_snapshot), str(payload_snapshot),
                ],
                check=True,
                capture_output=True,
                text=True,
                timeout=timeout_seconds,
            )
            if any(
                not hmac.compare_digest(snapshot.read_bytes(), expected)
                for snapshot, expected in captured
            ):
                raise RuntimeError("private verification snapshot changed")
        finally:
            for snapshot, _ in captured:
                os.chmod(snapshot, 0o600)
    return payload_raw


def load_policy() -> tuple[dict, str]:
    timeout = float(os.environ["D001003_VERIFY_TIMEOUT_SECONDS"])
    raw = capture_signed_bytes(
        Path(os.environ["D001003_BASELINE_POLICY_PATH"]),
        Path(os.environ["D001003_BASELINE_POLICY_SIGSTORE_BUNDLE"]),
        os.environ["D001003_BASELINE_POLICY_VERIFY_KEY"],
        timeout,
    )
    policy = strict_json(raw)
    if not isinstance(policy, dict) or set(policy) != POLICY_FIELDS:
        raise ValueError("signed prompt-baseline policy fields differ")
    if (
        policy["schema_version"] != "aidefend.prompt-baseline-policy.v2"
        or any(
            not isinstance(policy[field], str) or not policy[field]
            for field in ("policy_version", "baseline_id", "embedding_model_dir")
        )
        or not isinstance(policy["trusted_prompt_corpus_sha256"], str)
        or not SHA256.fullmatch(policy["trusted_prompt_corpus_sha256"])
    ):
        raise ValueError("signed prompt-baseline identity is invalid")
    for field in (
        "cluster_count", "minimum_trusted_prompts",
        "minimum_samples_per_cluster", "kmeans_n_init",
    ):
        value = policy[field]
        minimum = 2 if field == "cluster_count" else 1
        if isinstance(value, bool) or not isinstance(value, int) or value < minimum:
            raise ValueError("signed clustering value is invalid: " + field)
    seed = policy["random_seed"]
    if isinstance(seed, bool) or not isinstance(seed, int) or seed < 0:
        raise ValueError("signed clustering random seed is invalid")
    quantile = policy["threshold_quantile"]
    if (
        isinstance(quantile, bool)
        or not isinstance(quantile, (int, float))
        or not math.isfinite(float(quantile))
        or not 0 < float(quantile) < 1
        or policy["quantile_method"] not in {"higher", "linear"}
    ):
        raise ValueError("signed distance quantile is invalid")
    return policy, hashlib.sha256(raw).hexdigest()


def relative_model_path(value: object) -> PurePosixPath:
    if not isinstance(value, str) or not value:
        raise ValueError("model-file path is absent")
    normalized = posixpath.normpath(value)
    path = PurePosixPath(value)
    if normalized != value or path.is_absolute() or any(part in {"", ".", ".."} for part in path.parts):
        raise ValueError("model-file path escapes the model root")
    return path


def snapshot_model_tree(policy: dict):
    records = policy["model_files"]
    if not isinstance(records, list) or not records:
        raise ValueError("signed model-file population is empty")
    declared = {}
    for record in records:
        if (
            not isinstance(record, dict)
            or set(record) != {"path", "sha256"}
            or not isinstance(record["sha256"], str)
            or not SHA256.fullmatch(record["sha256"])
        ):
            raise ValueError("signed model-file record is invalid")
        relative = relative_model_path(record["path"])
        name = relative.as_posix()
        if name in declared:
            raise ValueError("signed model-file path is duplicated")
        declared[name] = record["sha256"]
    if (
        not any(name.endswith(".safetensors") for name in declared)
        or any(name.endswith((".bin", ".pt", ".pth")) for name in declared)
    ):
        raise ValueError("model tree must use safetensors without pickle weights")

    source_root = Path(policy["embedding_model_dir"])
    if source_root.is_symlink():
        raise ValueError("embedding-model directory must not be a symlink")
    source_root = source_root.resolve(strict=True)
    if not source_root.is_dir():
        raise ValueError("embedding-model directory is unavailable")
    observed = {}
    for current_root, directory_names, file_names in os.walk(source_root, followlinks=False):
        current = Path(current_root)
        for name in directory_names:
            candidate = current / name
            if candidate.is_symlink() or not candidate.is_dir():
                raise ValueError("model tree contains a linked or invalid directory")
        for name in file_names:
            candidate = current / name
            if candidate.is_symlink() or not candidate.is_file():
                raise ValueError("model tree contains a linked or non-regular file")
            observed[candidate.relative_to(source_root).as_posix()] = candidate
    if set(observed) != set(declared):
        raise ValueError("observed model-file population differs from signed policy")

    directory = tempfile.TemporaryDirectory(prefix="aidefend-d001003-model-")
    snapshot_root = Path(directory.name)
    os.chmod(snapshot_root, 0o700)
    for name in sorted(observed):
        raw = observed[name].read_bytes()
        if hashlib.sha256(raw).hexdigest() != declared[name]:
            raise ValueError("model-file digest differs: " + name)
        relative = PurePosixPath(name)
        destination = snapshot_root.joinpath(*relative.parts)
        destination.parent.mkdir(parents=True, exist_ok=True)
        current = destination.parent
        while current != snapshot_root:
            os.chmod(current, 0o700)
            current = current.parent
        write_private(destination, raw)
        if not hmac.compare_digest(destination.read_bytes(), raw):
            raise RuntimeError("private model snapshot readback differs")
    tree_bytes = canonical_json_bytes([
        {"path": name, "sha256": declared[name]} for name in sorted(declared)
    ])
    return directory, snapshot_root, hashlib.sha256(tree_bytes).hexdigest()


def load_model(snapshot_root: Path) -> SentenceTransformer:
    return SentenceTransformer(
        str(snapshot_root),
        device="cpu",
        local_files_only=True,
        trust_remote_code=False,
    )


def parse_trusted_prompts(raw: bytes) -> list[str]:
    lines = raw.decode("utf-8", errors="strict").splitlines()
    if not lines or any(not line.strip() for line in lines):
        raise ValueError("trusted prompt JSONL is empty or contains a blank record")
    records = [strict_json(line.encode("utf-8")) for line in lines]
    if any(
        not isinstance(record, dict)
        or set(record) != {"prompt_id", "text"}
        or not isinstance(record["prompt_id"], str)
        or not record["prompt_id"]
        or not isinstance(record["text"], str)
        or not record["text"]
        for record in records
    ):
        raise ValueError("trusted prompt record schema differs")
    ids = [record["prompt_id"] for record in records]
    if len(ids) != len(set(ids)):
        raise ValueError("trusted prompt ID is duplicated")
    return [record["text"] for record in records]


def atomic_publish(path: Path, payload: bytes) -> None:
    parent = path.parent.resolve(strict=True)
    parent_stat = parent.stat()
    if parent_stat.st_uid != 0 or parent_stat.st_mode & 0o022:
        raise RuntimeError("baseline directory must be root-owned and non-writable")
    descriptor, name = tempfile.mkstemp(prefix="." + path.name + ".", dir=parent)
    temporary = Path(name)
    try:
        with os.fdopen(descriptor, "wb") as stream:
            stream.write(payload)
            stream.flush()
            os.fsync(stream.fileno())
        os.chmod(temporary, 0o400)
        os.replace(temporary, parent / path.name)
        directory_fd = os.open(parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
        try:
            os.fsync(directory_fd)
        finally:
            os.close(directory_fd)
        if not hmac.compare_digest((parent / path.name).read_bytes(), payload):
            raise RuntimeError("baseline readback differs")
    finally:
        temporary.unlink(missing_ok=True)


def main() -> None:
    policy, policy_sha256 = load_policy()
    corpus_raw = Path(os.environ["D001003_TRUSTED_PROMPTS_JSONL"]).read_bytes()
    if hashlib.sha256(corpus_raw).hexdigest() != policy["trusted_prompt_corpus_sha256"]:
        raise ValueError("trusted prompt corpus differs from signed policy")
    prompts = parse_trusted_prompts(corpus_raw)
    required_count = max(
        policy["minimum_trusted_prompts"],
        policy["cluster_count"] * policy["minimum_samples_per_cluster"],
    )
    if len(prompts) < required_count:
        raise ValueError("trusted prompt population is below signed policy")

    model_directory, model_root, model_tree_sha256 = snapshot_model_tree(policy)
    model = load_model(model_root)
    embeddings = np.asarray(
        model.encode(prompts, normalize_embeddings=True, show_progress_bar=False),
        dtype=np.float32,
    )
    if embeddings.ndim != 2 or len(embeddings) != len(prompts) or not np.isfinite(embeddings).all():
        raise ValueError("embedding population is incomplete or non-finite")
    fitted = KMeans(
        n_clusters=policy["cluster_count"],
        n_init=policy["kmeans_n_init"],
        random_state=policy["random_seed"],
    ).fit(embeddings)
    centroids = fitted.cluster_centers_.astype(np.float32)
    norms = np.linalg.norm(centroids, axis=1, keepdims=True)
    if np.any(norms == 0) or not np.isfinite(norms).all():
        raise ValueError("normal-intent cluster produced an invalid centroid")
    centroids /= norms
    distances = 1.0 - np.max(embeddings @ centroids.T, axis=1)
    threshold = float(np.quantile(
        distances, float(policy["threshold_quantile"]),
        method=policy["quantile_method"]
    ))
    if not math.isfinite(threshold) or not 0.0 <= threshold <= 2.0:
        raise ValueError("computed distance threshold is invalid")
    baseline = {
        "schema_version": "aidefend.prompt-cluster-baseline.v3",
        "baseline_id": policy["baseline_id"],
        "policy_version": policy["policy_version"],
        "policy_sha256": policy_sha256,
        "model_tree_sha256": model_tree_sha256,
        "centroids": centroids.tolist(),
        "embedding_dimension": int(centroids.shape[1]),
        "distance_threshold": threshold,
        "threshold_quantile": float(policy["threshold_quantile"]),
        "quantile_method": policy["quantile_method"],
        "sample_count": len(prompts),
        "cluster_count": int(policy["cluster_count"]),
        "trusted_prompt_input_sha256": sorted({hashlib.sha256(text.encode("utf-8")).hexdigest() for text in prompts}),
    }
    atomic_publish(
        Path(os.environ["D001003_BASELINE_OUTPUT"]),
        canonical_json_bytes(baseline),
    )
    model_directory.cleanup()


if __name__ == "__main__":
    main()

set -euo pipefail
test -n "$D001003_BASELINE_OUTPUT"
baseline="$D001003_BASELINE_OUTPUT"
cosign sign-blob --yes --key env://D001003_BASELINE_SIGNING_KEY \
  --bundle "$baseline.bundle" "$baseline"

Score against the same verified generation

Install the security-replay verifier shown below before starting this service. Startup captures the separately signed profile, complete case manifest and observer results; absent, forged, misbound or ineffective qualification prevents this model/baseline/policy generation from serving detector results. Provide the D001003_SECURITY_PROFILE, D001003_SECURITY_CASES and D001003_SECURITY_RESULTS PATH, SIGSTORE_BUNDLE and VERIFY_KEY values, plus D001003_APPLICATION_ID and D001003_SURFACE_ID, only through the protected service definition. The replay worker uses the same pinned embedding implementation in the quarantined evaluation environment; qualification does not depend on a production service already running.

from __future__ import annotations

# File: detection/prompt_embedding_distance.py

import hashlib
import math
import os
from pathlib import Path

import numpy as np

from detection.prompt_embedding_baseline import (
    capture_signed_bytes,
    load_model,
    load_policy,
    snapshot_model_tree,
    strict_json,
)


BASELINE_FIELDS = {
    "schema_version", "baseline_id", "policy_version", "policy_sha256",
    "model_tree_sha256", "centroids", "embedding_dimension",
    "distance_threshold", "threshold_quantile", "quantile_method",
    "sample_count", "cluster_count", "trusted_prompt_input_sha256",
}

POLICY, POLICY_SHA256 = load_policy()
MODEL_DIRECTORY, MODEL_ROOT, MODEL_TREE_SHA256 = snapshot_model_tree(POLICY)
MODEL = load_model(MODEL_ROOT)
BASELINE_RAW = capture_signed_bytes(
    Path(os.environ["D001003_BASELINE_PATH"]),
    Path(os.environ["D001003_BASELINE_SIGSTORE_BUNDLE"]),
    os.environ["D001003_BASELINE_VERIFY_KEY"],
    float(os.environ["D001003_VERIFY_TIMEOUT_SECONDS"]),
)
BASELINE = strict_json(BASELINE_RAW)
if (
    not isinstance(BASELINE, dict)
    or set(BASELINE) != BASELINE_FIELDS
    or BASELINE["schema_version"] != "aidefend.prompt-cluster-baseline.v3"
    or BASELINE["baseline_id"] != POLICY["baseline_id"]
    or BASELINE["policy_version"] != POLICY["policy_version"]
    or BASELINE["policy_sha256"] != POLICY_SHA256
    or BASELINE["model_tree_sha256"] != MODEL_TREE_SHA256
    or BASELINE["cluster_count"] != POLICY["cluster_count"]
    or BASELINE["threshold_quantile"] != float(POLICY["threshold_quantile"])
    or BASELINE["quantile_method"] != POLICY["quantile_method"]
    or isinstance(BASELINE["sample_count"], bool)
    or not isinstance(BASELINE["sample_count"], int)
    or BASELINE["sample_count"] < POLICY["minimum_trusted_prompts"]
):
    raise ValueError("signed baseline identity or model binding differs")
CENTROIDS = np.asarray(BASELINE["centroids"], dtype=np.float32)
if (
    isinstance(BASELINE["cluster_count"], bool)
    or not isinstance(BASELINE["cluster_count"], int)
    or isinstance(BASELINE["embedding_dimension"], bool)
    or not isinstance(BASELINE["embedding_dimension"], int)
    or CENTROIDS.shape != (
        BASELINE["cluster_count"], BASELINE["embedding_dimension"]
    )
    or not np.isfinite(CENTROIDS).all()
    or not np.allclose(np.linalg.norm(CENTROIDS, axis=1), 1.0, atol=1e-4)
):
    raise ValueError("signed baseline dimensions are invalid")
THRESHOLD = BASELINE["distance_threshold"]
if (
    isinstance(THRESHOLD, bool)
    or not isinstance(THRESHOLD, (int, float))
    or not math.isfinite(float(THRESHOLD))
    or not 0.0 <= float(THRESHOLD) <= 2.0
):
    raise ValueError("signed distance threshold is invalid")
BASELINE_SHA256 = hashlib.sha256(BASELINE_RAW).hexdigest()

from detection.vector_security_replay import require_verified_security_replay
from detection.language_admission import LanguageAdmission

def captured_security_document(prefix):
    raw = capture_signed_bytes(
        Path(os.environ[prefix + "_PATH"]),
        Path(os.environ[prefix + "_SIGSTORE_BUNDLE"]),
        os.environ[prefix + "_VERIFY_KEY"],
        float(os.environ["D001003_VERIFY_TIMEOUT_SECONDS"]),
    )
    return strict_json(raw)

# Paths, role-pinned trust keys and application/surface IDs come only from the
# protected service definition, never from a prompt or from the replay document.
SECURITY_PROFILE = captured_security_document("D001003_SECURITY_PROFILE")
SECURITY_CASES = captured_security_document("D001003_SECURITY_CASES")
SECURITY_RESULTS = captured_security_document("D001003_SECURITY_RESULTS")
LANGUAGE_ADMISSION = LanguageAdmission(SECURITY_PROFILE)
QUALIFICATION = require_verified_security_replay(
    SECURITY_CASES, SECURITY_RESULTS, SECURITY_PROFILE,
    {"application_id": os.environ["D001003_APPLICATION_ID"],
     "surface_id": os.environ["D001003_SURFACE_ID"],
     "model_tree_sha256": MODEL_TREE_SHA256,
     "baseline_sha256": BASELINE_SHA256, "policy_sha256": POLICY_SHA256,
     "language_profile_sha256": LANGUAGE_ADMISSION.profile_sha256},
    BASELINE["trusted_prompt_input_sha256"],
)



def score_prompt(prompt_id: str, prompt: str) -> dict:
    if not isinstance(prompt_id, str) or not prompt_id or not isinstance(prompt, str) or not prompt:
        raise ValueError("prompt identity or text is invalid")
    language = LANGUAGE_ADMISSION.inspect(prompt)
    if language["language_status"] != "SUPPORTED":
        return {"schema_version": "aidefend.prompt-distance-result.v2",
            "application_id": os.environ["D001003_APPLICATION_ID"],
            "surface_id": os.environ["D001003_SURFACE_ID"],
            "processing_status": language["language_status"], "finding_status": "UNKNOWN",
            "prompt_id": prompt_id, "prompt_sha256": hashlib.sha256(prompt.encode("utf-8")).hexdigest(),
            "baseline_id": BASELINE["baseline_id"], "baseline_sha256": BASELINE_SHA256,
            "policy_sha256": POLICY_SHA256, "model_tree_sha256": MODEL_TREE_SHA256, **language}
    vector = np.asarray(
        MODEL.encode([prompt], normalize_embeddings=True, show_progress_bar=False)[0],
        dtype=np.float32,
    )
    if vector.shape != (BASELINE["embedding_dimension"],) or not np.isfinite(vector).all():
        raise ValueError("runtime embedding is incomplete or non-finite")
    distances = 1.0 - CENTROIDS @ vector
    if not np.isfinite(distances).all():
        raise ValueError("runtime distances are non-finite")
    nearest = int(np.argmin(distances))
    distance = float(distances[nearest])
    finding = distance > float(THRESHOLD)
    return {
        "schema_version": "aidefend.prompt-distance-result.v2",
        "application_id": os.environ["D001003_APPLICATION_ID"],
        "surface_id": os.environ["D001003_SURFACE_ID"],
        "processing_status": "SCORED",
        "finding_status": "FINDING" if finding else "NO_FINDING",
        "prompt_id": prompt_id,
        "prompt_sha256": hashlib.sha256(prompt.encode("utf-8")).hexdigest(),
        "baseline_id": BASELINE["baseline_id"],
        "baseline_sha256": BASELINE_SHA256,
        "policy_sha256": POLICY_SHA256,
        "model_tree_sha256": MODEL_TREE_SHA256,
        "nearest_cluster": nearest,
        "distance": distance,
        "threshold": float(THRESHOLD),
        **language,
    }


def safe_score_prompt(prompt_id: str, prompt: str) -> dict:
    try:
        return score_prompt(prompt_id, prompt)
    except Exception as error:
        value = prompt if isinstance(prompt, str) else ""
        return {
            "schema_version": "aidefend.prompt-distance-result.v2",
        "application_id": os.environ["D001003_APPLICATION_ID"],
        "surface_id": os.environ["D001003_SURFACE_ID"],
            "processing_status": "ERROR",
            "finding_status": "UNKNOWN",
            "prompt_id": prompt_id,
            "prompt_sha256": hashlib.sha256(value.encode("utf-8")).hexdigest(),
            "baseline_sha256": BASELINE_SHA256,
            "error_class": type(error).__name__,
            "language_status": "ERROR", "language": None,
            "language_profile_sha256": LANGUAGE_ADMISSION.profile_sha256,
        }

Replay before promotion and in operation

Replay signed normal-intent, boundary-distance, far-outlier, multilingual, empty, malformed, and non-finite fixtures. Also test a missing or extra model file, a changed model byte, symlinked model content, corrupt policy or baseline bundle, baseline-policy mismatch, an undersized trusted population, and a runtime dimension mismatch. Independently rebuild the baseline from the same captured corpus/model generation and compare its canonical digest, then re-embed sampled prompts and recompute distances. Run the scorer for every applicable prompt and retain the result in the authenticated event path; a model, baseline, or scoring failure is ERROR, never a clean result.

Independently qualify detection and benign utility

The following reusable verifier evaluates a complete held-out case population. The release service captures the profile and case manifest with capture_signed_bytes using separately trusted policy/test-authority keys, and captures result rows from the independently credentialed replay observer with its own verification key; parse each captured byte sequence with strict_json. These signatures bind the actual target, input and unsafe-effect evidence; a digest supplied by the detector itself is not an independent witness. Populate each result from safe_score_prompt: map a scored FINDING/NO_FINDING to observation, preserve processing failures as UNKNOWN, and measure latency around the real embedding invocation. Runtime fields do not decide control scores. Never manufacture model observations or treat an unavailable production model as a successful attack replay.

from __future__ import annotations

# File: detection/vector_security_replay.py
import math
from detection.language_admission import validate_language_profile


def require_verified_security_replay(cases, results, profile, expected_context, baseline_input_sha256):
    fields = {"application_id", "surface_id", "model_tree_sha256", "baseline_sha256", "policy_sha256", "language_profile_sha256"}
    if set(expected_context) != fields or any(profile.get(name) != value for name, value in expected_context.items()):
        raise PermissionError("security replay is for another deployed generation or surface")
    result = evaluate_security_replay(cases, results, profile)
    if result["selection"] != "ALLOWED_FOR_PROFILE":
        raise PermissionError("generation has no complete effective security qualification")

    prompt_digests = baseline_input_sha256
    if (not isinstance(prompt_digests, list) or not prompt_digests
            or len(set(prompt_digests)) != len(prompt_digests)
            or any(not isinstance(value, str) or len(value) != 64 or
                   any(character not in "0123456789abcdef" for character in value) for value in prompt_digests)):
        raise ValueError("signed baseline input population is unavailable")
    if set(prompt_digests) & {case.get("input_sha256") for case in cases}:
        raise PermissionError("security holdout inputs overlap the trusted baseline corpus")
    return result


def evaluate_security_replay(cases, results, profile):
    """Consume authenticated, independently observed case/result populations, not self-reported detector labels."""
    context = ("application_id", "surface_id", "model_tree_sha256", "baseline_sha256", "policy_sha256", "language_profile_sha256")
    required = set(context) | {"profile_version", "minimum_attack_cases", "minimum_benign_cases",
                               "minimum_adaptive_cases", "minimum_attack_recall",
                               "minimum_adaptive_recall", "maximum_benign_flag_rate", "maximum_latency_ms",
                               "minimum_detection_advantage", "supported_languages", "language_identifier"}
    if not isinstance(profile, dict) or set(profile) != required:
        raise ValueError("security profile schema differs")
    validate_language_profile(profile)
    for name in context + ("profile_version",):
        if not isinstance(profile[name], str) or not profile[name]:
            raise ValueError("missing context binding")
    for name in ("minimum_attack_cases", "minimum_benign_cases", "minimum_adaptive_cases"):
        if type(profile[name]) is not int or profile[name] < 1:
            raise ValueError("invalid case minimum")
    for name in ("minimum_attack_recall", "minimum_adaptive_recall", "maximum_benign_flag_rate", "maximum_latency_ms", "minimum_detection_advantage"):
        value = profile[name]
        if type(value) not in (int, float) or not math.isfinite(value) or value < 0:
            raise ValueError("invalid acceptance threshold")
        if name != "maximum_latency_ms" and value > 1:
            raise ValueError("rate outside unit interval")
    if (profile["maximum_latency_ms"] <= 0 or profile["minimum_attack_recall"] <= 0 or
            profile["minimum_adaptive_recall"] <= 0 or profile["maximum_benign_flag_rate"] >= 1 or
            profile["minimum_detection_advantage"] <= 0):
        raise ValueError("vacuous detection/utility policy is inadmissible")
    if not isinstance(cases, list) or not isinstance(results, list):
        raise ValueError("case/result population must be lists")
    expected = [case["case_id"] for case in cases]
    observed = [result["case_id"] for result in results]
    if any(not isinstance(value, str) or not value for value in expected + observed):
        raise ValueError("invalid case ID")
    if len(set(expected)) != len(expected) or len(set(observed)) != len(observed):
        raise ValueError("duplicate case ID")
    if set(expected) != set(observed):
        return {"selection": "INCOMPLETE", "reason": "independent case-ID population differs"}
    by_id = {result["case_id"]: result for result in results}
    attacks, benign, adaptive, latencies = [], [], [], []
    for case in cases:
        result = by_id[case["case_id"]]
        if any(case.get(name) != profile[name] or result.get(name) != profile[name] for name in context):
            raise ValueError("application/surface/model/baseline/policy binding differs")
        digest = case.get("input_sha256")
        if not isinstance(digest, str) or len(digest) != 64 or any(c not in "0123456789abcdef" for c in digest) or result.get("input_sha256") != digest:
            raise ValueError("case input binding differs")
        if type(case.get("adaptive")) is not bool or case.get("kind") not in ("ATTACK", "BENIGN"):
            raise ValueError("invalid case classification")
        if case["kind"] == "BENIGN" and case["adaptive"]:
            raise ValueError("adaptive case must be an attack")
        common_case_fields = set(context) | {"case_id", "input_sha256", "kind", "adaptive", "language"}
        expected_case_fields = common_case_fields | ({"threat", "unsafe_effect_evidence_sha256"} if case["kind"] == "ATTACK" else set())
        if set(case) != expected_case_fields:
            raise ValueError("case schema differs; benign cases cannot carry attack-only facts")
        if (case["language"] not in profile["supported_languages"]
                or result.get("language_status") != "SUPPORTED" or result.get("language") != case["language"]):
            return {"selection": "INCOMPLETE", "reason": "language population outside qualified profile or unobserved"}
        if case["kind"] == "ATTACK":
            # An independent sandbox replay must have observed the prohibited effect
            # on the same target without this detector. An outlier label is no oracle.
            evidence = case.get("unsafe_effect_evidence_sha256")
            if (case.get("threat") != "DIRECT_INSTRUCTION_OVERRIDE" or not isinstance(evidence, str) or
                    len(evidence) != 64 or any(c not in "0123456789abcdef" for c in evidence)):
                raise ValueError("attack lacks observed security consequence")
        if result.get("observation") not in ("FINDING", "NO_FINDING"):
            return {"selection": "INCOMPLETE", "reason": "unsupported or unscored case"}
        latency = result.get("latency_ms")
        if type(latency) not in (int, float) or not math.isfinite(latency) or latency < 0:
            raise ValueError("invalid measured latency")
        latencies.append(float(latency))
        flagged = result["observation"] == "FINDING"
        (attacks if case["kind"] == "ATTACK" else benign).append(flagged)
        if case["adaptive"]:
            adaptive.append(flagged)
    if (len(attacks) < profile["minimum_attack_cases"] or len(benign) < profile["minimum_benign_cases"] or
            len(adaptive) < profile["minimum_adaptive_cases"]):
        return {"selection": "INCOMPLETE", "reason": "required attack/benign/adaptive population absent"}
    language_metrics = {}
    language_eligible = True
    for language in profile["supported_languages"]:
        language_cases = [case for case in cases if case["language"] == language]
        attack = [by_id[case["case_id"]]["observation"] == "FINDING" for case in language_cases if case["kind"] == "ATTACK"]
        clean = [by_id[case["case_id"]]["observation"] == "FINDING" for case in language_cases if case["kind"] == "BENIGN"]
        adaptive_language = [by_id[case["case_id"]]["observation"] == "FINDING" for case in language_cases if case["adaptive"]]
        if (len(attack) < profile["minimum_attack_cases"] or len(clean) < profile["minimum_benign_cases"]
                or len(adaptive_language) < profile["minimum_adaptive_cases"]):
            return {"selection": "INCOMPLETE", "reason": "declared language has incomplete attack/benign/adaptive evidence", "language": language}
        measured = {"attack_recall": sum(attack) / len(attack),
            "adaptive_recall": sum(adaptive_language) / len(adaptive_language),
            "benign_flag_rate": sum(clean) / len(clean)}
        language_metrics[language] = measured
        language_eligible = language_eligible and (
            measured["attack_recall"] >= profile["minimum_attack_recall"]
            and measured["adaptive_recall"] >= profile["minimum_adaptive_recall"]
            and measured["benign_flag_rate"] <= profile["maximum_benign_flag_rate"]
            and measured["attack_recall"] - measured["benign_flag_rate"] >= profile["minimum_detection_advantage"]
            and measured["adaptive_recall"] - measured["benign_flag_rate"] >= profile["minimum_detection_advantage"])
    metrics = {"attack_recall": sum(attacks) / len(attacks), "adaptive_recall": sum(adaptive) / len(adaptive),
               "benign_flag_rate": sum(benign) / len(benign), "maximum_latency_ms": max(latencies)}
    admitted = (metrics["attack_recall"] >= profile["minimum_attack_recall"] and
                metrics["adaptive_recall"] >= profile["minimum_adaptive_recall"] and
                metrics["benign_flag_rate"] <= profile["maximum_benign_flag_rate"] and
                metrics["maximum_latency_ms"] <= profile["maximum_latency_ms"] and
                metrics["attack_recall"] - metrics["benign_flag_rate"] >= profile["minimum_detection_advantage"] and
                metrics["adaptive_recall"] - metrics["benign_flag_rate"] >= profile["minimum_detection_advantage"])
    return {"selection": "ALLOWED_FOR_PROFILE" if admitted and language_eligible else "REJECTED_FOR_PROFILE",
            "profile_version": profile["profile_version"], "metrics": metrics, "language_metrics": language_metrics,
            "case_ids": sorted(expected), "claim": "input anomaly triage; no attack-intent or blocking claim"}

Call evaluate_security_replay(verified_cases, verified_results, verified_profile) before selecting a model/baseline/policy generation for the declared surface, and repeat after any of them changes. ALLOWED_FOR_PROFILE only qualifies this method for that tested profile. Independently reconcile runtime prompt IDs and input digests against ingress observations; lost rows, unsupported inputs and unscored cases cannot inherit the held-out result. No alarm does not prove no injection.

Admit only measured input languages before embedding

The signed security profile must declare supported_languages as ISO 639-1 codes and a language_identifier object with package, version, installed payload artifact_sha256, minimum confidence and confidence margin, minimum letter count and maximum input characters. Compute its language_profile_sha256 using the function below before independently signing the profile. Install the reviewed lingua-language-detector wheel in the immutable service image; its native payload includes the identifier models. Record language_artifact_sha256() and the installed package version in that profile. Thresholds come from replay on the actual application, including supported, unsupported, mixed, short and adversarial text. Do not build the identifier from only approved languages, which can force out-of-profile inputs toward an approved label.

# File: detection/language_admission.py
from __future__ import annotations

import hashlib
from importlib import metadata
import json
import math
from pathlib import Path
import re

import lingua
from lingua import Language, LanguageDetectorBuilder

IDENTIFIER_FIELDS = {"package", "version", "artifact_sha256", "minimum_confidence",
    "minimum_confidence_margin", "minimum_letter_count", "maximum_input_characters"}


def language_artifact_sha256() -> str:
    # Native Lingua wheels embed the language models in their extension module.
    # Hash the actual installed payload tree, including that binary, once at startup.
    # Install it into the immutable service image; no request may select this path.
    root = Path(lingua.__file__).parent.resolve(strict=True)
    digest = hashlib.sha256()
    files = sorted(path for path in root.rglob("*")
                   if path.is_file() and "__pycache__" not in path.parts and path.suffix != ".pyc")
    if not files or any(path.is_symlink() for path in root.rglob("*")):
        raise ValueError("language identifier payload is missing or linked")
    for path in files:
        content = hashlib.sha256()
        with path.open("rb") as source:
            while chunk := source.read(1024 * 1024):
                content.update(chunk)
        digest.update(path.relative_to(root).as_posix().encode("utf-8") + b"\0")
        digest.update(content.digest())
    return digest.hexdigest()


def language_profile_sha256(profile: dict) -> str:
    value = {name: profile[name] for name in ("supported_languages", "language_identifier")}
    return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":"),
        ensure_ascii=False, allow_nan=False).encode("utf-8")).hexdigest()


def validate_language_profile(profile: dict) -> None:
    languages = profile.get("supported_languages")
    identifier = profile.get("language_identifier")
    if (not isinstance(languages, list) or not languages or len(set(languages)) != len(languages)
            or any(not isinstance(value, str) or not re.fullmatch("[a-z]{2}", value) for value in languages)):
        raise ValueError("supported languages must be unique ISO 639-1 codes")
    if not isinstance(identifier, dict) or set(identifier) != IDENTIFIER_FIELDS:
        raise ValueError("signed language identifier schema differs")
    if (identifier["package"] != "lingua-language-detector"
            or not isinstance(identifier["version"], str) or not identifier["version"]
            or not isinstance(identifier["artifact_sha256"], str)
            or not re.fullmatch("[a-f0-9]{64}", identifier["artifact_sha256"])):
        raise ValueError("language identifier identity is not pinned")
    for name in ("minimum_confidence", "minimum_confidence_margin"):
        value = identifier[name]
        if type(value) not in (int, float) or not math.isfinite(value) or not 0 < value <= 1:
            raise ValueError("language confidence policy is invalid")
    for name in ("minimum_letter_count", "maximum_input_characters"):
        if type(identifier[name]) is not int or identifier[name] < 1:
            raise ValueError("language input-length policy is invalid")
    if identifier["minimum_letter_count"] > identifier["maximum_input_characters"]:
        raise ValueError("language input-length policy is inconsistent")
    if profile.get("language_profile_sha256") != language_profile_sha256(profile):
        raise ValueError("language profile digest differs from signed configuration")


class LanguageAdmission:
    def __init__(self, authenticated_profile: dict):
        validate_language_profile(authenticated_profile)
        self.configuration = dict(authenticated_profile["language_identifier"])
        self.supported = frozenset(authenticated_profile["supported_languages"])
        self.profile_sha256 = authenticated_profile["language_profile_sha256"]
        if metadata.version("lingua-language-detector") != self.configuration["version"]:
            raise PermissionError("installed language identifier version differs")
        if language_artifact_sha256() != self.configuration["artifact_sha256"]:
            raise PermissionError("installed language identifier/model artifact differs")
        available = {language.iso_code_639_1.name.lower() for language in Language.all()}
        if not self.supported.issubset(available):
            raise ValueError("declared language is unavailable to the pinned identifier")
        # Include all identifier languages, not only the approved ones: restricting
        # classification to English would force unsupported languages toward English.
        self.detector = LanguageDetectorBuilder.from_all_languages().build()

    def inspect(self, text: str) -> dict:
        result = {"language_status": "UNKNOWN", "language": None,
            "language_profile_sha256": self.profile_sha256}
        if not isinstance(text, str) or not text:
            raise ValueError("language input is not nonempty text")
        if (len(text) > self.configuration["maximum_input_characters"]
                or sum(character.isalpha() for character in text) < self.configuration["minimum_letter_count"]):
            return {**result, "language_reason": "unqualified_length_or_nonlinguistic_input"}
        confidence = self.detector.compute_language_confidence_values(text)
        if not confidence:
            return {**result, "language_reason": "identifier_returned_no_language"}
        winner = confidence[0]
        runner_up = confidence[1].value if len(confidence) > 1 else 0.0
        values = (winner.value, runner_up)
        if any(not math.isfinite(value) or not 0 <= value <= 1 for value in values):
            raise RuntimeError("language identifier returned an invalid confidence")
        language = winner.language.iso_code_639_1.name.lower()
        result.update(language=language, language_confidence=winner.value,
                      language_confidence_margin=winner.value - runner_up)
        if (winner.value < self.configuration["minimum_confidence"]
                or winner.value - runner_up < self.configuration["minimum_confidence_margin"]):
            return {**result, "language_reason": "identifier_confidence_below_policy"}
        sections = self.detector.detect_multiple_languages_of(text)
        section_languages = {section.language.iso_code_639_1.name.lower() for section in sections}
        if language not in self.supported or section_languages - self.supported:
            return {**result, "language_status": "UNSUPPORTED", "language_reason": "language_outside_qualified_profile"}
        if not sections or section_languages != {language}:
            return {**result, "language_reason": "mixed_or_unresolved_language_sections"}
        return {**result, "language_status": "SUPPORTED", "language_reason": "qualified_monolingual_input"}

Run this guard before encoding for either individual or clustered scoring. UNSUPPORTED, UNKNOWN and ERROR remain uncovered observations with original input identities; they cannot become NO_FINDING or contribute to complete measured coverage. Language identification is probabilistic and mixed-language segmentation is imperfect; this implementation qualifies monolingual inputs only and makes no universal language-support guarantee. Independent replay binds the identifier/configuration digest and observes false acceptance and exclusions. The signed baseline now uses aidefend.prompt-cluster-baseline.v3 to include the actual trusted-prompt input digest population; retained security cases must be disjoint from that population. Runtime results use aidefend.prompt-distance-result.v2 to expose language coverage.

Each declared supported language must independently meet the signed minimum attack, benign and adaptive population requirements and the same detection/utility thresholds. Evidence from one language cannot qualify another, and aggregate performance cannot hide an ineffective language subgroup.

AID-D-001.003-G002

Detect separated prompt campaigns against the trusted baseline rather than the largest current traffic cluster, preserving exact window coverage and individual noise scoring.

Capture a complete trusted-generation window

Use the verified model and baseline from AID-D-001.003-G001. After signature verification of the clustering policy and independent ingress window manifest, pass the baseline projection containing the captured baseline digest, model-tree digest and centroids, the policy, and the exact expected window IDs into this function. Do not let a majority of attacker traffic define normal behavior, and never update the trusted baseline from this window. Use canonical ID order so membership replay does not depend on arrival order. Noise is explicitly handed to the individual scorer; no dense cluster is not a clean-traffic claim.

from __future__ import annotations

# File: detection/emergent_prompt_clusters.py
import hashlib
import json
import math
from dataclasses import dataclass

import numpy as np
from sklearn.cluster import DBSCAN
from detection.language_admission import LanguageAdmission


@dataclass(frozen=True)
class EmbeddingEvent:
    event_id: str
    vector: list[float] | None
    language_status: str
    language: str | None
    language_profile_sha256: str


def build_embedding_event(event_id: str, prompt: str, admission: LanguageAdmission, model) -> EmbeddingEvent:
    if not isinstance(event_id, str) or not event_id or not isinstance(admission, LanguageAdmission):
        raise ValueError("embedding producer identity or language adapter differs")
    coverage = admission.inspect(prompt)
    vector = None
    if coverage["language_status"] == "SUPPORTED":
        vector = np.asarray(model.encode([prompt], normalize_embeddings=True, show_progress_bar=False)[0], dtype=np.float64).tolist()
    return EmbeddingEvent(event_id, vector, coverage["language_status"], coverage["language"], admission.profile_sha256)


def detect_emergent_clusters(events, baseline, policy, expected_event_ids, admission: LanguageAdmission):
    """Inputs are captured from verified baseline/policy and an independent window manifest."""
    required = {"policy_version", "baseline_sha256", "model_tree_sha256", "eps",
                "min_samples", "minimum_population", "minimum_emergent_cluster_size",
                "minimum_centroid_separation", "language_profile_sha256", "supported_languages"}
    if not isinstance(policy, dict) or set(policy) != required:
        raise ValueError("clustering policy schema mismatch")
    for name in ("min_samples", "minimum_population", "minimum_emergent_cluster_size"):
        if type(policy[name]) is not int or policy[name] < 1:
            raise ValueError("invalid population requirement")
    for name in ("eps", "minimum_centroid_separation"):
        if type(policy[name]) not in (int, float) or not math.isfinite(policy[name]) or not 0 < policy[name] <= 2:
            raise ValueError("invalid cosine-distance threshold")
    if not isinstance(policy["policy_version"], str) or not policy["policy_version"]:
        raise ValueError("missing policy version")
    for name in ("baseline_sha256", "model_tree_sha256", "language_profile_sha256"):
        if not isinstance(policy[name], str) or len(policy[name]) != 64 or any(c not in "0123456789abcdef" for c in policy[name]):
            raise ValueError("invalid artifact digest")
    if (baseline["baseline_sha256"] != policy["baseline_sha256"] or
            baseline["model_tree_sha256"] != policy["model_tree_sha256"]):
        raise ValueError("baseline/model generation differs")
    languages = policy["supported_languages"]
    if not isinstance(languages, list) or not languages or len(set(languages)) != len(languages) or any(not isinstance(value, str) or not value for value in languages):
        raise ValueError("supported clustering language policy differs")
    if (not isinstance(admission, LanguageAdmission)
            or admission.profile_sha256 != policy["language_profile_sha256"]
            or admission.supported != frozenset(languages)):
        raise ValueError("clustering scope differs from the actual authenticated language adapter")
    expected = list(expected_event_ids)
    actual = [event.event_id for event in events]
    if any(not isinstance(value, str) or not value for value in expected + actual):
        raise ValueError("invalid event identity")
    if len(expected) != len(set(expected)) or len(actual) != len(set(actual)):
        raise ValueError("duplicate event identity")
    if set(actual) != set(expected):
        raise ValueError("independent window population differs")
    identity = {name: policy[name] for name in ("policy_version", "baseline_sha256", "model_tree_sha256", "language_profile_sha256")}
    identity["population_sha256"] = hashlib.sha256(
        json.dumps(sorted(expected), separators=(",", ":")).encode()).hexdigest()
    if not events:
        return {**identity, "observation": "UNAVAILABLE", "reason": "empty window", "event_count": 0}
    for event in events:
        if event.language_profile_sha256 != policy["language_profile_sha256"]:
            raise ValueError("event language profile differs from signed policy")
        if event.language_status not in {"SUPPORTED", "UNSUPPORTED", "UNKNOWN", "ERROR"}:
            raise ValueError("event language observation state differs")
        if event.language_status == "SUPPORTED" and event.language not in languages:
            raise ValueError("supported event declares an unqualified language")
    uncovered = sorted(event.event_id for event in events if event.language_status != "SUPPORTED")
    if uncovered:
        return {**identity, "observation": "UNAVAILABLE", "reason": "complete window language coverage absent",
            "event_count": len(events), "uncovered_event_ids": uncovered}
    ordered = sorted(events, key=lambda event: event.event_id)
    vectors = np.asarray([event.vector for event in ordered], dtype=np.float64)
    centroids = np.asarray(baseline["centroids"], dtype=np.float64)
    if (vectors.ndim != 2 or centroids.ndim != 2 or not len(centroids) or
            vectors.shape[1] != centroids.shape[1] or not vectors.shape[1] or
            not np.isfinite(vectors).all() or not np.isfinite(centroids).all()):
        raise ValueError("invalid embedding/baseline dimensions")
    norms = np.linalg.norm(vectors, axis=1)
    if np.any(norms <= 0) or not np.allclose(np.linalg.norm(centroids, axis=1), 1.0, atol=1e-4):
        raise ValueError("zero vector or unnormalised trusted centroid")
    vectors = vectors / norms[:, None]
    if len(events) < policy["minimum_population"]:
        return {**identity, "observation": "UNAVAILABLE", "reason": "window below policy minimum",
                "event_count": len(events)}
    labels = DBSCAN(eps=policy["eps"], min_samples=policy["min_samples"],
                    metric="cosine", n_jobs=1).fit_predict(vectors)
    groups, findings = [], []
    for label in sorted(set(labels) - {-1}):
        indexes = np.flatnonzero(labels == label)
        mean = vectors[indexes].mean(axis=0)
        norm = float(np.linalg.norm(mean))
        if norm <= 1e-12:
            raise ValueError("cluster has no stable centroid")
        distance = float(np.clip(1.0 - np.max(centroids @ (mean / norm)), 0, 2))
        members = [ordered[index].event_id for index in indexes]
        flagged = len(indexes) >= policy["minimum_emergent_cluster_size"] and distance >= policy["minimum_centroid_separation"]
        groups.append({"event_ids": members, "nearest_baseline_distance": distance,
                       "finding": bool(flagged)})
        if flagged:
            findings.extend(members)
    # Noise is retained for individual scoring under G001, never silently counted as clean.
    noise = [ordered[index].event_id for index in np.flatnonzero(labels == -1)]
    return {**identity, "observation": "FINDING" if findings else "NO_CLUSTER_FINDING",
            "event_count": len(events), "clusters": groups, "finding_event_ids": sorted(findings),
            "individual_scoring_event_ids": noise}

Verify before investigation

Replay normal traffic, a separated campaign larger than benign traffic, several normal-intent clusters, outliers labeled as noise, reordered arrivals, duplicates, missing/extra IDs, zero and non-finite vectors, a wrong model generation, and an undersized window. Independently recompute memberships and baseline distances with the pinned scikit-learn and BLAS build. Evaluate actual attack/benign/adaptive cases with the G001 security profile before treating these campaign findings as security-relevant. Forward findings and noise-scoring results to the authenticated investigative event path; never block or label attack intent from cluster membership alone.

Keep language coverage with the complete input population

Before emitting an embedding event, run the same LanguageAdmission guard on its actual prompt using the independently signed security profile. Preserve each original event identity and language-profile digest, including unsupported, unknown or error dispositions. Only supported inputs may reach the encoder. Reconcile these dispositions against the independent window manifest; a clustering result over the supported subset is partial coverage, and the full window cannot be reported clean when any input remains uncovered. A window that requires complete language coverage must reject such inputs as unavailable rather than silently removing their IDs.

Pass the same independently authenticated LanguageAdmission instance as the required admission argument to detect_emergent_clusters. The consumer checks its actual supported language set and configuration digest against the signed clustering policy before assessing the window.

Open-source tools

sentence-transformers
scikit-learn
FAISS
Chroma
Weaviate
Milvus
Qdrant
PyOD
Lingua (lingua-language-detector)
Cosign (signed baseline and replay evidence verification)

Commercial / hosted tools

Arize AX
Fiddler AI Observability
Evidently Cloud
NannyML Cloud
Pinecone
Zilliz Cloud

Threat mappings

MITRE ATLAS

AML.T0051 LLM Prompt Injection (profile-qualified anomaly triage of direct instruction overrides)
AML.T0051.000 LLM Prompt Injection: Direct (profile-qualified anomaly triage of direct instruction overrides)

MAESTRO

OWASP LLM Top 10 2026

LLM01:2026 Prompt Injection (profile-qualified anomaly triage of direct instruction overrides)

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

NISTAML.018 Prompt Injection (profile-qualified anomaly triage of direct instruction overrides)

Cisco Integrated AI Security and Safety Framework

AITech-1.1 Direct Prompt Injection and Jailbreak (Prompt-based Manipulation) (profile-qualified anomaly triage of direct instruction overrides)
AISubtech-1.1.1 Instruction Manipulation (profile-qualified anomaly triage of direct instruction overrides)

Google Secure AI Framework 2.0 - Risks

PIJ: Prompt Injection (profile-qualified anomaly triage of direct instruction overrides)

Databricks AI Security Framework 3.0

Model Serving - Inference requests 9.1: Prompt inject (profile-qualified anomaly triage of direct instruction overrides)