AID-M-007.003: Fairness & Disparity Release Evaluation

Version 1.20261015 · Actionable sub-technique

Use only when the signed AID-M-007.001 boundary classifies protected_group_fairness as applicable because the system scores, ranks, classifies, recommends, allocates, or denies outcomes for persons or groups. Evaluate the exact release candidate over the complete policy-selected population, compute the specified disparity and coverage checks, independently replay row-level predictions and metrics, and block promotion on threshold breach, incomplete population, stale release binding, or replay mismatch.

Pillars: data, model

Lifecycle: validation

Scope boundary

Owns pre-release statistical fairness and disparity admission when the signed boundary makes protected-group fairness applicable, including policy-selected population coverage, task-appropriate metrics, row-level replay, threshold decisions, and release binding.

AID-M-007.002: AID-M-007.003 evaluates population-level fairness and disparity; AID-M-007.002 executes deterministic safety, content, disclosure, and action cases. A deterministic case suite cannot establish statistical parity or disparity across a governed population.

AID-M-003.001: AID-M-007.003 computes and verifies the fairness release result; AID-M-003.001 documents the result and links its evidence in release documentation. A fairness statement in a model card is not a substitute for task-appropriate metrics, complete population coverage, and replay.

AID-D-003.007: AID-M-007.003 evaluates pre-release protected-group outcome disparity; AID-D-003.007 detects runtime attempts or behavior that infer sensitive attributes. Fairness of release outcomes and sensitive-attribute inference during operation are different risks, populations, and evidence paths.

Implementation guidance

AID-M-007.003-G001

Evaluate fairness over a signed, complete release population using policy-owned metrics and thresholds.
Applicability and statistical boundary

Run this method only when the signed safety boundary includes protected_group_fairness in applicable_dimensions. The architecture must contain a person- or group-affecting binary-classification path. A missing protected column, unavailable evaluator, incomplete population, or unsupported task type is an implementation or data failure, not a reason to mark this control not applicable. Use a separate task-specific implementation path for ranking, recommendation, multiclass, regression, allocation, or generative outcomes.

Bind the fairness plan, source snapshot, and exact selected population

The numeric values below illustrate a policy-rendered binary-classification fixture; they are not AIDEFEND defaults. The signed plan owns the immutable dataset snapshot, all-rows selection, exact row-ID population, metrics, groups, coverage, freshness, and thresholds.

{
  "schema_version": "aidefend.fairness_plan.v2",
  "plan_id": "credit-decision-fairness",
  "plan_version": "8",
  "boundary_policy_id": "credit-decision-boundary",
  "boundary_policy_version": "5.0.0",
  "boundary_sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
  "dimension": "protected_group_fairness",
  "task_type": "binary_classification",
  "protected_column": "group",
  "label_column": "y_true",
  "prediction_column": "y_pred",
  "source_dataset_id": "credit-validation-2026q2",
  "source_dataset_path": "fairness/credit-validation-2026q2.csv",
  "source_dataset_sha256": "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
  "population_selection": {"type": "all_rows", "row_id_column": "row_id"},
  "expected_row_count": 400,
  "expected_row_ids_sha256": "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd",
  "required_groups": ["group_a", "group_b"],
  "minimum_rows_per_group": 100,
  "minimum_positive_labels_per_group": 20,
  "minimum_negative_labels_per_group": 20,
  "metrics": {
    "demographic_parity_difference": {"maximum": 0.05},
    "equalized_odds_difference": {"maximum": 0.08}
  },
  "max_result_age_seconds": 86400
}
{
  "schema_version": "aidefend.fairness_population.v2",
  "run_id": "fairness-20260719-001",
  "generator_identity": "spiffe://company.example/evaluation/fairness-generator",
  "scorer_code_sha256": "6666666666666666666666666666666666666666666666666666666666666666",
  "scorer_config_sha256": "7777777777777777777777777777777777777777777777777777777777777777",
  "execution_receipt_path": "fairness/receipts/primary-execution.json",
  "execution_receipt_sha256": "8888888888888888888888888888888888888888888888888888888888888888",
  "execution_receipt_signature_path": "fairness/receipts/primary-execution.sig",
  "release_readback_sha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
  "candidate_bundle_sha256": "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee",
  "selection_sha256": "af23e856e820119c10b2aabb9c0081ed532d98bc657234a6daa273b5bcf6ab2f",
  "source_dataset_id": "credit-validation-2026q2",
  "source_dataset_path": "fairness/credit-validation-2026q2.csv",
  "source_dataset_sha256": "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
  "population_path": "fairness/fairness-population.csv",
  "population_sha256": "1111111111111111111111111111111111111111111111111111111111111111",
  "row_count": 400,
  "generated_at": "2026-07-19T18:00:00Z"
}

An independently credentialed replay scorer publishes the same manifest schema as fairness-population-manifest.json to fairness-replay-manifest.json, with a distinct run_id, authorized generator_identity, invocation receipt, and prediction population. Each signed manifest binds the exact scorer code digest, scorer configuration digest, candidate, source snapshot, selected row population, execution receipt, and row-level output. Both scorers must bind the same plan-selected source snapshot, selection digest, candidate bundle, and release readback.

cosign sign-blob --yes --key env://FAIRNESS_POLICY_SIGNING_KEY --bundle fairness/fairness-plan.sig fairness/fairness-plan.json
cosign sign-blob --yes --key env://FAIRNESS_GENERATOR_SIGNING_KEY --bundle fairness/fairness-population-manifest.sig fairness/fairness-population-manifest.json
cosign sign-blob --yes --key env://FAIRNESS_REPLAY_SIGNING_KEY --bundle fairness/fairness-replay-manifest.sig fairness/fairness-replay-manifest.json
Verify the complete population, replay, coverage, and disparity metrics
# File: fairness/evaluate_fairness.py
from __future__ import annotations

import csv
import hashlib
import io
import json
import math
import os
import stat
import subprocess
import sys
import tempfile
from collections import defaultdict
from datetime import datetime, timedelta, timezone
from pathlib import Path

BOUNDARY = Path("safety/safety-boundary.json")
PLAN = Path("fairness/fairness-plan.json")
MANIFEST = Path("fairness/fairness-population-manifest.json")
REPLAY_MANIFEST = Path("fairness/fairness-replay-manifest.json")
READBACK = Path("evidence/deployed-release-readback.json")
OUTPUT = Path("evidence/fairness-evaluation.json")
FAIRNESS_ROOT = Path("fairness").resolve()
SUPPORTED_METRICS = {
    "demographic_parity_difference",
    "equalized_odds_difference",
}

try:
    COMMAND_TIMEOUT_SECONDS = float(
        os.environ["M007_FAIRNESS_COMMAND_TIMEOUT_SECONDS"]
    )
except (KeyError, ValueError) as error:
    raise RuntimeError("required fairness command timeout is missing") from error
if (
    not math.isfinite(COMMAND_TIMEOUT_SECONDS)
    or COMMAND_TIMEOUT_SECONDS <= 0
):
    raise RuntimeError("fairness command timeout must be positive and finite")


class InsufficientData(Exception):
    pass


_CAPTURED_BYTES: dict[Path, bytes] = {}


def captured_bytes(path: Path) -> bytes:
    if path not in _CAPTURED_BYTES:
        descriptor = os.open(
            path,
            os.O_RDONLY
            | getattr(os, "O_BINARY", 0)
            | getattr(os, "O_NOFOLLOW", 0),
        )
        with os.fdopen(descriptor, "rb") as handle:
            status = os.fstat(handle.fileno())
            if not stat.S_ISREG(status.st_mode):
                raise ValueError(f"input is not a regular file: {path}")
            payload = handle.read()
        if not payload:
            raise ValueError(f"input is empty: {path}")
        _CAPTURED_BYTES[path] = payload
    return _CAPTURED_BYTES[path]


def digest(path: Path) -> str:
    return hashlib.sha256(captured_bytes(path)).hexdigest()


def safe_digest(path: Path) -> str | None:
    try:
        return digest(path)
    except (OSError, ValueError):
        return None


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


def reject_nonfinite(value):
    raise ValueError(f"non-finite JSON number: {value}")


def load_json(path: Path):
    return json.loads(
        captured_bytes(path).decode("utf-8", errors="strict"),
        object_pairs_hook=unique_object,
        parse_constant=reject_nonfinite,
    )


def canonical_sha256(value: object) -> str:
    encoded = json.dumps(
        value, sort_keys=True, separators=(",", ":"), allow_nan=False
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def row_ids_sha256(row_ids: list[str]) -> str:
    return canonical_sha256(sorted(row_ids))


def fairness_path(value: object) -> Path:
    if not isinstance(value, str) or not value or Path(value).is_absolute():
        raise ValueError("fairness path must be a non-empty relative path")
    resolved = Path(value).resolve()
    if not resolved.is_relative_to(FAIRNESS_ROOT):
        raise ValueError("fairness path escapes the bounded fairness root")
    return resolved


def verify(path: Path, signature: Path, key: Path) -> None:
    payload = captured_bytes(path)
    signature_bytes = captured_bytes(signature)
    with tempfile.TemporaryDirectory(prefix="aidefend-fairness-") as stage_name:
        stage = Path(stage_name)
        os.chmod(stage, 0o700)
        payload_snapshot = stage / "payload"
        signature_snapshot = stage / "payload.sig"
        payload_snapshot.write_bytes(payload)
        signature_snapshot.write_bytes(signature_bytes)
        os.chmod(payload_snapshot, 0o400)
        os.chmod(signature_snapshot, 0o400)
        subprocess.run(
            [
                "cosign", "verify-blob", "--key", str(key),
                "--bundle", str(signature_snapshot), str(payload_snapshot),
            ],
            check=True,
            capture_output=True,
            text=True,
            timeout=COMMAND_TIMEOUT_SECONDS,
        )
        if (
            payload_snapshot.read_bytes() != payload
            or signature_snapshot.read_bytes() != signature_bytes
        ):
            raise ValueError("signed fairness snapshot changed during verification")


def csv_rows(path: Path) -> tuple[list[str], list[dict]]:
    stream = io.StringIO(
        captured_bytes(path).decode("utf-8", errors="strict"), newline=""
    )
    reader = csv.DictReader(stream)
    fields = reader.fieldnames or []
    if (
        not fields
        or len(fields) != len(set(fields))
        or any(not isinstance(field, str) or not field for field in fields)
    ):
        raise ValueError(f"CSV headers are empty or duplicated: {path}")
    return fields, list(reader)


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


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


now = datetime.now(timezone.utc)
outcome = "ERROR"
reasons: list[str] = []
metric_values: dict[str, float] = {}
group_counts: dict[str, dict[str, int]] = {}
expires_at = None
plan: dict = {}
manifest: dict = {}
replay_manifest: dict = {}
try:
    verify(
        BOUNDARY,
        BOUNDARY.with_suffix(".sig"),
        Path("keys/safety-policy.pub"),
    )
    verify(
        PLAN,
        PLAN.with_suffix(".sig"),
        Path("keys/fairness-policy.pub"),
    )
    verify(
        MANIFEST,
        MANIFEST.with_suffix(".sig"),
        Path("keys/fairness-generator.pub"),
    )
    verify(
        REPLAY_MANIFEST,
        REPLAY_MANIFEST.with_suffix(".sig"),
        Path("keys/fairness-replay.pub"),
    )
    verify(
        READBACK,
        READBACK.with_suffix(".sig"),
        Path("keys/deployment-verifier.pub"),
    )
    subprocess.run(
        [sys.executable, "safety/validate_boundary.py"],
        check=True,
        capture_output=True,
        text=True,
        timeout=COMMAND_TIMEOUT_SECONDS,
    )

    boundary = load_json(BOUNDARY)
    plan = load_json(PLAN)
    manifest = load_json(MANIFEST)
    replay_manifest = load_json(REPLAY_MANIFEST)
    readback = load_json(READBACK)
    if "protected_group_fairness" not in set(
        boundary["applicable_dimensions"]
    ):
        raise ValueError(
            "fairness control invoked for a boundary where fairness does not apply"
        )
    plan_fields = {
        "schema_version", "plan_id", "plan_version",
        "boundary_policy_id", "boundary_policy_version",
        "boundary_sha256", "dimension", "task_type",
        "protected_column", "label_column", "prediction_column",
        "source_dataset_id", "source_dataset_path", "source_dataset_sha256",
        "population_selection", "expected_row_count",
        "expected_row_ids_sha256",
        "required_groups", "minimum_rows_per_group",
        "minimum_positive_labels_per_group",
        "minimum_negative_labels_per_group", "metrics",
        "max_result_age_seconds",
    }
    if (
        set(plan) != plan_fields
        or plan["schema_version"] != "aidefend.fairness_plan.v2"
        or plan["dimension"] != "protected_group_fairness"
        or plan["task_type"] != "binary_classification"
        or plan["boundary_sha256"] != digest(BOUNDARY)
        or plan["boundary_policy_id"] != boundary["policy_id"]
        or plan["boundary_policy_version"] != boundary["policy_version"]
        or not isinstance(plan["source_dataset_id"], str)
        or not plan["source_dataset_id"]
        or not isinstance(plan["source_dataset_path"], str)
        or not plan["source_dataset_path"]
        or not isinstance(plan["source_dataset_sha256"], str)
        or len(plan["source_dataset_sha256"]) != 64
        or set(plan["source_dataset_sha256"]) - set("0123456789abcdef")
        or plan["population_selection"]
            != {"type": "all_rows", "row_id_column": "row_id"}
        or not isinstance(plan["expected_row_ids_sha256"], str)
        or len(plan["expected_row_ids_sha256"]) != 64
        or set(plan["expected_row_ids_sha256"]) - set("0123456789abcdef")
    ):
        raise ValueError("fairness plan schema or boundary binding differs")
    expected_row_count = positive_integer(
        plan["expected_row_count"], "expected_row_count"
    )
    required_groups = plan["required_groups"]
    if (
        not isinstance(required_groups, list)
        or len(required_groups) < 2
        or len(required_groups) != len(set(required_groups))
        or any(not isinstance(group, str) or not group for group in required_groups)
    ):
        raise ValueError("required fairness groups are invalid")
    minimum_rows = positive_integer(
        plan["minimum_rows_per_group"],
        "minimum_rows_per_group",
    )
    minimum_positive = positive_integer(
        plan["minimum_positive_labels_per_group"],
        "minimum_positive_labels_per_group",
    )
    minimum_negative = positive_integer(
        plan["minimum_negative_labels_per_group"],
        "minimum_negative_labels_per_group",
    )
    max_result_age = positive_integer(
        plan["max_result_age_seconds"],
        "max_result_age_seconds",
    )
    metric_policy = plan["metrics"]
    if (
        not isinstance(metric_policy, dict)
        or not metric_policy
        or set(metric_policy) - SUPPORTED_METRICS
    ):
        raise ValueError("fairness metric policy is empty or unsupported")
    thresholds: dict[str, float] = {}
    for name, rule in metric_policy.items():
        if set(rule) != {"maximum"}:
            raise ValueError(f"metric policy schema differs: {name}")
        maximum = float(rule["maximum"])
        if not math.isfinite(maximum) or not 0.0 <= maximum <= 1.0:
            raise ValueError(f"metric threshold is invalid: {name}")
        thresholds[name] = maximum

    if (
        readback.get("schema_version")
        != "aidefend.deployed_release_readback.v2"
        or readback.get("outcome") != "PASS"
        or now >= when(readback["expires_at"])
    ):
        raise InsufficientData("deployed release readback is not a fresh PASS")
    manifest_fields = {
        "schema_version", "run_id", "generator_identity",
        "scorer_code_sha256", "scorer_config_sha256",
        "execution_receipt_path", "execution_receipt_sha256",
        "execution_receipt_signature_path", "release_readback_sha256", "candidate_bundle_sha256",
        "selection_sha256", "source_dataset_id",
        "source_dataset_path", "source_dataset_sha256",
        "population_path", "population_sha256", "row_count",
        "generated_at",
    }
    readback_sha256 = digest(READBACK)
    selection_sha256 = canonical_sha256(plan["population_selection"])
    source_dataset = fairness_path(plan["source_dataset_path"])
    if digest(source_dataset) != plan["source_dataset_sha256"]:
        raise ValueError("policy-bound source dataset digest differs")

    receipt_fields = {
        "schema_version", "invocation_id", "run_id", "generator_identity",
        "scorer_code_sha256", "scorer_config_sha256",
        "candidate_bundle_sha256", "source_dataset_sha256",
        "population_sha256", "row_count", "started_at", "finished_at",
    }

    def validate_manifest(
        current: dict, role: str, receipt_key: Path
    ) -> tuple[Path, datetime, dict]:
        digest_fields = (
            "scorer_code_sha256", "scorer_config_sha256",
            "execution_receipt_sha256", "source_dataset_sha256",
            "population_sha256",
        )
        if (
            not isinstance(current, dict)
            or set(current) != manifest_fields
            or current["schema_version"] != "aidefend.fairness_population.v2"
            or not isinstance(current["run_id"], str)
            or not current["run_id"]
            or not isinstance(current["generator_identity"], str)
            or not current["generator_identity"]
            or any(
                not isinstance(current[field], str)
                or len(current[field]) != 64
                or set(current[field]) - set("0123456789abcdef")
                for field in digest_fields
            )
            or current["release_readback_sha256"] != readback_sha256
            or current["candidate_bundle_sha256"] != readback.get("bundle_sha256")
            or current["selection_sha256"] != selection_sha256
            or current["source_dataset_id"] != plan["source_dataset_id"]
            or current["source_dataset_path"] != plan["source_dataset_path"]
            or current["source_dataset_sha256"] != plan["source_dataset_sha256"]
            or positive_integer(current["row_count"], role + ".row_count")
                != expected_row_count
        ):
            raise ValueError(f"{role} fairness manifest or binding differs")
        population_path = fairness_path(current["population_path"])
        if digest(population_path) != current["population_sha256"]:
            raise ValueError(f"{role} fairness population digest differs")
        receipt_path = fairness_path(current["execution_receipt_path"])
        receipt_signature = fairness_path(
            current["execution_receipt_signature_path"]
        )
        if digest(receipt_path) != current["execution_receipt_sha256"]:
            raise ValueError(f"{role} execution receipt digest differs")
        verify(receipt_path, receipt_signature, receipt_key)
        receipt = load_json(receipt_path)
        if (
            not isinstance(receipt, dict)
            or set(receipt) != receipt_fields
            or receipt["schema_version"]
                != "aidefend.fairness_execution_receipt.v1"
            or not isinstance(receipt["invocation_id"], str)
            or not receipt["invocation_id"]
            or receipt["run_id"] != current["run_id"]
            or receipt["generator_identity"] != current["generator_identity"]
            or receipt["scorer_code_sha256"]
                != current["scorer_code_sha256"]
            or receipt["scorer_config_sha256"]
                != current["scorer_config_sha256"]
            or receipt["candidate_bundle_sha256"]
                != current["candidate_bundle_sha256"]
            or receipt["source_dataset_sha256"]
                != current["source_dataset_sha256"]
            or receipt["population_sha256"] != current["population_sha256"]
            or positive_integer(receipt["row_count"], role + ".receipt.row_count")
                != expected_row_count
        ):
            raise ValueError(f"{role} scorer execution binding differs")
        started = when(receipt["started_at"])
        finished = when(receipt["finished_at"])
        generated = when(current["generated_at"])
        if (
            not started <= finished <= generated <= now
            or generated >= when(readback["expires_at"])
        ):
            raise ValueError(f"{role} fairness execution time is invalid")
        return population_path, generated, receipt

    population, primary_generated_at, primary_receipt = validate_manifest(
        manifest, "primary", Path("keys/fairness-generator-execution.pub")
    )
    replay_population, replay_generated_at, replay_receipt = validate_manifest(
        replay_manifest, "replay", Path("keys/fairness-replay-execution.pub")
    )
    if (
        manifest["run_id"] == replay_manifest["run_id"]
        or manifest["generator_identity"] == replay_manifest["generator_identity"]
        or population == replay_population
        or primary_receipt["invocation_id"]
            == replay_receipt["invocation_id"]
    ):
        raise ValueError("fairness scorers are not independently executed")
    evidence_generated_at = max(primary_generated_at, replay_generated_at)
    expires_at = min(
        evidence_generated_at + timedelta(seconds=max_result_age),
        when(readback["expires_at"]),
    )
    if now >= expires_at:
        raise InsufficientData("fairness population is stale")

    source_headers, source_rows = csv_rows(source_dataset)
    source_fields = set(source_headers)
    required_source_fields = {
        "row_id", plan["label_column"], plan["protected_column"]
    }
    if not required_source_fields.issubset(source_fields):
        raise ValueError("source dataset lacks policy-bound columns")
    source_ids = [row["row_id"] for row in source_rows]
    if (
        len(source_rows) != expected_row_count
        or any(not row_id for row_id in source_ids)
        or len(source_ids) != len(set(source_ids))
        or row_ids_sha256(source_ids) != plan["expected_row_ids_sha256"]
    ):
        raise InsufficientData("signed source snapshot population differs")
    source_by_id = {row["row_id"]: row for row in source_rows}

    expected_columns = {
        "row_id",
        plan["label_column"],
        plan["prediction_column"],
        plan["protected_column"],
    }

    def load_scored_population(path: Path, role: str) -> list[dict]:
        headers, scored_rows = csv_rows(path)
        if set(headers) != expected_columns:
            raise ValueError(f"{role} fairness population columns differ")
        scored_ids = [row["row_id"] for row in scored_rows]
        if (
            len(scored_rows) != expected_row_count
            or any(not row_id for row_id in scored_ids)
            or len(scored_ids) != len(set(scored_ids))
            or set(scored_ids) != set(source_by_id)
            or row_ids_sha256(scored_ids) != plan["expected_row_ids_sha256"]
        ):
            raise InsufficientData(f"{role} scored population is incomplete")
        for row in scored_rows:
            source = source_by_id[row["row_id"]]
            if (
                row[plan["label_column"]] != source[plan["label_column"]]
                or row[plan["protected_column"]]
                    != source[plan["protected_column"]]
            ):
                raise ValueError(f"{role} labels or groups differ from source")
        return scored_rows

    rows = load_scored_population(population, "primary")
    replay_rows = load_scored_population(replay_population, "replay")
    primary_predictions = {
        row["row_id"]: row[plan["prediction_column"]] for row in rows
    }
    replay_predictions = {
        row["row_id"]: row[plan["prediction_column"]] for row in replay_rows
    }
    if primary_predictions != replay_predictions:
        raise InsufficientData("independent fairness replay predictions differ")

    required_group_set = set(required_groups)
    observed_groups = {
        row[plan["protected_column"]]
        for row in rows
    }
    if observed_groups != required_group_set:
        raise InsufficientData("protected-group population differs")
    grouped = defaultdict(
        lambda: {
            "rows": 0, "selected": 0,
            "positive": 0, "true_positive": 0,
            "negative": 0, "false_positive": 0,
        }
    )
    for row in rows:
        label = row[plan["label_column"]]
        prediction = row[plan["prediction_column"]]
        group = row[plan["protected_column"]]
        if label not in {"0", "1"} or prediction not in {"0", "1"}:
            raise ValueError("fairness label or prediction is not binary")
        bucket = grouped[group]
        bucket["rows"] += 1
        bucket["selected"] += prediction == "1"
        if label == "1":
            bucket["positive"] += 1
            bucket["true_positive"] += prediction == "1"
        else:
            bucket["negative"] += 1
            bucket["false_positive"] += prediction == "1"
    group_counts = dict(grouped)
    for group, counts in grouped.items():
        if (
            counts["rows"] < minimum_rows
            or counts["positive"] < minimum_positive
            or counts["negative"] < minimum_negative
        ):
            raise InsufficientData(
                "fairness coverage threshold not met: " + group
            )

    selection_rates = [
        counts["selected"] / counts["rows"]
        for counts in grouped.values()
    ]
    true_positive_rates = [
        counts["true_positive"] / counts["positive"]
        for counts in grouped.values()
    ]
    false_positive_rates = [
        counts["false_positive"] / counts["negative"]
        for counts in grouped.values()
    ]
    calculated = {
        "demographic_parity_difference":
            max(selection_rates) - min(selection_rates),
        "equalized_odds_difference": max(
            max(true_positive_rates) - min(true_positive_rates),
            max(false_positive_rates) - min(false_positive_rates),
        ),
    }
    metric_values = {
        name: calculated[name] for name in metric_policy
    }
    failures = [
        name
        for name, value in metric_values.items()
        if value > thresholds[name]
    ]
    outcome = "FAIL" if failures else "PASS"
    reasons.extend(
        "threshold_exceeded:" + name for name in failures
    )
except InsufficientData as error:
    outcome = "INSUFFICIENT_DATA"
    reasons.append(str(error))
except (
    OSError,
    subprocess.SubprocessError,
    KeyError,
    TypeError,
    ValueError,
    json.JSONDecodeError,
) as error:
    outcome = "ERROR"
    reasons.append(str(error))

result = {
    "schema_version": "aidefend.fairness_evaluation.v2",
    "control": "AID-M-007.003",
    "plan_id": plan.get("plan_id"),
    "plan_version": plan.get("plan_version"),
    "fairness_plan_sha256": safe_digest(PLAN),
    "boundary_sha256": plan.get("boundary_sha256"),
    "population_manifest_sha256": safe_digest(MANIFEST),
    "replay_manifest_sha256": safe_digest(REPLAY_MANIFEST),
    "source_dataset_sha256": plan.get("source_dataset_sha256"),
    "expected_row_ids_sha256": plan.get("expected_row_ids_sha256"),
    "release_readback_sha256": manifest.get(
        "release_readback_sha256"
    ),
    "row_count": manifest.get("row_count"),
    "group_counts": group_counts,
    "metrics": metric_values,
    "outcome": outcome,
    "reasons": reasons,
    "generated_at": now.isoformat(),
    "expires_at": expires_at.isoformat() if expires_at else None,
}
OUTPUT.parent.mkdir(parents=True, exist_ok=True)
OUTPUT.write_text(
    json.dumps(
        result,
        sort_keys=True,
        separators=(",", ":"),
        allow_nan=False,
    )
    + chr(10),
    encoding="utf-8",
)
if outcome != "PASS":
    raise SystemExit(f"fairness evaluation did not pass: {outcome}")
python fairness/evaluate_fairness.py
cosign sign-blob --yes --key env://FAIRNESS_VERIFIER_SIGNING_KEY --bundle evidence/fairness-evaluation.sig evidence/fairness-evaluation.json
cosign verify-blob --key keys/fairness-verifier.pub --bundle evidence/fairness-evaluation.sig evidence/fairness-evaluation.json
Independent replay

The verifier accepts PASS only after both separately signed scorer manifests bind the same exact boundary, release readback, candidate bundle, source dataset snapshot, all-rows selection, and row-ID population, and their row-level predictions match exactly. It then recomputes coverage and every selected metric locally. A missing row, source-label or group drift, stale scorer, shared scorer identity, replay mismatch, threshold breach, or unsupported task type blocks promotion. This binary-classification result does not establish privacy, toxicity, general human-harm safety, adversarial robustness, or fairness for another task type.

Open-source tools

Commercial / hosted tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-M-007.003: Fairness & Disparity Release Evaluation | AIDEFEND

Loading interactive view…

A Practical Knowledge Base for AI Security Defenses

Version