AID-H-026.001: Hard-Sample Mining & Controlled Detector Retraining

Version 1.20261015 · Actionable sub-technique

Systematically collect detector errors, especially false negatives on malicious prompts and false positives on benign prompts, and use them as governed hard samples for periodic retraining of the prompt injection detector. The purpose is to harden the detector against the exact decision boundaries it currently fails on. Candidate detector versions must be benchmarked against a fixed out-of-loop evaluation set and explicit promotion criteria before replacing the production detector.

Pillars: model, app

Lifecycle: validation, operation, improvement

Scope boundary

Owns governed detector-error collection, hard-sample training, candidate production, the final promotion decision over independently authenticated benchmark and utility evidence, deployment mutation, and complete effective-version readback. It consumes taxonomy evaluation evidence and does not claim a second category-analysis result.

AID-H-026.002: AID-H-026.001 learns from detector errors observed on governed samples; AID-H-026.002 deliberately generates new adversarial variants to expand coverage. Error mining exploits known failures, while fuzzing probes for failures not yet represented.

AID-H-026.003: AID-H-026.001 trains candidates and owns the final promotion and deployed-version decision; AID-H-026.003 independently measures sealed taxonomy coverage and supplies ELIGIBLE or HOLD evidence. Coverage eligibility is a necessary input where required, not permission to deploy; raw predictions, complete case identities and policy bindings are independently checked again by the final gate.

Implementation guidance

AID-H-026.001-G001

Maintain a governed hard-sample repository and periodic retraining loop.
Purpose and admission boundary

Turn authenticated detector errors into a reproducible candidate corpus without allowing an analyst, fuzzer, or training job to self-admit records. Preserve the full reviewed population, enforce the policy-derived per-category cap deterministically, and keep the fixed evaluation population disjoint. The eval_corpus passed to the trainer is a development evaluation corpus used for checkpoint selection, never the sealed out-of-loop holdout. Neither training nor tuning jobs can read sealed holdout cases. The independent benchmark authority checks separation of case and text identities against the admitted training and development manifests before authorizing the holdout run.

Step 1: Build a strict provenance-bound hard-sample record
# File: detector_hardening/hard_samples.py
from __future__ import annotations

import datetime
import hashlib
import json
import re
from dataclasses import asdict, dataclass
from enum import Enum

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


class Disposition(str, Enum):
    FALSE_NEGATIVE = "false_negative"
    FALSE_POSITIVE = "false_positive"
    EVASION = "evasion"
    NEAR_MISS = "near_miss"


@dataclass(frozen=True)
class HardSample:
    sample_id: str
    text: str
    text_sha256: str
    label: str
    disposition: Disposition
    taxonomy_category: str
    detector_sha256: str
    source_ref: str
    reviewer_subject: str
    review_receipt_sha256: str
    observed_at: str


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


def utc_timestamp(value: str) -> str:
    parsed = datetime.datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None or parsed.utcoffset() != datetime.timedelta(0):
        raise ValueError("observed_at must be UTC")
    return parsed.isoformat().replace("+00:00", "Z")


def build_hard_sample(
    *, text: str, label: str, disposition: Disposition,
    taxonomy_category: str, detector_sha256: str, source_ref: str,
    reviewer_subject: str, review_receipt_sha256: str, observed_at: str,
) -> HardSample:
    if not isinstance(text, str) or not text.strip() or label not in {"malicious", "benign"}:
        raise ValueError("hard-sample text or label differs")
    if any(not isinstance(value, str) or not value for value in
           (taxonomy_category, source_ref, reviewer_subject)):
        raise ValueError("hard-sample identity is incomplete")
    if not SHA256.fullmatch(detector_sha256) or not SHA256.fullmatch(review_receipt_sha256):
        raise ValueError("detector or review receipt digest is invalid")
    text_sha256 = hashlib.sha256(text.encode("utf-8")).hexdigest()
    record = {
        "text_sha256": text_sha256, "label": label,
        "disposition": disposition.value, "taxonomy_category": taxonomy_category,
        "detector_sha256": detector_sha256, "source_ref": source_ref,
        "reviewer_subject": reviewer_subject,
        "review_receipt_sha256": review_receipt_sha256,
        "observed_at": utc_timestamp(observed_at),
    }
    return HardSample(sample_id=hashlib.sha256(canonical(record)).hexdigest(),
                      text=text, **record)
Step 2: Reconcile and deterministically cap every category
# File: detector_hardening/corpus_merge.py
import hashlib
import math

import pandas as pd

from detector_hardening.hard_samples import HardSample, canonical

BASE_COLUMNS = {"sample_id", "text", "label", "taxonomy_category", "source_ref"}


def merge_hard_samples(
    base_corpus: pd.DataFrame,
    hard_samples: list[HardSample],
    hard_sample_final_fraction: float,
    selection_seed: str,
) -> dict:
    if (set(base_corpus.columns) != BASE_COLUMNS or base_corpus.empty
            or base_corpus["sample_id"].isna().any()
            or base_corpus["sample_id"].duplicated().any()
            or not base_corpus["label"].isin(["malicious", "benign"]).all()):
        raise ValueError("base corpus schema, population, IDs, or labels differ")
    if (isinstance(hard_sample_final_fraction, bool)
            or not isinstance(hard_sample_final_fraction, (int, float))
            or not 0.0 <= hard_sample_final_fraction < 1.0 or not selection_seed):
        raise ValueError("signed hard-sample fraction or seed is invalid")
    if not hard_samples:
        raise ValueError("reviewed hard-sample population is empty")

    hard_ids = [sample.sample_id for sample in hard_samples]
    if len(set(hard_ids)) != len(hard_ids):
        raise ValueError("hard-sample IDs are duplicated")
    base_ids = set(base_corpus["sample_id"].astype(str))
    if base_ids.intersection(hard_ids):
        raise ValueError("base and hard-sample IDs overlap")
    base_text = {hashlib.sha256(str(value).encode("utf-8")).hexdigest()
                 for value in base_corpus["text"]}
    if base_text.intersection(sample.text_sha256 for sample in hard_samples):
        raise ValueError("base and hard-sample text populations overlap")

    categories = set(base_corpus["taxonomy_category"].astype(str))
    if {sample.taxonomy_category for sample in hard_samples} - categories:
        raise ValueError("hard sample is outside the governed taxonomy")
    selected: list[HardSample] = []
    dispositions: list[dict] = []
    for category in sorted(categories):
        base_count = int((base_corpus["taxonomy_category"] == category).sum())
        eligible = [sample for sample in hard_samples if sample.taxonomy_category == category]
        maximum = math.floor(
            base_count * hard_sample_final_fraction / (1.0 - hard_sample_final_fraction)
        )
        ordered = sorted(eligible, key=lambda sample: hashlib.sha256(
            (selection_seed + ":" + category + ":" + sample.sample_id).encode("utf-8")
        ).hexdigest())
        admitted_ids = {sample.sample_id for sample in ordered[:maximum]}
        selected.extend(sample for sample in ordered if sample.sample_id in admitted_ids)
        dispositions.extend({
            "sample_id": sample.sample_id, "category": category,
            "status": "selected" if sample.sample_id in admitted_ids else "excluded_by_cap",
        } for sample in ordered)

    if len(dispositions) != len(hard_samples):
        raise RuntimeError("hard-sample disposition population is incomplete")
    base_rows = base_corpus.assign(record_type="base")
    hard_rows = pd.DataFrame([{
        "sample_id": sample.sample_id, "text": sample.text, "label": sample.label,
        "taxonomy_category": sample.taxonomy_category,
        "source_ref": sample.source_ref, "record_type": "reviewed_hard_sample",
    } for sample in selected], columns=sorted(BASE_COLUMNS | {"record_type"}))
    merged = pd.concat([base_rows, hard_rows], ignore_index=True)
    if merged["sample_id"].duplicated().any():
        raise RuntimeError("merged corpus IDs are not unique")
    return {"corpus": merged, "eligible_count": len(hard_samples),
            "selected_count": len(selected), "dispositions": dispositions,
            "corpus_sha256": hashlib.sha256(canonical(
                merged.sort_values("sample_id").to_dict(orient="records"))).hexdigest()}
Step 3: Train only from exact local assets and emit an immutable candidate manifest
# File: detector_hardening/retrain.py
import hashlib
from dataclasses import dataclass
from pathlib import Path

import mlflow
import pandas as pd
from datasets import Dataset
from transformers import (AutoModelForSequenceClassification, AutoTokenizer,
                          DataCollatorWithPadding, Trainer, TrainingArguments)

from detector_hardening.corpus_merge import BASE_COLUMNS
from detector_hardening.hard_samples import canonical


@dataclass(frozen=True)
class TrainingPolicy:
    tokenizer_max_length: int
    num_train_epochs: float
    train_batch_size: int
    learning_rate: float
    weight_decay: float
    eval_strategy: str
    save_strategy: str
    load_best_model_at_end: bool
    metric_for_best_model: str
    greater_is_better: bool
    seed: int
    registered_model_name: str


def tree_sha256(root: Path) -> str:
    if not root.is_dir():
        raise ValueError("model tree is absent")
    digest = hashlib.sha256()
    files = sorted(path for path in root.rglob("*") if path.is_file())
    if not files:
        raise ValueError("model tree is empty")
    for path in files:
        digest.update(path.relative_to(root).as_posix().encode("utf-8") + b"\0")
        digest.update(hashlib.sha256(path.read_bytes()).digest())
    return digest.hexdigest()


def dataframe_sha256(frame: pd.DataFrame) -> str:
    return hashlib.sha256(canonical(
        frame.sort_values("sample_id").to_dict(orient="records"))).hexdigest()


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


def retrain_detector(
    train_corpus: pd.DataFrame, eval_corpus: pd.DataFrame,
    base_model_path: str, expected_base_model_sha256: str,
    output_dir: str, policy: TrainingPolicy,
) -> dict:
    positive_integer(policy.tokenizer_max_length, "tokenizer_max_length")
    positive_integer(policy.train_batch_size, "train_batch_size")
    if (isinstance(policy.num_train_epochs, bool) or policy.num_train_epochs <= 0
            or isinstance(policy.learning_rate, bool) or policy.learning_rate <= 0
            or isinstance(policy.weight_decay, bool) or policy.weight_decay < 0
            or not policy.eval_strategy or not policy.save_strategy
            or not policy.metric_for_best_model or not policy.registered_model_name
            or not isinstance(policy.load_best_model_at_end, bool)
            or not isinstance(policy.greater_is_better, bool)
            or isinstance(policy.seed, bool) or not isinstance(policy.seed, int)):
        raise ValueError("signed training policy value differs")
    required = BASE_COLUMNS | {"record_type"}
    for name, frame in (("train", train_corpus), ("eval", eval_corpus)):
        if (set(frame.columns) != required or frame.empty
                or frame["sample_id"].duplicated().any()
                or set(frame["label"]) != {"malicious", "benign"}):
            raise ValueError(name + " corpus schema or population differs")
    if (set(train_corpus["sample_id"]) & set(eval_corpus["sample_id"])):
        raise ValueError("training and evaluation case IDs overlap")
    train_text = {hashlib.sha256(str(value).encode("utf-8")).hexdigest()
                  for value in train_corpus["text"]}
    eval_text = {hashlib.sha256(str(value).encode("utf-8")).hexdigest()
                 for value in eval_corpus["text"]}
    if train_text & eval_text:
        raise ValueError("training and evaluation text populations overlap")

    base_path, target = Path(base_model_path), Path(output_dir)
    if tree_sha256(base_path) != expected_base_model_sha256:
        raise RuntimeError("base-model tree differs from signed scope")
    if target.exists() and any(target.iterdir()):
        raise RuntimeError("candidate output directory is not empty")
    target.mkdir(parents=True, exist_ok=True)
    tokenizer = AutoTokenizer.from_pretrained(base_path, local_files_only=True)
    model = AutoModelForSequenceClassification.from_pretrained(
        base_path, num_labels=2, local_files_only=True)
    label_to_id = {"benign": 0, "malicious": 1}

    def dataset(frame: pd.DataFrame) -> Dataset:
        value = Dataset.from_pandas(frame[["text", "label"]], preserve_index=False)
        def encode(batch):
            encoded = tokenizer(batch["text"], truncation=True,
                                max_length=policy.tokenizer_max_length)
            encoded["labels"] = [label_to_id[item] for item in batch["label"]]
            return encoded
        return value.map(encode, batched=True, remove_columns=value.column_names)

    arguments = TrainingArguments(
        output_dir=str(target), num_train_epochs=policy.num_train_epochs,
        per_device_train_batch_size=policy.train_batch_size,
        learning_rate=policy.learning_rate, weight_decay=policy.weight_decay,
        eval_strategy=policy.eval_strategy, save_strategy=policy.save_strategy,
        load_best_model_at_end=policy.load_best_model_at_end,
        metric_for_best_model=policy.metric_for_best_model,
        greater_is_better=policy.greater_is_better, seed=policy.seed,
    )
    trainer = Trainer(model=model, args=arguments, train_dataset=dataset(train_corpus),
                      eval_dataset=dataset(eval_corpus),
                      data_collator=DataCollatorWithPadding(tokenizer=tokenizer))
    with mlflow.start_run(run_name="detector-hardening") as run:
        trainer.train()
        trainer.save_model(target)
        tokenizer.save_pretrained(target)
        candidate_sha256 = tree_sha256(target)
        mlflow.log_dict({"train_sha256": dataframe_sha256(train_corpus),
                         "eval_sha256": dataframe_sha256(eval_corpus),
                         "base_model_sha256": expected_base_model_sha256,
                         "candidate_sha256": candidate_sha256}, "candidate-manifest.json")
        mlflow.transformers.log_model(
            transformers_model={"model": model, "tokenizer": tokenizer},
            artifact_path="detector", registered_model_name=policy.registered_model_name)
        return {"run_id": run.info.run_id, "candidate_sha256": candidate_sha256,
                "train_sha256": dataframe_sha256(train_corpus),
                "eval_sha256": dataframe_sha256(eval_corpus)}
Verify complete coverage

Materialize the complete population of every eligible error or approved adversarial/benign sample, every selected or excluded member, every training/evaluation record, and the produced candidate files before evaluation; require stable unique IDs and reject duplicates or undeclared members.

Verify safely

Emit a canonical signed receipt binding AID-H-026.001-G001, policy ID/version/digest, source and selected-corpus roots, per-category before/after counts, exclusion reasons, base/candidate digests, training-policy and run-manifest digests, and registry write/readback, complete-population manifest digest and counts, immutable code/model/tool/container versions, timestamps, disposition reasons, and the applied action. Write the receipt and exact replay inputs to an append-only or WORM sink under an identity that the detector and training job cannot alter, for the policy-derived retention period.

AID-H-026.001-G002

Make and enforce the final detector-promotion decision from authenticated complete holdout, prediction, baseline, latency, and actual candidate-byte evidence.
Install the shared signed-input verifier

Install detector_hardening/sealed_holdout.py from AID-H-026.003-G002 alongside the following modules. Provision its fixed, role-specific public trust under /opt/aidefend/trust/detector-promotion independently of training jobs.

Canonical promotion authority

Consume H-026.003’s independently signed sealed benchmark, then recompute category and aggregate effects and join every latency observation to the exact measured case/request. Only this release authority may produce a PROMOTE decision and mutate deployed detector versions. Mount role-specific verification keys from the independent deployment authority outside training/model control. The scope authority, benchmark observer, latency observer, and final decision signer use distinct protected identities.

Recompute the complete decision
# File: detector_hardening/promotion_evaluation.py
from __future__ import annotations
import hashlib
import math
import os
import stat
from datetime import datetime, timezone
from pathlib import Path
from detector_hardening.sealed_holdout import canonical, verify_document

SCOPE_FIELDS={"schema_version","policy_id","policy_version","expires_at","candidate_sha256",
    "candidate_version","holdout_manifest_sha256","baseline_report_sha256","required_categories",
    "benign_category","min_malicious_recall","min_category_recall","max_benign_fpr",
    "max_category_regression","latency_quantile","quantile_method","max_latency_ms","max_candidate_bytes",
    "deployment_replicas","deployment_generation","decision_store_timeout_seconds"}
BENCHMARK_FIELDS={"schema_version","evaluation_policy_sha256","baseline_report_sha256",
    "detector_version","detector_sha256","holdout_id","holdout_version","holdout_manifest_sha256",
    "holdout_signature_sha256","expected_case_count","measured_case_count","raw_predictions",
    "raw_predictions_sha256","stratified_report","weak_categories","eligibility"}
ROW_FIELDS={"case_id","label","pred","category","request_id"}
LATENCY_FIELDS={"schema_version","detector_sha256","evaluation_policy_sha256","holdout_manifest_sha256","observations"}
LATENCY_ROW_FIELDS={"case_id","request_id","latency_ms"}

def number(value,name,minimum=0.0,maximum=None):
    if isinstance(value,bool) or not isinstance(value,(int,float)):
        raise ValueError(name+" must be numeric")
    result=float(value)
    if not math.isfinite(result) or result<minimum or (maximum is not None and result>maximum):
        raise ValueError(name+" is outside its finite domain")
    return result

def measure_candidate(root: Path) -> tuple[str,int]:
    # The gate's immutable read-only staging tree is unavailable to training jobs.
    if not root.is_dir() or root.is_symlink():raise ValueError("candidate tree missing")
    digest=hashlib.sha256();size=0;files=[]
    for path in sorted(root.rglob("*")):
        if path.is_symlink():raise PermissionError("candidate contains symlink")
        if path.is_file():files.append(path)
        elif not path.is_dir():raise PermissionError("candidate contains special file")
    if not files:raise ValueError("candidate tree empty")
    for path in files:
        fd=os.open(path,os.O_RDONLY|getattr(os,"O_NOFOLLOW",0)|getattr(os,"O_CLOEXEC",0))
        try:
            before=os.fstat(fd)
            if not stat.S_ISREG(before.st_mode):raise PermissionError("candidate file changed type")
            content=hashlib.sha256();count=0
            while chunk:=os.read(fd,1024*1024):content.update(chunk);count+=len(chunk)
            after=os.fstat(fd)
            if (before.st_dev,before.st_ino,before.st_size,before.st_mtime_ns)!=(after.st_dev,after.st_ino,after.st_size,after.st_mtime_ns) or count!=before.st_size:
                raise PermissionError("candidate file changed while measured")
            digest.update(path.relative_to(root).as_posix().encode()+b"\0");digest.update(content.digest());size+=count
        finally:os.close(fd)
    return digest.hexdigest(),size

def rows_by_case(result: dict,holdout: dict,required_categories: list[str],benign: str) -> dict:
    if set(result)!=BENCHMARK_FIELDS or result["schema_version"]!="aidefend.sealed-detector-benchmark.v1":
        raise ValueError("sealed benchmark schema differs")
    if result["eligibility"] not in {"ELIGIBLE","HOLD"}:raise ValueError("invalid benchmark eligibility")
    if result["holdout_id"]!=holdout["holdout_id"] or result["holdout_version"]!=holdout["version"]:
        raise PermissionError("benchmark holdout identity differs")
    samples=holdout["samples"]
    expected={}
    for sample in samples:
        if sample["case_id"] in expected:raise ValueError("duplicate sealed holdout case")
        if sample["category"] not in required_categories or type(sample["label_int"]) is not int:
            raise ValueError("holdout taxonomy or label differs")
        label=0 if sample["category"]==benign else 1
        if sample["label_int"]!=label:raise ValueError("holdout category contains mixed label population")
        expected[sample["case_id"]]=(label,sample["category"])
    rows=result["raw_predictions"]
    if not isinstance(rows,list) or not rows:raise ValueError("benchmark population empty")
    if hashlib.sha256(canonical(rows)).hexdigest()!=result["raw_predictions_sha256"]:
        raise PermissionError("raw benchmark prediction digest differs")
    selected={};requests=set()
    for row in rows:
        if not isinstance(row,dict) or set(row)!=ROW_FIELDS:raise ValueError("prediction schema differs")
        if (not isinstance(row["case_id"],str) or row["case_id"] in selected
                or not isinstance(row["request_id"],str) or not row["request_id"] or row["request_id"] in requests
                or type(row["label"]) is not int or type(row["pred"]) is not int
                or row["pred"] not in (0,1) or expected.get(row["case_id"])!=(row["label"],row["category"])):
            raise PermissionError("prediction identity or sealed population differs")
        selected[row["case_id"]]=row;requests.add(row["request_id"])
    if set(selected)!=set(expected) or type(result["expected_case_count"]) is not int or type(result["measured_case_count"]) is not int:
        raise PermissionError("sealed benchmark case coverage incomplete")
    if result["expected_case_count"]!=len(expected) or result["measured_case_count"]!=len(expected):
        raise PermissionError("sealed benchmark count differs")
    if {r["category"] for r in rows}!=set(required_categories):raise PermissionError("taxonomy category missing")
    return selected

def category_metrics(rows: dict, categories: list[str],benign: str) -> tuple[dict,float,float]:
    grouped={category:[row for row in rows.values() if row["category"]==category] for category in categories}
    if any(not population for population in grouped.values()):raise ValueError("required category missing")
    recall={category:sum(r["pred"] for r in population)/len(population) for category,population in grouped.items() if category!=benign}
    malicious=[r for r in rows.values() if r["label"]==1]
    if not malicious:raise ValueError("malicious evaluation population empty")
    return recall,sum(r["pred"] for r in malicious)/len(malicious),sum(r["pred"] for r in grouped[benign])/len(grouped[benign])

def evaluate_candidate(candidate_root: Path, *, scope_bytes: bytes,scope_signature: bytes,
        benchmark_bytes: bytes,benchmark_signature: bytes,baseline_bytes: bytes,baseline_signature: bytes,
        holdout_bytes: bytes,holdout_signature: bytes,latency_bytes: bytes,latency_signature: bytes) -> dict:
    scope=verify_document("scope",scope_bytes,scope_signature)
    benchmark=verify_document("benchmark",benchmark_bytes,benchmark_signature)
    baseline=verify_document("benchmark",baseline_bytes,baseline_signature)
    holdout=verify_document("holdout",holdout_bytes,holdout_signature)
    latency=verify_document("latency",latency_bytes,latency_signature)
    if set(scope)!=SCOPE_FIELDS or scope["schema_version"]!="aidefend.detector-promotion-scope.v1":
        raise ValueError("promotion scope schema differs")
    expiry=datetime.fromisoformat(scope["expires_at"].replace("Z","+00:00"))
    if expiry.tzinfo is None or expiry<=datetime.now(timezone.utc):raise PermissionError("promotion scope expired")
    for name in ("min_malicious_recall","min_category_recall","max_benign_fpr","max_category_regression","latency_quantile"):
        number(scope[name],name,maximum=1.0)
    number(scope["max_latency_ms"],"max_latency_ms")
    number(scope["decision_store_timeout_seconds"],"decision_store_timeout_seconds")
    if type(scope["max_candidate_bytes"]) is not int or scope["max_candidate_bytes"]<1:raise ValueError("invalid artifact size policy")
    categories=scope["required_categories"];benign=scope["benign_category"]
    if not isinstance(categories,list) or len(categories)<2 or len(set(categories))!=len(categories) or benign not in categories:
        raise ValueError("taxonomy policy differs")
    if any(not isinstance(x,str) or not x for x in categories):raise ValueError("invalid taxonomy name")
    replicas=scope["deployment_replicas"]
    if not isinstance(replicas,list) or not replicas or len(set(replicas))!=len(replicas) or any(not isinstance(x,str) or not x for x in replicas):
        raise ValueError("signed deployment population differs")
    if type(scope["deployment_generation"]) is not int or scope["deployment_generation"]<1:raise ValueError("signed deployment generation invalid")
    if set(holdout)!={"schema_version","holdout_id","version","created_at","samples"} or holdout["schema_version"]!="aidefend.h026-holdout.v1":
        raise ValueError("sealed holdout schema differs")
    if not isinstance(holdout["samples"],list) or not holdout["samples"]:raise ValueError("holdout population empty")
    holdout_sha=hashlib.sha256(holdout_bytes).hexdigest();scope_sha=hashlib.sha256(scope_bytes).hexdigest()
    if scope["holdout_manifest_sha256"]!=holdout_sha or scope["baseline_report_sha256"]!=hashlib.sha256(baseline_bytes).hexdigest():
        raise PermissionError("holdout or baseline differs from signed scope")
    candidate_sha,size=measure_candidate(candidate_root)
    if (scope["candidate_sha256"]!=candidate_sha or benchmark["detector_sha256"]!=candidate_sha
            or benchmark["detector_version"]!=scope["candidate_version"] or benchmark["evaluation_policy_sha256"]!=scope_sha
            or benchmark["baseline_report_sha256"]!=scope["baseline_report_sha256"]):
        raise PermissionError("candidate or evaluation-policy binding differs")
    for document in (benchmark,baseline):
        if document["holdout_manifest_sha256"]!=holdout_sha or document["holdout_signature_sha256"]!=hashlib.sha256(holdout_signature).hexdigest():
            raise PermissionError("benchmark refers to another sealed holdout")
    candidate_rows=rows_by_case(benchmark,holdout,categories,benign)
    baseline_rows=rows_by_case(baseline,holdout,categories,benign)
    recall,malicious_recall,benign_fpr=category_metrics(candidate_rows,categories,benign)
    baseline_recall,_,_=category_metrics(baseline_rows,categories,benign)
    if set(latency)!=LATENCY_FIELDS or latency["schema_version"]!="aidefend.detector-latency.v1":raise ValueError("latency evidence schema differs")
    if (latency["detector_sha256"]!=candidate_sha or latency["evaluation_policy_sha256"]!=scope_sha or latency["holdout_manifest_sha256"]!=holdout_sha):
        raise PermissionError("latency evidence scope differs")
    observed={}
    for row in latency["observations"]:
        if not isinstance(row,dict) or set(row)!=LATENCY_ROW_FIELDS or row["case_id"] in observed:
            raise ValueError("latency row schema or uniqueness differs")
        expected=candidate_rows.get(row["case_id"])
        if expected is None or row["request_id"]!=expected["request_id"]:raise PermissionError("latency-to-prediction join differs")
        observed[row["case_id"]]=number(row["latency_ms"],"latency_ms")
    if set(observed)!=set(candidate_rows):raise PermissionError("latency population incomplete or extra")
    times=sorted(observed.values());q=scope["latency_quantile"]
    if scope["quantile_method"]=="linear":
        position=(len(times)-1)*q;lower=math.floor(position);upper=math.ceil(position)
        quantile=times[lower]+(times[upper]-times[lower])*(position-lower)
    elif scope["quantile_method"]=="inverted_cdf":quantile=times[max(0,math.ceil(q*len(times))-1)]
    else:raise ValueError("unsupported versioned quantile profile")
    failures=[]
    if benchmark["eligibility"]!="ELIGIBLE" or benchmark["weak_categories"]:failures.append("coverage_veto")
    if malicious_recall<scope["min_malicious_recall"]:failures.append("malicious_recall_below_policy")
    if benign_fpr>scope["max_benign_fpr"]:failures.append("benign_fpr_above_policy")
    if any(v<scope["min_category_recall"] for v in recall.values()):failures.append("category_recall_below_policy")
    if any(baseline_recall[k]-recall[k]>scope["max_category_regression"] for k in recall):failures.append("category_recall_regression")
    if quantile>scope["max_latency_ms"]:failures.append("latency_above_policy")
    if size>scope["max_candidate_bytes"]:failures.append("candidate_size_above_policy")
    return {"schema_version":"aidefend.detector-promotion-decision.v1","control":"AID-H-026.001",
        "candidate_sha256":candidate_sha,"candidate_version":scope["candidate_version"],"candidate_bytes":size,
        "policy_id":scope["policy_id"],"policy_version":scope["policy_version"],"scope_sha256":scope_sha,
        "benchmark_sha256":hashlib.sha256(benchmark_bytes).hexdigest(),"baseline_sha256":hashlib.sha256(baseline_bytes).hexdigest(),
        "holdout_manifest_sha256":holdout_sha,"latency_evidence_sha256":hashlib.sha256(latency_bytes).hexdigest(),
        "complete_case_count":len(candidate_rows),"category_recall":recall,"malicious_recall":malicious_recall,
        "benign_fpr":benign_fpr,"latency_quantile_ms":quantile,"violations":failures,
        "expires_at":scope["expires_at"],"deployment_replicas":replicas,
        "deployment_generation":scope["deployment_generation"],
        "decision_store_timeout_seconds":scope["decision_store_timeout_seconds"],"decision":"PROMOTE" if not failures else "HOLD"}
Sign and independently read back the final decision
# File: detector_hardening/promotion_decision.py
from detector_hardening.sealed_holdout import canonical, verify_document
from detector_hardening.promotion_evaluation import evaluate_candidate

def record_promotion_decision(candidate_root,evaluation_inputs: dict,append_and_readback) -> dict:
    # The signer never accepts a caller-supplied PROMOTE dictionary.
    evaluation=evaluate_candidate(candidate_root,**evaluation_inputs)
    if evaluation.get("control")!="AID-H-026.001" or evaluation.get("decision") not in {"PROMOTE","HOLD"}:
        raise ValueError("not a final detector-promotion decision")
    payload=canonical(evaluation)
    readback,signature=append_and_readback(payload)
    observed=verify_document("decision",readback,signature)
    if readback!=payload or observed!=evaluation:raise PermissionError("signed final-decision readback differs")
    return {"payload":readback,"signature":signature}
Mutate and verify the complete deployed population
# File: detector_hardening/promotion_deploy.py
import hashlib
from datetime import datetime,timezone
from detector_hardening.sealed_holdout import canonical, verify_document
from detector_hardening.promotion_evaluation import measure_candidate
from detector_hardening.promotion_store import OneUsePromotionStore, one_use_mutation

def promote_and_verify(decision_record: dict,candidate_root,
        mutate_target,read_targets,rollback_target,append_and_readback, *,
        decision_store: OneUsePromotionStore, read_current_authority) -> dict:
    if not isinstance(decision_store, OneUsePromotionStore) or not callable(read_current_authority):
        raise TypeError("an existing independent decision store and authority reader are required")
    # Verify independently signed bytes again at the mutation boundary.
    decision=verify_document("decision",decision_record["payload"],decision_record["signature"])
    if (decision.get("schema_version")!="aidefend.detector-promotion-decision.v1"
            or decision.get("control")!="AID-H-026.001" or decision.get("decision")!="PROMOTE"
            or decision.get("violations")):
        raise PermissionError("detector promotion blocked")
    expiry=datetime.fromisoformat(decision["expires_at"].replace("Z","+00:00"))
    if expiry.tzinfo is None or expiry<=datetime.now(timezone.utc):raise PermissionError("decision expired")
    digest,size=measure_candidate(candidate_root)
    if digest!=decision["candidate_sha256"] or size!=decision["candidate_bytes"]:
        raise PermissionError("candidate changed before mutation")
    expected_replicas=tuple(decision["deployment_replicas"])
    if not expected_replicas or len(set(expected_replicas))!=len(expected_replicas) or any(not isinstance(r,str) or not r for r in expected_replicas):
        raise ValueError("deployment replica population differs")
    # The authenticated deployment adapter resolves the digest only from the
    # immutable candidate store, uses one_use_mutation with an authority-owned OneUsePromotionStore to durably consume this one-time decision,
    # and compares current deployment generation/policy before mutation.
    mutation=one_use_mutation(decision_store,read_current_authority,mutate_target)(digest,decision_record)
    try:
        targets=read_targets()
        if not isinstance(targets,dict) or set(targets)!=set(expected_replicas) or any(v!=digest for v in targets.values()):
            raise PermissionError("effective deployment population differs")
        event={"schema_version":"aidefend.detector-promotion-effect.v1","control":"AID-H-026.001",
            "candidate_sha256":digest,"decision_sha256":hashlib.sha256(decision_record["payload"]).hexdigest(),
            "mutation_receipt":mutation,"effective_targets":targets}
        payload=canonical(event);readback,signature=append_and_readback(payload)
        if verify_document("decision",readback,signature)!=event or readback!=payload:
            raise PermissionError("deployment evidence differs")
        return {"effect":event,"payload":readback,"signature":signature}
    except Exception:
        rollback_target(mutation)
        raise
Deployment integration and verification

The release worker receives immutable scope and evidence bytes from separately authenticated read-only stores and signs only the result of the local evaluator. Its signer is unavailable to training, benchmarking or callers. Publish candidates into a read-only digest-addressed store before evaluation; the deployment adapter must verify and atomically consume the same signed decision, enforce current generation/policy, and load only that immutable digest. Missing replicas, divergent readback, evidence uncertainty or rollback failure remain unresolved release failures; never report complete promotion from a command acknowledgment. Test forged signatures, candidate/holdout/policy substitutions, missing/extra/duplicate cases, mixed category labels, unmatched latency requests, HOLD coverage, artifact changes and incomplete replica readback. Numeric gates and supported quantile algorithms come from the versioned scope; this profile implements linear and inverted-CDF quantiles and rejects unsupported profiles.

Durably consume authorization before changing any target

The mutation adapter is part of this required path. Provision its store once under the independent deployment authority with python -m detector_hardening.promotion_store --provision /var/lib/aidefend/detector-promotion/decisions.sqlite --timeout-seconds "$AIDEFEND_SIGNED_POLICY_LOCK_TIMEOUT_SECONDS". Mount the existing store and parent under service-owned permissions; runtime refuses missing stores. Pass decision_store=OneUsePromotionStore(existing_path, verified_scope["decision_store_timeout_seconds"]) and read_current_authority=authority_reader as mandatory keyword arguments to promote_and_verify, which applies one_use_mutation internally before invoking the actual deployment API. An older call with only a mutation callback is rejected. The authority reader returns the active signed scope digest and actual current deployment generation from the deployment control plane, never from the presented decision. The mutation endpoint must CAS-check that same scope and generation against live target state; a prior read alone does not make a distributed mutation atomic. Keep this adapter and the mutation endpoint unavailable to training jobs. Reservations survive restart, failures and rollback; reconcile uncertain mutations before issuing a fresh authorized generation, and never rewind the deployment epoch when restoring prior content.

# File: detector_hardening/promotion_store.py
from __future__ import annotations

import hashlib
import math
import os
from datetime import datetime, timezone
from pathlib import Path
import sqlite3

from detector_hardening.sealed_holdout import verify_document


def validate_lock_timeout(value: object) -> float:
    if type(value) not in (int, float) or not math.isfinite(value) or value < 0:
        raise ValueError("verified policy lock timeout must be finite and nonnegative")
    return float(value)


def provision_decision_store(path: Path, lock_timeout_seconds: float) -> None:
    # Run once under the deployment authority. Runtime never initializes a store.
    lock_timeout_seconds = validate_lock_timeout(lock_timeout_seconds)
    path.parent.mkdir(parents=True, exist_ok=True)
    descriptor = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
    os.close(descriptor)
    database = sqlite3.connect(path, timeout=lock_timeout_seconds)
    try:
        database.execute("PRAGMA synchronous=FULL")
        database.execute("PRAGMA journal_mode=WAL")
        database.execute("""CREATE TABLE consumed (
            decision_sha256 TEXT PRIMARY KEY,
            scope_sha256 TEXT NOT NULL,
            deployment_generation INTEGER NOT NULL UNIQUE,
            candidate_sha256 TEXT NOT NULL
        )""")
        database.commit()
    finally:
        database.close()


class OneUsePromotionStore:
    def __init__(self, path: Path, lock_timeout_seconds: float):
        self.lock_timeout_seconds = validate_lock_timeout(lock_timeout_seconds)
        # The independently managed mount and its parent are not job-writable.
        if path.is_symlink() or not path.is_file():
            raise FileNotFoundError("deployment decision store is not provisioned")
        uri = path.resolve(strict=True).as_uri() + "?mode=rw"
        self.database = sqlite3.connect(uri, uri=True, timeout=self.lock_timeout_seconds, isolation_level=None)
        try:
            self.database.execute("PRAGMA synchronous=FULL")
            self.database.execute("SELECT decision_sha256 FROM consumed LIMIT 0")
        except BaseException:
            self.database.close()
            raise

    def close(self) -> None:
        self.database.close()

    def consume(self, record: dict, digest: str, active_scope_sha256: str,
                current_generation: int) -> None:
        # Read these two authority values from the live deployment control plane,
        # not from the record, benchmark, training job or HTTP request.
        decision = verify_document("decision", record["payload"], record["signature"])
        expiry = datetime.fromisoformat(decision["expires_at"].replace("Z", "+00:00"))
        if expiry.tzinfo is None or expiry <= datetime.now(timezone.utc):
            raise PermissionError("promotion decision expired before reservation")
        if (decision.get("schema_version") != "aidefend.detector-promotion-decision.v1"
                or decision.get("control") != "AID-H-026.001"
                or decision.get("decision") != "PROMOTE" or decision.get("violations")
                or decision.get("candidate_sha256") != digest
                or decision.get("scope_sha256") != active_scope_sha256
                or decision.get("decision_store_timeout_seconds") != self.lock_timeout_seconds
                or type(current_generation) is not int or current_generation < 0
                or type(decision.get("deployment_generation")) is not int
                or decision["deployment_generation"] != current_generation + 1):
            raise PermissionError("deployment authority or generation differs")
        decision_sha256 = hashlib.sha256(record["payload"]).hexdigest()
        self.database.execute("BEGIN IMMEDIATE")
        try:
            self.database.execute("INSERT INTO consumed VALUES (?,?,?,?)", (
                decision_sha256, active_scope_sha256,
                decision["deployment_generation"], digest))
            self.database.execute("COMMIT")
        except sqlite3.IntegrityError as exc:
            self.database.execute("ROLLBACK")
            raise PermissionError("decision or deployment generation already consumed") from exc
        except BaseException:
            self.database.execute("ROLLBACK")
            raise
        # The durable reservation precedes mutation. Timeout/crash/rollback does
        # not release it. Reconcile actual targets under deployment authority;
        # any retry needs a newly authorized generation, never blind replay.


def one_use_mutation(store: OneUsePromotionStore, read_current_authority, mutate):
    def guarded(digest: str, record: dict):
        scope_sha256, current_generation = read_current_authority()
        store.consume(record, digest, scope_sha256, current_generation)
        # The actual mutation endpoint also CAS-checks this scope/generation
        # against live deployment state. A prior read is not an atomic update.
        return mutate(digest, record)
    return guarded


if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description="Provision an authority-owned one-use decision store")
    parser.add_argument("--provision", type=Path, required=True)
    parser.add_argument("--timeout-seconds", type=float, required=True, help="lock timeout from verified versioned deployment policy")
    arguments = parser.parse_args()
    provision_decision_store(arguments.provision, arguments.timeout_seconds)

Read decision_store_timeout_seconds only from the independently verified, versioned signed promotion scope; validate its finite nonnegative domain and bind it into the signed final decision. Provisioning and runtime require this value explicitly. A store configured with a different timeout rejects the decision. No universal lock-wait deadline is implied.

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-H-026.001: Hard-Sample Mining & Controlled Detector Retraining | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service