AID-D-002.001: Input / Output Distribution Drift Monitoring

Version 1.20261015 · Actionable sub-technique

Compare live feature, prediction, or token-probability distributions against trusted reference baselines to detect data drift, concept drift, poisoned traffic, uncertainty shifts, and subtle degradation in model behavior. This sub-technique applies to classical ML services and to self-hosted or vendor-hosted inference APIs that expose a complete, documented top-k log-probability signal.

Pillars: data, model

Lifecycle: operation

Scope boundary

Owns unlabeled production drift findings from live input-feature, prediction-output, and provider- or runtime-exposed token-probability distributions compared with signed reference distributions. It does not claim access to hidden logits that the serving interface does not return.

AID-D-002.002: AID-D-002.001 detects distribution shifts without requiring outcome labels; AID-D-002.002 measures actual predictive quality after joining predictions to trusted ground truth. Distribution stability does not prove correctness, and delayed labeled degradation may occur without a large marginal distribution shift.

Implementation guidance

AID-D-002.001-G001: Monitor the statistical distribution of model inputs to detect data drift.
AID-D-002.001-G002: Monitor the statistical distribution of model outputs to detect concept drift.
AID-D-002.001-G003: When a serving API exposes top-k token log probabilities, monitor their distribution as an output signal and preserve the unreturned probability mass as unknown.

AID-D-002.001-G001

Monitor the statistical distribution of model inputs to detect data drift.

Reusable mixed-feature drift evaluator

Evaluate numeric, categorical, and missingness drift separately. Keep the full live and reference denominators, apply a declared multiple-testing correction, and obtain all feature lists and thresholds from the signed policy rather than from library defaults.

from __future__ import annotations

# File: detection/input_drift.py

from collections import Counter

import numpy as np
import pandas as pd
from scipy.spatial.distance import jensenshannon
from scipy.stats import ks_2samp


def categorical_distribution(values: pd.Series, categories: list[str]) -> np.ndarray:
    normalized = values.astype("string").fillna("__MISSING__")
    counts = Counter(normalized.tolist())
    unknown = set(counts) - set(categories)
    if unknown:
        raise ValueError(f"category policy missing values: {sorted(unknown)}")
    result = np.asarray([counts.get(category, 0) for category in categories], dtype=float)
    return result / max(1.0, result.sum())


def evaluate_input_drift(
    reference: pd.DataFrame,
    live: pd.DataFrame,
    policy: dict,
) -> dict:
    required = {
        "policy_version", "numeric_features", "categorical_features",
        "categories", "minimum_reference_rows", "minimum_live_rows",
        "numeric_ks_alpha", "categorical_js_threshold",
        "maximum_missing_rate_delta",
    }
    if set(policy) != required:
        raise ValueError("input drift policy schema mismatch")
    if len(reference) < int(policy["minimum_reference_rows"]) or len(live) < int(policy["minimum_live_rows"]):
        return {
            "control_status": "INSUFFICIENT_DATA",
            "finding_status": "UNKNOWN",
            "reference_rows": len(reference),
            "live_rows": len(live),
            "policy_version": policy["policy_version"],
        }
    features = list(policy["numeric_features"]) + list(policy["categorical_features"])
    missing_columns = sorted(set(features) - set(reference.columns) | (set(features) - set(live.columns)))
    if missing_columns:
        return {
            "control_status": "INSUFFICIENT_DATA",
            "finding_status": "UNKNOWN",
            "reason": "required feature columns missing",
            "missing_columns": missing_columns,
            "policy_version": policy["policy_version"],
        }
    results = []
    findings = []
    numeric_count = max(1, len(policy["numeric_features"]))
    corrected_alpha = float(policy["numeric_ks_alpha"]) / numeric_count
    for feature in policy["numeric_features"]:
        reference_values = pd.to_numeric(reference[feature], errors="coerce")
        live_values = pd.to_numeric(live[feature], errors="coerce")
        ref_observed = reference_values.dropna().to_numpy(dtype=float)
        live_observed = live_values.dropna().to_numpy(dtype=float)
        missing_delta = abs(float(live_values.isna().mean() - reference_values.isna().mean()))
        if not len(ref_observed) or not len(live_observed):
            result = {"feature": feature, "kind": "numeric", "state": "INSUFFICIENT_DATA", "missing_rate_delta": missing_delta}
        else:
            statistic, p_value = ks_2samp(ref_observed, live_observed, method="auto")
            drift = p_value < corrected_alpha or missing_delta > float(policy["maximum_missing_rate_delta"])
            result = {
                "feature": feature, "kind": "numeric",
                "reference_observed": len(ref_observed), "live_observed": len(live_observed),
                "ks_statistic": round(float(statistic), 6),
                "p_value": round(float(p_value), 8),
                "corrected_alpha": corrected_alpha,
                "missing_rate_delta": round(missing_delta, 6),
                "finding": drift,
            }
            if drift:
                findings.append(feature)
        results.append(result)
    for feature in policy["categorical_features"]:
        categories = list(policy["categories"][feature])
        ref_dist = categorical_distribution(reference[feature], categories)
        live_dist = categorical_distribution(live[feature], categories)
        distance = float(jensenshannon(ref_dist, live_dist, base=2.0))
        missing_delta = abs(float(live[feature].isna().mean() - reference[feature].isna().mean()))
        drift = distance >= float(policy["categorical_js_threshold"]) or missing_delta > float(policy["maximum_missing_rate_delta"])
        results.append({
            "feature": feature, "kind": "categorical",
            "jensen_shannon_distance": round(distance, 6),
            "missing_rate_delta": round(missing_delta, 6),
            "finding": drift,
        })
        if drift:
            findings.append(feature)
    insufficient = any(item.get("state") == "INSUFFICIENT_DATA" for item in results)
    return {
        "schema_version": "aidefend.input_drift.v1",
        "control_status": "INSUFFICIENT_DATA" if insufficient else "PASS",
        "finding_status": "UNKNOWN" if insufficient else ("FINDING" if findings else "NO_FINDING"),
        "policy_version": policy["policy_version"],
        "reference_rows": len(reference),
        "live_rows": len(live),
        "finding_features": sorted(set(findings)),
        "feature_results": results,
    }

Verification: Replay a signed corpus with stable distributions, shifted numeric values, new categorical values, missingness drift, absent columns, undersized windows, and reordered rows. Recompute the result from immutable Parquet inputs and compare canonical JSON.

Action: Run this evaluator per signed segment and window, surface unmeasured features and rows, and route drift findings for investigation without initiating retraining from this detector.

AID-D-002.001-G002

Monitor the statistical distribution of model outputs to detect concept drift.

Reusable output-distribution evaluator

Measure categorical output drift with a stable label vocabulary that includes abstention and error states. Jensen-Shannon distance avoids invalid chi-square results when an approved output class has a zero reference count; all minimum counts and thresholds still come from the signed policy.

from __future__ import annotations

# File: detection/output_distribution_drift.py

from collections import Counter

import numpy as np
from scipy.spatial.distance import jensenshannon


def normalized_counts(counts: dict[str, int], labels: list[str]) -> np.ndarray:
    if set(counts) - set(labels):
        raise ValueError("output contains labels absent from signed vocabulary")
    values = []
    for label in labels:
        value = counts.get(label, 0)
        if isinstance(value, bool) or not isinstance(value, int) or value < 0:
            raise ValueError("output count must be a non-negative integer")
        values.append(value)
    total = sum(values)
    if total == 0:
        raise ValueError("output population is empty")
    return np.asarray(values, dtype=float) / total


def evaluate_output_drift(
    reference_counts: dict[str, int],
    live_counts: dict[str, int],
    policy: dict,
) -> dict:
    required = {
        "policy_version", "labels", "minimum_reference_count",
        "minimum_live_count", "maximum_js_distance",
    }
    if set(policy) != required:
        raise ValueError("output drift policy schema mismatch")
    labels = list(policy["labels"])
    reference_total = sum(reference_counts.values())
    live_total = sum(live_counts.values())
    if (
        reference_total < int(policy["minimum_reference_count"])
        or live_total < int(policy["minimum_live_count"])
    ):
        return {
            "control_status": "INSUFFICIENT_DATA",
            "finding_status": "UNKNOWN",
            "reference_count": reference_total,
            "live_count": live_total,
            "policy_version": policy["policy_version"],
        }
    reference_distribution = normalized_counts(reference_counts, labels)
    live_distribution = normalized_counts(live_counts, labels)
    distance = float(jensenshannon(reference_distribution, live_distribution, base=2.0))
    finding = distance >= float(policy["maximum_js_distance"])
    return {
        "schema_version": "aidefend.output_distribution_drift.v1",
        "control_status": "PASS",
        "finding_status": "FINDING" if finding else "NO_FINDING",
        "policy_version": policy["policy_version"],
        "labels": labels,
        "reference_counts": [reference_counts.get(label, 0) for label in labels],
        "live_counts": [live_counts.get(label, 0) for label in labels],
        "reference_count": reference_total,
        "live_count": live_total,
        "jensen_shannon_distance": round(distance, 8),
        "threshold": float(policy["maximum_js_distance"]),
    }

Verification: Replay stable, shifted, new-label, zero-reference-bin, abstention-heavy, error-heavy, undersized, and empty populations. Independently rebuild counts from immutable response events rather than trusting a pre-aggregated pass result.

Action: Evaluate every applicable model, route, and segment window and retain exact label counts, method, threshold, and population evidence with the finding.

AID-D-002.001-G003

When a serving API exposes top-k token log probabilities, monitor their distribution as an output signal and preserve the unreturned probability mass as unknown.

Use only documented, returned distributions

Enable log probabilities only on an approved route and bind the observation to provider, endpoint, model identifier, model or deployment revision, tokenizer, API version, sampling settings, requested top_k, and returned token population. Never reconstruct unreturned logits or treat top-k probabilities as the full vocabulary.

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

import math
from collections.abc import Sequence


def topk_features(logprobs: Sequence[float], *, requested_top_k: int) -> dict:
    if type(requested_top_k) is not int or requested_top_k < 2:
        raise ValueError("requested_top_k must be at least two")
    if not isinstance(logprobs, Sequence) or isinstance(logprobs, (str, bytes)):
        raise TypeError("logprobs must be a sequence")
    if len(logprobs) != requested_top_k:
        return {"control_status": "INSUFFICIENT_DATA",
                "finding_status": "UNKNOWN",
                "reason": "returned top-k population is incomplete",
                "requested_top_k": requested_top_k,
                "returned_count": len(logprobs)}
    values = [float(value) for value in logprobs]
    if any(not math.isfinite(value) or value > 0 for value in values):
        raise ValueError("log probabilities must be finite and non-positive")
    probabilities = [math.exp(value) for value in values]
    topk_mass = sum(probabilities)
    if not 0 < topk_mass <= 1.000001:
        raise ValueError("returned probability mass is invalid")
    normalized = [value / topk_mass for value in probabilities]
    entropy = -sum(value * math.log(value) for value in normalized if value > 0)
    ordered = sorted(probabilities, reverse=True)
    return {"control_status": "PASS", "finding_status": "NOT_EVALUATED",
            "requested_top_k": requested_top_k, "returned_count": len(values),
            "topk_mass": round(topk_mass, 8),
            "missing_probability_mass": round(max(0.0, 1.0 - topk_mass), 8),
            "conditional_topk_entropy_nats": round(entropy, 8),
            "top1_top2_probability_margin": round(ordered[0] - ordered[1], 8)}

Aggregate these features over the complete signed window and compare them with a baseline produced under the same binding. A missing field, truncated candidate list, changed tokenizer or sampling configuration, provider semantic change, or mixed model revision is INSUFFICIENT_DATA or ERROR, never a clean result. Independently replay feature extraction from immutable sanitized response envelopes and retain coverage denominators and missing-field counts.

Action: Compute these features only from the exact requested top-k population; record incomplete provider output as INSUFFICIENT_DATA and never treat unreturned mass as zero.

Open-source tools

Evidently
NannyML
SciPy
Prometheus
Grafana

Source-available / open-weight tools

Alibi Detect (Business Source License 1.1; source-available)

Commercial / hosted tools

Arize AX
Fiddler AI Observability
Evidently Cloud
NannyML Cloud
Snowflake ML Observability
Amazon SageMaker Model Monitor
Google Vertex AI Model Monitoring
Azure Machine Learning model monitoring

Threat mappings

MITRE ATLAS

AML.T0031 Erode AI Model Integrity
AML.T0015 Evade AI Model
AML.T0020 Training Data Poisoning
AML.T0043 Craft Adversarial Data (distribution shifts can reveal crafted inputs and poisoned traffic)
AML.T0076 Corrupt AI Model

MAESTRO

Data Tampering (L2) (distribution shifts can reveal tampered or poisoned data)
Data Poisoning (L2) (poisoned traffic changes live feature and output distributions)
Adversarial Examples (L1) (evasion attempts often create anomalous output distributions)

OWASP LLM Top 10 2026

LLM05:2026 Data and Model Poisoning (distribution drift is corroborating evidence of poisoning effects, not root-cause proof)

OWASP ML Top 10 2023

ML08:2023 Model Skewing
ML02:2023 Data Poisoning Attack (drift monitoring reveals poisoning effects)
ML01:2023 Input Manipulation Attack (adversarial inputs can shift live output distributions)

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

NISTAML.013 Data Poisoning
NISTAML.022 Evasion (distribution shifts can reveal evasion effects)
NISTAML.012 Clean-label Poisoning
NISTAML.027 Misaligned Outputs

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

DP: Data Poisoning
MEV: Model Evasion

Databricks AI Security Framework 3.0

Algorithms 5.2: Model drift
Datasets 3.1: Data poisoning
Model Serving - Inference response 10.1: Lack of audit and monitoring inference quality