Scope boundary

Owns provenance, change history, annotator quality, noise detection, and tamper evidence for supervised labels, evaluation labels, preference pairs, and related supervisory signals.

AID-H-002.001: AID-H-002.005 protects labels and preference signals attached to records; AID-H-002.001 sanitizes the record contents used for training or fine-tuning. Label flipping can poison an otherwise normal-looking dataset and therefore requires independent evidence.

AID-H-012.002: AID-H-002.005 protects human or source-provided labels and preference data; AID-H-012.002 protects and calibrates a learned reward signal used by reinforcement-learning workflows. Preference-label integrity is an input to reward learning, while reward-model calibration and integrity govern the learned signal.

Implementation guidance

• AID-H-002.005-G001: Store label assignments and label changes in an append-only event log that records annotator identity, timestamp, guideline version, and previous-event linkage.
• AID-H-002.005-G002: Run automated label-noise detection before training and route suspicious samples to human review instead of silently admitting them.
• AID-H-002.005-G003: Track annotator agreement and gold-sample accuracy over time to detect compromised or unreliable labelers.
• AID-H-002.005-G004: For RLHF or preference data, validate pairwise preference consistency and detect suspicious patterns that systematically push the model toward unsafe outcomes.

AID-H-002.005-G001

Production implementation: transactional label-event hash chain

Use a database transaction and per-dataset lock so concurrent writers cannot fork the chain. The event hash covers canonical event bytes plus the previous 32-byte hash; the first event uses 64 zeroes.

# File: labels/append_event.py
from __future__ import annotations
import hashlib
import json
import re
from datetime import datetime, timezone

SHA256 = re.compile(r"[a-f0-9]{64}")

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

def append_label_event(connection, event: dict) -> dict:
    required = {"dataset_id", "sample_id", "label", "annotator_id", "guideline_version", "source_snapshot_sha256"}
    if set(event) != required or not all(event.values()):
        raise ValueError("label event schema differs")
    with connection.transaction():
        connection.execute("SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))", (event["dataset_id"],))
        prior = connection.execute(
            "SELECT event_hash FROM label_event_log WHERE dataset_id=%s ORDER BY sequence_id DESC LIMIT 1 FOR UPDATE",
            (event["dataset_id"],),
        ).fetchone()
        previous_hash = prior[0] if prior else "0" * 64
        if not SHA256.fullmatch(previous_hash):
            raise RuntimeError("stored chain head is invalid")
        body = dict(event)
        body["recorded_at"] = datetime.now(timezone.utc).isoformat()
        body["previous_event_hash"] = previous_hash
        event_hash = hashlib.sha256(bytes.fromhex(previous_hash) + canonical(body)).hexdigest()
        row = connection.execute(
            "INSERT INTO label_event_log(dataset_id,event_body,previous_event_hash,event_hash) VALUES (%s,%s::jsonb,%s,%s) RETURNING sequence_id",
            (event["dataset_id"], canonical(body).decode(), previous_hash, event_hash),
        ).fetchone()
    return {"sequence_id": row[0], "event_hash": event_hash, **body}

def verify_chain(events: list[dict]) -> None:
    previous = "0" * 64
    for expected_sequence, row in enumerate(events, start=1):
        if row["sequence_id"] != expected_sequence or row["previous_event_hash"] != previous:
            raise PermissionError("label event sequence or linkage differs")
        body = row["event_body"]
        actual = hashlib.sha256(bytes.fromhex(previous) + canonical(body)).hexdigest()
        if actual != row["event_hash"]:
            raise PermissionError("label event hash differs")
        previous = actual

Storage: Deny UPDATE/DELETE to the writer role, stream committed rows to an object-lock/WORM evidence bucket, and periodically sign a checkpoint containing dataset ID, final sequence, final hash, row count, and snapshot digest. Admission independently verifies the entire chain and checkpoint before materializing labels.

Before you begin

Apply AID-H-002.005-G001 only when human, automated, or preference labels influence training, fine-tuning, ranking, evaluation, or reward-model data.

AID-H-002.005-G002

Delivery level: reusable label-review gate

This implementation is for numeric tabular classification. It produces out-of-fold probabilities for every row, validates that complete probability population, runs Cleanlab, and writes stable row IDs to a review queue. Replace the estimator for another modality, but keep the same out-of-sample and evidence contract.

Step 1: Use a signed review profile

{
  "schema_version": "aidefend.label-review-profile.v1",
  "policy_version": "fraud-labels-2026-07",
  "row_id_column": "sample_id",
  "label_column": "label",
  "feature_columns": ["amount", "account_age_days", "failed_login_count"],
  "cross_validation_folds": 5,
  "random_seed": 20260726,
  "estimator_max_iterations": 2000,
  "maximum_issue_fraction_before_snapshot_block": 0.02
}

The fold count, seed, feature set, estimator iteration cap, and blocking fraction belong to the versioned organizational profile. Sign it with the approved policy identity; do not let a training job supply its own threshold.

Step 2: Generate out-of-fold predictions and the review queue

# File: label_integrity/find_label_issues.py
from __future__ import annotations

import argparse
import hashlib
import io
import json
import math
import os
import subprocess
import tempfile
from pathlib import Path
from typing import Any

import numpy as np
import pandas as pd
from cleanlab.filter import find_label_issues
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import StratifiedKFold, cross_val_predict
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler


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


def load_profile(
    path: Path, bundle: Path, key: Path, timeout_seconds: float
) -> tuple[dict[str, Any], str]:
    if not math.isfinite(timeout_seconds) or timeout_seconds <= 0:
        raise ValueError("verification timeout must be finite and positive")
    payload = path.read_bytes()
    with tempfile.NamedTemporaryFile(prefix="label-review-profile-", delete=False) as handle:
        handle.write(payload)
        handle.flush()
        os.fsync(handle.fileno())
        snapshot = Path(handle.name)
    try:
        subprocess.run(
            ["cosign", "verify-blob", "--key", str(key), "--bundle", str(bundle), str(snapshot)],
            check=True, capture_output=True, text=True, timeout=timeout_seconds,
        )
    finally:
        snapshot.unlink(missing_ok=True)
    profile = json.loads(
        payload.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_object,
        parse_constant=lambda value: (_ for _ in ()).throw(ValueError(f"non-finite JSON number: {value}")),
    )
    expected = {
        "schema_version", "policy_version", "row_id_column", "label_column",
        "feature_columns", "cross_validation_folds", "random_seed",
        "estimator_max_iterations",
        "maximum_issue_fraction_before_snapshot_block",
    }
    if not isinstance(profile, dict) or set(profile) != expected:
        raise ValueError("label-review profile schema mismatch")
    if profile["schema_version"] != "aidefend.label-review-profile.v1":
        raise ValueError("unsupported label-review profile")
    if not isinstance(profile["policy_version"], str) or not profile["policy_version"]:
        raise ValueError("policy_version is required")
    for key_name in ("row_id_column", "label_column"):
        if not isinstance(profile[key_name], str) or not profile[key_name]:
            raise ValueError(f"{key_name} is invalid")
    features = profile["feature_columns"]
    if not isinstance(features, list) or not features or len(features) != len(set(features)):
        raise ValueError("feature_columns must be unique and non-empty")
    folds = profile["cross_validation_folds"]
    if type(folds) is not int or folds < 2:
        raise ValueError("cross_validation_folds must be at least two")
    if type(profile["random_seed"]) is not int:
        raise ValueError("random_seed must be an integer")
    iterations = profile["estimator_max_iterations"]
    if type(iterations) is not int or iterations < 1:
        raise ValueError("estimator_max_iterations must be a positive integer")
    limit = profile["maximum_issue_fraction_before_snapshot_block"]
    if isinstance(limit, bool) or not isinstance(limit, (int, float)) or not math.isfinite(float(limit)) or not 0 <= float(limit) <= 1:
        raise ValueError("maximum issue fraction is invalid")
    return profile, hashlib.sha256(payload).hexdigest()


def audit_labels(input_path: Path, profile: dict[str, Any]) -> tuple[dict[str, Any], list[dict[str, Any]]]:
    input_bytes = input_path.read_bytes()
    frame = pd.read_csv(io.BytesIO(input_bytes))
    required = [profile["row_id_column"], profile["label_column"], *profile["feature_columns"]]
    if any(column not in frame.columns for column in required):
        raise ValueError("input is missing a required signed-profile column")
    row_ids = frame[profile["row_id_column"]]
    if row_ids.isna().any() or row_ids.astype(str).str.len().eq(0).any() or row_ids.astype(str).duplicated().any():
        raise ValueError("sample IDs must be unique and non-empty")
    labels = frame[profile["label_column"]]
    if labels.isna().any():
        raise ValueError("labels cannot be null")
    features = frame[profile["feature_columns"]].apply(pd.to_numeric, errors="raise")
    feature_values = features.to_numpy(dtype=np.float64)
    if np.isinf(feature_values).any():
        raise ValueError("features contain infinite values")
    folds = profile["cross_validation_folds"]
    class_counts = labels.value_counts(dropna=False)
    if len(class_counts) < 2 or int(class_counts.min()) < folds:
        raise ValueError("every class must have at least one sample per cross-validation fold")

    splitter = StratifiedKFold(
        n_splits=folds, shuffle=True, random_state=profile["random_seed"]
    )
    estimator = make_pipeline(
        SimpleImputer(strategy="median"),
        StandardScaler(),
        LogisticRegression(
            max_iter=profile["estimator_max_iterations"],
            random_state=profile["random_seed"],
        ),
    )
    probabilities = cross_val_predict(
        estimator, features, labels, cv=splitter, method="predict_proba", n_jobs=1
    )
    if (
        probabilities.ndim != 2
        or probabilities.shape[0] != len(frame)
        or not np.isfinite(probabilities).all()
        or (probabilities < 0).any()
        or (probabilities > 1).any()
        or not np.allclose(probabilities.sum(axis=1), 1.0, rtol=0.0, atol=1e-6)
    ):
        raise RuntimeError("out-of-fold probability population is incomplete or invalid")

    issue_indices = np.asarray(find_label_issues(
        labels=labels.to_numpy(),
        pred_probs=probabilities,
        return_indices_ranked_by="self_confidence",
    ), dtype=np.int64)
    if issue_indices.ndim != 1 or len(set(issue_indices.tolist())) != len(issue_indices):
        raise RuntimeError("Cleanlab returned invalid issue indices")
    if len(issue_indices) and (int(issue_indices.min()) < 0 or int(issue_indices.max()) >= len(frame)):
        raise RuntimeError("Cleanlab returned an out-of-population index")

    classes = np.unique(labels.to_numpy())
    if probabilities.shape[1] != len(classes):
        raise RuntimeError("probability columns do not match the observed classes")
    queue = [
        {
            "sample_id": str(row_ids.iloc[index]),
            "given_label": str(labels.iloc[index]),
            "suggested_label": str(classes[int(np.argmax(probabilities[index]))]),
            "rank": rank + 1,
        }
        for rank, index in enumerate(issue_indices.tolist())
    ]
    issue_fraction = len(queue) / len(frame) if len(frame) else 1.0
    if not len(frame):
        raise ValueError("label population is empty")
    outcome = (
        "BLOCK"
        if issue_fraction > float(profile["maximum_issue_fraction_before_snapshot_block"])
        else ("REVIEW_REQUIRED" if queue else "PASS")
    )
    report = {
        "schema_version": "aidefend.label-review-result.v1",
        "policy_version": profile["policy_version"],
        "input_sha256": hashlib.sha256(input_bytes).hexdigest(),
        "population_count": len(frame),
        "evaluated_count": probabilities.shape[0],
        "class_count": probabilities.shape[1],
        "issue_count": len(queue),
        "issue_fraction": issue_fraction,
        "outcome": outcome,
    }
    return report, queue


def write_json(path: Path, value: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    os.replace(temporary, path)


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", type=Path, required=True)
    parser.add_argument("--profile", type=Path, required=True)
    parser.add_argument("--profile-bundle", type=Path, required=True)
    parser.add_argument("--profile-key", type=Path, required=True)
    parser.add_argument("--report", type=Path, required=True)
    parser.add_argument("--review-queue", type=Path, required=True)
    parser.add_argument("--verify-timeout-seconds", type=float, required=True)
    args = parser.parse_args()
    profile, profile_sha256 = load_profile(
        args.profile, args.profile_bundle, args.profile_key, args.verify_timeout_seconds
    )
    report, queue = audit_labels(args.input, profile)
    report["profile_sha256"] = profile_sha256
    write_json(args.report, report)
    write_json(args.review_queue, queue)
    return 0 if report["outcome"] == "PASS" else 1


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

Step 3: Keep disputed rows out of training

A nonzero exit code holds the candidate snapshot. Reviewers decide each queued stable sample ID and produce a new versioned label snapshot; the training job must not silently drop or relabel flagged rows. Re-run the module on the adjudicated snapshot and promote only when its exact digest has a PASS report.

Action: Run the cross-validated label review against the complete governed label population; route REVIEW_REQUIRED items to the review queue and block release on invalid inputs, incomplete folds, storage failure, or policy BLOCK.

AID-H-002.005-G003

Concept:

Monitor hidden gold-sample accuracy and independently assigned overlap items for every active annotator. Averages across the whole workforce can hide one compromised or confused annotator, so preserve per-annotator and measured-pair populations and treat missing coverage separately from acceptable quality.

Compute a complete review result

# File: label_quality/evaluate_annotators.py
from __future__ import annotations

from itertools import combinations
import hashlib
import json
import math
from pathlib import Path

import pandas as pd


LABELS = Path("artifacts/label_events.csv")
POLICY = Path("policy/label_quality_policy.json")
REPORT = Path("artifacts/annotator_quality_report.json")
REQUIRED_COLUMNS = {
    "item_id", "annotator_id", "label", "is_gold", "gold_label",
    "guideline_version", "created_at",
}


def finite_rate(value: object, field: str) -> float:
    if isinstance(value, bool):
        raise ValueError(f"{field} must be numeric")
    number = float(value)
    if not math.isfinite(number) or not 0.0 <= number <= 1.0:
        raise ValueError(f"{field} must be in [0, 1]")
    return number


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


policy = json.loads(POLICY.read_text(encoding="utf-8"))
required_policy = {
    "schema_version", "policy_version",
    "minimum_gold_samples_per_annotator", "minimum_gold_accuracy",
    "minimum_overlap_samples_per_pair", "minimum_pair_agreement",
}
if set(policy) != required_policy or policy["schema_version"] != "aidefend.label-quality-policy.v1":
    raise ValueError("label-quality policy schema differs")
minimum_gold = positive_integer(
    policy["minimum_gold_samples_per_annotator"],
    "minimum_gold_samples_per_annotator",
)
minimum_overlap = positive_integer(
    policy["minimum_overlap_samples_per_pair"],
    "minimum_overlap_samples_per_pair",
)
minimum_gold_accuracy = finite_rate(
    policy["minimum_gold_accuracy"], "minimum_gold_accuracy"
)
minimum_pair_agreement = finite_rate(
    policy["minimum_pair_agreement"], "minimum_pair_agreement"
)

labels = pd.read_csv(LABELS, dtype="string", keep_default_na=False)
if labels.empty or not REQUIRED_COLUMNS.issubset(labels.columns):
    raise RuntimeError("label population or required columns are missing")
if labels[list(REQUIRED_COLUMNS - {"gold_label"})].eq("").any().any():
    raise RuntimeError("required label fields contain empty values")
if labels.duplicated(["item_id", "annotator_id"]).any():
    raise RuntimeError("one annotator labeled the same item more than once")
normalized_gold = labels["is_gold"].str.lower()
if not normalized_gold.isin({"true", "false"}).all():
    raise ValueError("is_gold must contain only true or false")
labels["is_gold"] = normalized_gold.eq("true")
if labels.loc[labels["is_gold"], "gold_label"].eq("").any():
    raise RuntimeError("gold rows are missing gold_label")

annotators = sorted(labels["annotator_id"].unique().tolist())
failures: list[dict] = []
insufficient: list[dict] = []
gold_results: list[dict] = []
for annotator_id in annotators:
    gold = labels[(labels["annotator_id"] == annotator_id) & labels["is_gold"]]
    if len(gold) < minimum_gold:
        insufficient.append({
            "annotator_id": annotator_id,
            "reason": "insufficient_gold_samples",
            "observed": len(gold),
            "required": minimum_gold,
        })
        continue
    accuracy = float((gold["label"] == gold["gold_label"]).mean())
    result = {
        "annotator_id": annotator_id,
        "gold_samples": len(gold),
        "gold_accuracy": accuracy,
    }
    gold_results.append(result)
    if accuracy < minimum_gold_accuracy:
        failures.append({**result, "reason": "gold_accuracy_below_policy"})

pair_results: list[dict] = []
covered_by_sufficient_overlap: set[str] = set()
for left, right in combinations(annotators, 2):
    left_rows = labels.loc[
        labels["annotator_id"] == left, ["item_id", "label"]
    ].rename(columns={"label": "left_label"})
    right_rows = labels.loc[
        labels["annotator_id"] == right, ["item_id", "label"]
    ].rename(columns={"label": "right_label"})
    overlap = left_rows.merge(right_rows, on="item_id", how="inner", validate="one_to_one")
    if not overlap.empty and len(overlap) < minimum_overlap:
        insufficient.append({
            "annotator_pair": [left, right],
            "reason": "insufficient_overlap_samples",
            "observed": len(overlap),
            "required": minimum_overlap,
        })
        continue
    if overlap.empty:
        continue
    covered_by_sufficient_overlap.update({left, right})
    agreement = float((overlap["left_label"] == overlap["right_label"]).mean())
    result = {
        "annotator_pair": [left, right],
        "overlap_samples": len(overlap),
        "exact_agreement": agreement,
    }
    pair_results.append(result)
    if agreement < minimum_pair_agreement:
        failures.append({**result, "reason": "pair_agreement_below_policy"})

for annotator_id in sorted(set(annotators) - covered_by_sufficient_overlap):
    insufficient.append({
        "annotator_id": annotator_id,
        "reason": "no_sufficient_independent_overlap",
    })

status = "FAIL" if failures else ("INSUFFICIENT_DATA" if insufficient else "PASS")
report = {
    "schema_version": "aidefend.annotator-quality-report.v1",
    "status": status,
    "policy_version": policy["policy_version"],
    "label_population": len(labels),
    "annotator_population": len(annotators),
    "gold_results": gold_results,
    "pair_results": pair_results,
    "failures": failures,
    "insufficient_populations": insufficient,
    "labels_sha256": hashlib.sha256(LABELS.read_bytes()).hexdigest(),
    "policy_sha256": hashlib.sha256(POLICY.read_bytes()).hexdigest(),
}
REPORT.parent.mkdir(parents=True, exist_ok=True)
REPORT.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
print(json.dumps({"status": status, "report": str(REPORT)}))
raise SystemExit(0 if status == "PASS" else 2)

Apply and verify the result

Verify the versioned policy before executing this job, derive the active annotator population from the assignment system, and reconcile it with annotator_population. Route labels from a failing annotator to secondary review and pause new assignments according to policy; do not delete or overwrite the original append-only label events. Re-run after a guideline, gold set, annotator assignment, or threshold change, and independently recompute the report from the exact label-event bytes.

Exercise accurate and inaccurate gold labels, too few gold samples, adequate and inadequate overlap, disagreement below policy, duplicate item/annotator rows, malformed booleans, missing columns, an unmeasured active annotator, and policy/read failures. Missing gold or overlap coverage remains INSUFFICIENT_DATA; measured quality below policy is FAIL; parse or dependency failure is ERROR at the orchestration boundary and must block snapshot admission.

Before you begin

Apply AID-H-002.005-G003 only when human, automated, or preference labels influence training, fine-tuning, ranking, evaluation, or reward-model data.

AID-H-002.005-G004

Concept:

Preference data is still supervisory data. If preference pairs are manipulated, the model can be aligned toward unsafe or biased behavior.

Example checks

1. Check transitivity where comparable triplets exist.
2. Check whether a single annotator disproportionately prefers outputs with a risky trait.
3. Measure agreement on shared preference pairs.
4. Track changes in preference distribution after policy or guideline updates.

Operational notes: Treat preference data as high-sensitivity training input. Keep the same audit, versioning, and review discipline as you would for ordinary supervised labels.

Before you begin

Apply AID-H-002.005-G004 only when human, automated, or preference labels influence training, fine-tuning, ranking, evaluation, or reward-model data.

Keep the action safely bounded

exclude or quarantine disputed labels and block snapshot admission when the event chain, population, or adjudication evidence is incomplete.

Open-source tools

• Label Studio
• cleanlab
• Argilla
• Snorkel
• DVC
• scikit-learn
• statsmodels

Commercial / hosted tools

• Scale AI (managed labeling with QA controls)
• Labelbox (annotation platform with review workflows)
• Amazon SageMaker Ground Truth
• Appen (managed annotation operations)
• Surge AI
• Prodigy

Threat mappings

MITRE ATLAS

• AML.T0115 Publish Poisoned AI Artifacts
• AML.T0115.000 Publish Poisoned AI Artifacts: Datasets (datasets with manipulated labels)
• AML.T0020 Training Data Poisoning (label flipping as poisoning vector)
• AML.T0059 Erode Dataset Integrity (label manipulation erodes supervisory integrity)

MAESTRO

• Data Poisoning (L2) (label-level poisoning)
• Data Poisoning (Training Phase) (L1) (label manipulation during dataset preparation)
• Data Tampering (L2) (label tampering)

OWASP LLM Top 10 2026

• LLM05:2026 Data and Model Poisoning

OWASP ML Top 10 2023

• ML02:2023 Data Poisoning Attack (label-specific poisoning)
• ML08:2023 Model Skewing (biased or corrupted labels skew model behavior)

OWASP Top 10 for Agentic Applications 2026

• ASI04:2026 Agentic Supply Chain Vulnerabilities (compromised labeling pipeline in agent training supply chain)

NIST Adversarial Machine Learning 2025

• NISTAML.013 Data Poisoning (label flipping and noisy labeling)
• NISTAML.024 Targeted Poisoning (targeted label manipulation)

Cisco Integrated AI Security and Safety Framework

• AISubtech-2.3.2 Verifier Manipulation
• AISubtech-6.1.2 Reinforcement Biasing
• AISubtech-6.1.3 Reinforcement Signal Corruption
• AITech-2.3 Reward Hacking
• AITech-6.1 Training Data Poisoning

Google Secure AI Framework 2.0 - Risks

• DP: Data Poisoning (label manipulation as poisoning vector)

Databricks AI Security Framework 3.0

• Datasets 3.1: Data poisoning (label-level poisoning)
• Datasets 3.3: Label flipping
• Evaluation 6.1: Evaluation data poisoning (label poisoning affects evaluation integrity)
• Raw Data 1.6: Insufficient data lineage (label provenance tracking requires end-to-end lineage)