AID-D-003.001: Harmful Text & Token Output Policy-Violation Detection
Inspect model-generated text, token streams, and text-bearing structured fields that are intended for user display or downstream application consumption for policy-defined harmful, abusive, deceptive, regulated, or otherwise disallowed language. Emit matched rule IDs, classifier versions, scores, language and stream coverage, and alert evidence. Non-user-facing emitted reasoning fields are owned by AID-D-002.005.
Pillars: app
Lifecycle: operation
Scope boundary
Owns policy-violation findings for generated text, token streams, and text-bearing structured fields intended for display or downstream application use. It does not inspect a non-user-facing emitted reasoning channel, non-text media, or binary containers, and does not block, sanitize, approve, or safely render content.
AID-D-003.008: AID-D-003.001 inspects text and token-bearing fields; AID-D-003.008 inspects generated images, audio, video, documents, archives, serialized objects, executables, and other binary artifacts. Choose the detector by the actual decoded output type, and route mixed outputs through both applicable populations.
AID-D-002.005: AID-D-003.001 monitors text intended for a user or downstream application; AID-D-002.005 monitors a separately identified, operator-visible emitted reasoning field for attack and evasion signals. If the same field is deliberately exposed as ordinary output, assign one primary measured population and do not score it twice.
Implementation guidance
- AID-D-003.001-G001: Deploy a fast safety and abuse classifier to scan AI output for policy violations.
- AID-D-003.001-G002: Use a dedicated critic LLM to review the main model's output for safety and compliance.
- AID-D-003.001-G003: Apply deterministic keyword and regex detectors and emit their matched rule identifiers.
AID-D-003.001-G001
Deploy a fast safety and abuse classifier to scan AI output for policy violations.
Detector boundary and pinned artifacts
Package the tokenizer, configuration, safetensors weights, and label map locally and list every file in a signed manifest. Resolve model and policy artifacts only from a root-owned, read-only service definition; use the deployment-fixed verification keys and versioned Cosign timeout profile below, never request-selected paths, keys, or timeouts. Signed thresholds may name only labels the model exposes. This detector emits findings; AID-H-006 owns deny, redact, rewrite, and release.
Verify artifacts and scan the complete output
from __future__ import annotations
# File: output_filters/safety_classifier.py
import hashlib
import hmac
import json
import math
import os
import re
import subprocess
import tempfile
from pathlib import Path, PurePosixPath
from typing import Literal
import torch
from transformers import AutoModelForSequenceClassification, AutoTokenizer, AutoProcessor, Llama4ForConditionalGeneration
_AIDEFEND_CAPTURED: dict[Path, bytes] = {}
def _aidefend_bytes(path: Path) -> bytes:
normalized = Path(path)
if normalized not in _AIDEFEND_CAPTURED:
_AIDEFEND_CAPTURED[normalized] = normalized.read_bytes()
return _AIDEFEND_CAPTURED[normalized]
def _aidefend_write_private(path: Path, payload: bytes) -> None:
descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
with os.fdopen(descriptor, "wb") as stream:
stream.write(payload)
stream.flush()
os.fsync(stream.fileno())
os.chmod(path, 0o400)
def _aidefend_strict_json(raw: bytes) -> object:
def unique_object(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError("duplicate JSON key: " + key)
value[key] = item
return value
return json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=unique_object,
parse_constant=lambda token: (_ for _ in ()).throw(
ValueError("non-finite JSON constant: " + token)
),
)
def _aidefend_strict_jsonl(raw: bytes) -> list[object]:
lines = raw.decode("utf-8", errors="strict").splitlines()
if not lines or any(not line.strip() for line in lines):
raise ValueError("JSONL input is empty or contains a blank record")
return [_aidefend_strict_json(line.encode("utf-8")) for line in lines]
def _aidefend_capture_signed_bytes(
payload_path: Path,
proof_path: Path,
key_reference: str | Path,
timeout_seconds: float,
) -> bytes:
payload_path = Path(payload_path)
proof_path = Path(proof_path)
if payload_path == proof_path:
raise RuntimeError("signed payload and signature paths collide")
payload_raw = payload_path.read_bytes()
proof_raw = proof_path.read_bytes()
key_text = str(key_reference)
remote_key = bool(re.match(
r"^[A-Za-z][A-Za-z0-9+.-]*://", key_text
)) or key_text.startswith("pkcs11:")
key_raw = None if remote_key else (
key_text.encode("utf-8")
if key_text.startswith("-----BEGIN ")
else Path(key_text).read_bytes()
)
with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
payload_snapshot = root / "payload"
proof_snapshot = root / "payload.sig"
_aidefend_write_private(payload_snapshot, payload_raw)
_aidefend_write_private(proof_snapshot, proof_raw)
captured = [(payload_snapshot, payload_raw), (proof_snapshot, proof_raw)]
key_argument = key_text
if key_raw is not None:
key_snapshot = root / "verify-key.pub"
_aidefend_write_private(key_snapshot, key_raw)
captured.append((key_snapshot, key_raw))
key_argument = str(key_snapshot)
try:
subprocess.run(
["cosign", "verify-blob", "--key", key_argument,
"--bundle", str(proof_snapshot), str(payload_snapshot)],
check=True, capture_output=True, text=True,
timeout=timeout_seconds,
)
if any(
not hmac.compare_digest(snapshot.read_bytes(), expected)
for snapshot, expected in captured
):
raise RuntimeError("private signature-verification snapshot changed")
finally:
for snapshot, _ in captured:
os.chmod(snapshot, 0o600)
_AIDEFEND_CAPTURED[payload_path] = payload_raw
return payload_raw
ControlStatus = Literal["PASS", "INSUFFICIENT_DATA", "ERROR"]
FindingStatus = Literal["FINDING", "NO_FINDING", "UNKNOWN"]
MODEL_DIR = Path(os.environ["D003001_MODEL_DIR"])
MANIFEST = Path(os.environ["D003001_MODEL_MANIFEST"])
MANIFEST_SIGNATURE = Path(os.environ["D003001_MODEL_MANIFEST_SIGNATURE"])
POLICY = Path(os.environ["D003001_OUTPUT_POLICY"])
POLICY_SIGNATURE = Path(os.environ["D003001_OUTPUT_POLICY_SIGNATURE"])
TRUST_DIR = Path("/etc/aidefend/d003001/trust")
MANIFEST_VERIFY_KEY = str(TRUST_DIR / "model-manifest.pub")
POLICY_VERIFY_KEY = str(TRUST_DIR / "output-policy.pub")
RUNTIME_PROFILE_PATH = TRUST_DIR / "runtime-profile.json"
RUNTIME_PROFILE_BYTES = RUNTIME_PROFILE_PATH.read_bytes()
RUNTIME_PROFILE = _aidefend_strict_json(RUNTIME_PROFILE_BYTES)
RUNTIME_PROFILE_SHA256 = hashlib.sha256(RUNTIME_PROFILE_BYTES).hexdigest()
profile_timeout = RUNTIME_PROFILE.get("cosign_verify_timeout_seconds") if isinstance(RUNTIME_PROFILE, dict) else None
if (
not isinstance(RUNTIME_PROFILE, dict)
or set(RUNTIME_PROFILE) != {"schema_version", "profile_version", "cosign_verify_timeout_seconds"}
or RUNTIME_PROFILE["schema_version"] != "aidefend.d003001-runtime-profile.v1"
or not isinstance(RUNTIME_PROFILE["profile_version"], str)
or not RUNTIME_PROFILE["profile_version"]
or isinstance(profile_timeout, bool)
or not isinstance(profile_timeout, (int, float))
or not math.isfinite(float(profile_timeout))
or profile_timeout <= 0
):
raise ValueError("runtime profile is invalid")
VERIFY_TIMEOUT_SECONDS = float(profile_timeout)
def sha256_file(path: Path) -> str:
return hashlib.sha256(_aidefend_bytes(Path(path))).hexdigest()
def verify_blob(path: Path, signature: Path, key: str) -> bytes:
key_path = Path(key)
if key_path.is_symlink():
raise ValueError("verification key must not be a symlink")
key_path = key_path.resolve(strict=True)
if not key_path.is_file() or key_path.stat().st_mode & 0o022:
raise ValueError("verification key must be a non-writable regular file")
return _aidefend_capture_signed_bytes(
path, signature, key_path, VERIFY_TIMEOUT_SECONDS,
)
def bounded_relative_path(value: object) -> PurePosixPath:
if not isinstance(value, str) or not value:
raise ValueError("model manifest path is absent")
path = PurePosixPath(value)
if path.is_absolute() or any(part in {"", ".", ".."} for part in path.parts):
raise ValueError("model manifest path escapes the model directory")
return path
def observe_model_population() -> dict[str, Path]:
if MODEL_DIR.is_symlink():
raise ValueError("model directory must not be a symlink")
root = MODEL_DIR.resolve(strict=True)
if not root.is_dir():
raise ValueError("model directory is not a directory")
observed: dict[str, Path] = {}
for current_root, directory_names, file_names in os.walk(root, followlinks=False):
current = Path(current_root)
for name in directory_names:
candidate = current / name
if candidate.is_symlink() or not candidate.is_dir():
raise ValueError("model directory contains a linked or invalid directory")
for name in file_names:
candidate = current / name
if candidate.is_symlink() or not candidate.is_file():
raise ValueError("model directory contains a linked or non-regular file")
relative = candidate.relative_to(root).as_posix()
if relative in observed:
raise ValueError("model directory contains a duplicate relative path")
observed[relative] = candidate
if not observed:
raise ValueError("model directory contains no files")
return observed
def load_artifacts() -> tuple[
dict, dict, str, str, tempfile.TemporaryDirectory, Path,
]:
manifest_bytes = verify_blob(
MANIFEST, MANIFEST_SIGNATURE, MANIFEST_VERIFY_KEY
)
policy_bytes = verify_blob(POLICY, POLICY_SIGNATURE, POLICY_VERIFY_KEY)
manifest = _aidefend_strict_json(manifest_bytes)
policy = _aidefend_strict_json(policy_bytes)
if (
not isinstance(manifest, dict)
or set(manifest)
!= {"schema_version", "model_id", "model_revision", "model_family", "files"}
or manifest["schema_version"] != "aidefend.local_classifier_manifest.v2"
or manifest["model_family"] not in {"sequence_classifier", "llama_guard4"}
or not isinstance(manifest["model_id"], str)
or not manifest["model_id"]
or not isinstance(manifest["model_revision"], str)
or not manifest["model_revision"]
or not isinstance(manifest["files"], list)
or not manifest["files"]
):
raise ValueError("model manifest schema differs")
declared: dict[str, str] = {}
for record in manifest["files"]:
if (
not isinstance(record, dict)
or set(record) != {"path", "sha256"}
or not isinstance(record["sha256"], str)
or len(record["sha256"]) != 64
or set(record["sha256"]) - set("0123456789abcdef")
):
raise ValueError("model manifest file record is invalid")
relative = bounded_relative_path(record["path"])
key = relative.as_posix()
if key in declared:
raise ValueError("model manifest contains a duplicate path")
declared[key] = record["sha256"]
if (
not any(path.endswith(".safetensors") for path in declared)
or any(path.endswith((".bin", ".pt", ".pth")) for path in declared)
):
raise ValueError("classifier must use safetensors without pickle weight files")
observed = observe_model_population()
missing = sorted(set(declared) - set(observed))
unexpected = sorted(set(observed) - set(declared))
if missing or unexpected:
raise ValueError(
f"model manifest population differs: missing={missing}, unexpected={unexpected}"
)
snapshot_directory = tempfile.TemporaryDirectory(
prefix="aidefend-d003001-model-"
)
snapshot_root = Path(snapshot_directory.name)
os.chmod(snapshot_root, 0o700)
for key, candidate in observed.items():
model_bytes = _aidefend_bytes(candidate)
if hashlib.sha256(model_bytes).hexdigest() != declared[key]:
raise ValueError(f"model artifact digest differs: {key}")
relative = PurePosixPath(key)
destination = snapshot_root.joinpath(*relative.parts)
destination.parent.mkdir(parents=True, exist_ok=True)
current = destination.parent
while current != snapshot_root:
os.chmod(current, 0o700)
current = current.parent
_aidefend_write_private(destination, model_bytes)
if not hmac.compare_digest(destination.read_bytes(), model_bytes):
raise RuntimeError(f"private model snapshot readback differs: {key}")
manifest_sha256 = hashlib.sha256(manifest_bytes).hexdigest()
required_policy = {
"schema_version",
"policy_version",
"model_manifest_sha256",
"score_transform",
"label_thresholds",
"maximum_output_characters",
"maximum_tokens_per_window",
"window_overlap_tokens",
}
if (
not isinstance(policy, dict)
or set(policy) != required_policy
or policy["schema_version"] != "aidefend.output_safety_policy.v1"
or not isinstance(policy["policy_version"], str)
or not policy["policy_version"]
or policy["model_manifest_sha256"] != manifest_sha256
or policy["score_transform"] not in {"sigmoid", "softmax", "categorical"}
or type(policy["maximum_output_characters"]) is not int
or policy["maximum_output_characters"] < 1
or type(policy["maximum_tokens_per_window"]) is not int
or policy["maximum_tokens_per_window"] < 2
or type(policy["window_overlap_tokens"]) is not int
or not 0 <= policy["window_overlap_tokens"] < policy["maximum_tokens_per_window"]
or not isinstance(policy["label_thresholds"], dict)
or not policy["label_thresholds"]
):
raise ValueError("output-safety policy schema differs")
for label, threshold in policy["label_thresholds"].items():
if (
not isinstance(label, str)
or not label
or isinstance(threshold, bool)
or not isinstance(threshold, (int, float))
or not math.isfinite(float(threshold))
or not 0.0 <= float(threshold) <= 1.0
):
raise ValueError("output-safety threshold is invalid")
return (
policy, manifest, manifest_sha256,
hashlib.sha256(policy_bytes).hexdigest(),
snapshot_directory, snapshot_root,
)
(POLICY_DOCUMENT, MODEL_MANIFEST, MODEL_MANIFEST_SHA256, POLICY_SHA256,
MODEL_SNAPSHOT_DIRECTORY, MODEL_SNAPSHOT) = load_artifacts()
if MODEL_MANIFEST["model_family"] == "llama_guard4":
if (MODEL_MANIFEST["model_id"] != "meta-llama/Llama-Guard-4-12B"
or POLICY_DOCUMENT["score_transform"] != "categorical"
or set(POLICY_DOCUMENT["label_thresholds"]) != {"S11"}
or POLICY_DOCUMENT["label_thresholds"]["S11"] != 1.0):
raise RuntimeError("Llama Guard path requires the native S11 decision policy")
TOKENIZER = AutoProcessor.from_pretrained(
MODEL_SNAPSHOT, local_files_only=True, trust_remote_code=False
)
MODEL = Llama4ForConditionalGeneration.from_pretrained(
MODEL_SNAPSHOT, local_files_only=True, trust_remote_code=False,
use_safetensors=True, torch_dtype=torch.bfloat16,
)
else:
if POLICY_DOCUMENT["score_transform"] == "categorical":
raise RuntimeError("sequence classifier requires its calibrated probability transform")
TOKENIZER = AutoTokenizer.from_pretrained(
MODEL_SNAPSHOT, local_files_only=True, trust_remote_code=False, use_fast=True
)
if not getattr(TOKENIZER, "is_fast", False) or TOKENIZER.pad_token_id is None:
raise RuntimeError("pinned tokenizer must support overflow windows and padding")
MODEL = AutoModelForSequenceClassification.from_pretrained(
MODEL_SNAPSHOT, local_files_only=True, trust_remote_code=False, use_safetensors=True,
)
LABELS = {int(index): str(label) for index, label in MODEL.config.id2label.items()}
if (set(LABELS) != set(range(MODEL.config.num_labels))
or len(set(LABELS.values())) != len(LABELS)
or set(POLICY_DOCUMENT["label_thresholds"]) - set(LABELS.values())):
raise RuntimeError("signed policy labels do not match the pinned model")
MODEL.eval()
for record in MODEL_MANIFEST["files"]:
relative = PurePosixPath(record["path"])
snapshot_path = MODEL_SNAPSHOT.joinpath(*relative.parts)
if sha256_file(snapshot_path) != record["sha256"]:
raise RuntimeError("private model snapshot changed during load")
def evidence_binding():
return {
"model_manifest_sha256": MODEL_MANIFEST_SHA256,
"policy_version": POLICY_DOCUMENT["policy_version"],
"policy_sha256": POLICY_SHA256,
"runtime_profile_version": RUNTIME_PROFILE["profile_version"],
"runtime_profile_sha256": RUNTIME_PROFILE_SHA256,
}
def scan_output_safety(response_id: str, text: str) -> dict:
if MODEL_MANIFEST["model_family"] != "sequence_classifier":
raise RuntimeError("use the native Llama Guard adapter for this model family")
if not isinstance(response_id, str) or not response_id or not isinstance(text, str):
raise ValueError("response identity or text is invalid")
response_sha256 = hashlib.sha256(text.encode("utf-8")).hexdigest()
if not text:
return {
"schema_version": "aidefend.output_safety_finding.v1",
"control_status": "PASS",
"finding_status": "NO_FINDING",
"response_id": response_id,
"response_sha256": response_sha256,
"scores": {},
"findings": [],
**evidence_binding(),
}
if len(text) > POLICY_DOCUMENT["maximum_output_characters"]:
return {
"schema_version": "aidefend.output_safety_finding.v1",
"control_status": "INSUFFICIENT_DATA",
"finding_status": "UNKNOWN",
"reason": "output exceeds signed detector character bound",
"response_id": response_id,
"response_sha256": response_sha256,
**evidence_binding(),
}
encoded = TOKENIZER(
text,
return_tensors="pt",
truncation=True,
max_length=POLICY_DOCUMENT["maximum_tokens_per_window"],
stride=POLICY_DOCUMENT["window_overlap_tokens"],
return_overflowing_tokens=True,
padding=True,
)
model_inputs = {
key: value
for key, value in encoded.items()
if key in {"input_ids", "attention_mask", "token_type_ids"}
}
if not model_inputs or model_inputs["input_ids"].shape[0] < 1:
raise RuntimeError("tokenizer produced no classifier window")
with torch.inference_mode():
logits = MODEL(**model_inputs).logits
if logits.ndim != 2 or logits.shape[1] != MODEL.config.num_labels:
raise RuntimeError("classifier output shape differs")
if not torch.isfinite(logits).all():
raise RuntimeError("classifier returned non-finite logits")
probabilities = (
torch.sigmoid(logits)
if POLICY_DOCUMENT["score_transform"] == "sigmoid"
else torch.softmax(logits, dim=-1)
)
maximum_scores = probabilities.max(dim=0).values.tolist()
scores = {
LABELS[index]: round(float(score), 6)
for index, score in enumerate(maximum_scores)
if LABELS[index] in POLICY_DOCUMENT["label_thresholds"]
}
findings = sorted(
label
for label, threshold in POLICY_DOCUMENT["label_thresholds"].items()
if scores[label] >= float(threshold)
)
return {
"schema_version": "aidefend.output_safety_finding.v1",
"control_status": "PASS",
"finding_status": "FINDING" if findings else "NO_FINDING",
"response_id": response_id,
"response_sha256": response_sha256,
"model_id": MODEL_MANIFEST["model_id"],
"model_revision": MODEL_MANIFEST["model_revision"],
**evidence_binding(),
"window_count": int(logits.shape[0]),
"scores": scores,
"findings": findings,
}
def safe_scan_output_safety(response_id: str, text: str) -> dict:
try:
return scan_output_safety(response_id, text)
except Exception as error:
value = text if isinstance(text, str) else ""
return {
"schema_version": "aidefend.output_safety_finding.v1",
"control_status": "ERROR",
"finding_status": "UNKNOWN",
"response_id": response_id,
"response_sha256": hashlib.sha256(value.encode("utf-8")).hexdigest(),
**evidence_binding(),
"error_class": type(error).__name__,
}Exercise the deployed adapter
Calibrate thresholds on a signed, language- and policy-segmented corpus, then replay benign, each supported violation class, Unicode and mixed-script text, long text with the finding placed across a window boundary, empty output, oversized output, corrupt or substituted model files, an unlisted model file, a symlink or special file in the model directory, manifest-policy mismatch, an unknown policy label, malformed tokenizer output, non-finite logits, and model invocation failure. Compare exact response digests, per-label scores, window counts, and findings with the retained fixture expectations. A model or policy update requires the same replay before promotion.
Operational check
Run the classifier over every applicable text output before delivery, write the result to the authenticated event path, and reconcile response IDs from the generation manifest with classifier results. Missing results, tokenizer/model failures, and outputs beyond the signed bound must remain non-clean states for AID-H-006; this detector never decides release.
Package the native Detoxify unbiased labels
Use the official Detoxify unbiased release and its matching pinned tokenizer/configuration. Its upstream checkpoint uses a pickle-based loader. Acquire and independently approve its full SHA-256 in the isolated build pipeline, then run the conversion below there. Serving receives only the exported safetensors files, immutable model revision, signed v2 manifest and signed output policy. Pin the build dependencies and archive the conversion evidence with their lockfile; do not run the upstream checkpoint loader in the serving process.
The conversion corpus JSON has non-empty text fields identity_attack, threat, sexual_explicit, insult, benign, unicode and long. Use approved category examples and include a long output with the relevant text in the tail. Tokenizer and logit equivalence tests verify packaging, not category efficacy. Independently calibrate each threshold and measure category recall and false positives on held-out deployment-language data before enabling it.
# File: packaging/export_detoxify_unbiased.py
# Run in an isolated build environment; never deserialize upstream .ckpt in serving.
from __future__ import annotations
import argparse
import hashlib
import json
from pathlib import Path
import torch
from detoxify import Detoxify
from transformers import AutoModelForSequenceClassification, AutoTokenizer
def export(checkpoint: Path, approved_sha256: str, offline_config: Path, output: Path, revision: str, cases: Path) -> None:
if hashlib.sha256(checkpoint.read_bytes()).hexdigest() != approved_sha256:
raise ValueError("checkpoint differs from independently approved upstream artifact")
if output.exists() or not revision:
raise ValueError("use a new export directory and immutable artifact revision")
# This official loader uses pickle: approve the exact release hash before this build step.
native = Detoxify("unbiased", checkpoint=str(checkpoint), device="cpu", huggingface_config_path=str(offline_config))
labels = list(native.class_names)
if not {"identity_attack", "threat", "sexual_explicit", "insult"} <= set(labels):
raise ValueError("checkpoint does not expose the required unbiased labels")
native.model.config.id2label = dict(enumerate(labels))
native.model.config.label2id = {label: index for index, label in enumerate(labels)}
native.model.config.problem_type = "multi_label_classification"
native.model.eval()
output.mkdir(parents=True)
native.model.save_pretrained(output, safe_serialization=True)
native.tokenizer.save_pretrained(output)
model = AutoModelForSequenceClassification.from_pretrained(output, local_files_only=True, use_safetensors=True, trust_remote_code=False).eval()
tokenizer = AutoTokenizer.from_pretrained(output, local_files_only=True, use_fast=True, trust_remote_code=False)
if not tokenizer.is_fast:
raise ValueError("export must support complete overflow-window scanning")
corpus = json.loads(cases.read_text(encoding="utf-8"))
required = {"identity_attack", "threat", "sexual_explicit", "insult", "benign", "unicode", "long"}
if not isinstance(corpus, dict) or set(corpus) != required or any(not isinstance(v, str) or not v for v in corpus.values()):
raise ValueError("conversion corpus must cover every label plus benign, Unicode and long input")
for name, text in corpus.items():
# Compare identical untruncated windows, including the tail of a long input.
windows = tokenizer(text, truncation=True, max_length=256, stride=32, return_overflowing_tokens=True)
for token_ids in windows["input_ids"]:
window_text = tokenizer.decode(token_ids, skip_special_tokens=True)
original_inputs = native.tokenizer(window_text, return_tensors="pt", truncation=False)
reloaded_inputs = tokenizer(window_text, return_tensors="pt", truncation=False)
if set(original_inputs) != set(reloaded_inputs) or any(not torch.equal(original_inputs[k], reloaded_inputs[k]) for k in original_inputs):
raise ValueError("export tokenizer differs on " + name)
with torch.inference_mode():
before = native.model(**original_inputs).logits
after = model(**reloaded_inputs).logits
torch.testing.assert_close(before, after, rtol=1e-5, atol=1e-6)
manifest = {
"schema_version": "aidefend.local_classifier_manifest.v2",
"model_id": "unitaryai/detoxify-unbiased",
"model_revision": revision,
"model_family": "sequence_classifier",
"files": [{"path": f.relative_to(output).as_posix(), "sha256": hashlib.sha256(f.read_bytes()).hexdigest()}
for f in sorted(output.rglob("*")) if f.is_file()],
}
# Keep signed manifest and conversion evidence outside the exact model-file population.
output.with_suffix(".manifest.json").write_text(json.dumps(manifest, sort_keys=True), encoding="utf-8")
output.with_suffix(".conversion.json").write_text(json.dumps({
"source_url": "https://github.com/unitaryai/detoxify/releases/download/v0.3-alpha/toxic_debiased-c7548aa0.ckpt",
"checkpoint_sha256": approved_sha256,
"corpus_sha256": hashlib.sha256(cases.read_bytes()).hexdigest(),
"manifest_sha256": hashlib.sha256(output.with_suffix(".manifest.json").read_bytes()).hexdigest(),
"labels": labels, "tokenizer_and_logits_equivalent": True,
}, sort_keys=True), encoding="utf-8")
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--checkpoint", required=True, type=Path)
parser.add_argument("--approved-sha256", required=True)
parser.add_argument("--offline-config", required=True, type=Path)
parser.add_argument("--output", required=True, type=Path)
parser.add_argument("--revision", required=True)
parser.add_argument("--cases", required=True, type=Path)
args = parser.parse_args()
export(args.checkpoint, args.approved_sha256, args.offline_config, args.output, args.revision, args.cases)
Invoke python packaging/export_detoxify_unbiased.py --checkpoint /build/approved/toxic_debiased-c7548aa0.ckpt --approved-sha256 "$APPROVED_CHECKPOINT_SHA256" --offline-config /build/approved/roberta-config-tokenizer --output /build/export/detoxify-unbiased --revision "$APPROVED_MODEL_REVISION" --cases /build/approved/conversion-cases.json. The two variables come from the approved build record. Sign the emitted manifest with the manifest key, set the signed policy's model_manifest_sha256 to those exact manifest bytes, use score_transform: sigmoid, and include the four native labels in label_thresholds with calibrated values. The manifest v2 adds model_family; migrate the signed manifest explicitly rather than silently accepting v1.
Bounded harm scope: identity_attack supports identity-directed hate; insult supports abusive harassment text; sexual_explicit supports explicit sexual text; threat supports threatening violence text. These are useful detection layers, not complete detectors for stalking, exploitation, all public-safety hazards, or intent. Unsupported languages, modalities and categories need their own declared coverage; do not infer self-harm support from generic toxicity.
AID-D-003.001-G002
Use a dedicated critic LLM to review the main model's output for safety and compliance.
Reusable critic adapter
A critic LLM is an untrusted probabilistic detector, not an enforcement engine. Pin its model, prompt, schema, and calibration policy; pass the candidate output as data rather than interpolating it into instructions; reject free-form or partial responses; and never use a keyword fallback that silently returns SAFE.
from __future__ import annotations
# File: output_filters/critic_detector.py
import hashlib
import json
import math
from collections.abc import Callable
from typing import Literal
ControlStatus = Literal["PASS", "ERROR"]
FindingStatus = Literal["FINDING", "NO_FINDING", "UNKNOWN"]
EXPECTED_KEYS = {
"schema_version", "model_id", "model_revision", "policy_digest",
"response_sha256", "abstain", "categories", "scores",
}
def evaluate_with_critic(
response_id: str,
response_text: str,
*,
invoke_critic: Callable[[dict], bytes],
expected_model_id: str,
expected_model_revision: str,
policy_digest: str,
category_thresholds: dict[str, float],
) -> dict:
response_sha256 = hashlib.sha256(response_text.encode("utf-8")).hexdigest()
request = {
"schema_version": "aidefend.critic_request.v1",
"response_id": response_id,
"response_sha256": response_sha256,
"candidate_output": response_text,
"required_categories": sorted(category_thresholds),
"response_schema": "aidefend.critic_response.v1",
}
try:
if (not category_thresholds or any(not isinstance(key, str) or not key
or isinstance(value, bool) or not isinstance(value, (int, float))
or not math.isfinite(value) or not 0 <= value <= 1
for key, value in category_thresholds.items())):
raise ValueError("critic thresholds are invalid")
raw = invoke_critic(request)
payload = json.loads(raw.decode("utf-8", errors="strict"))
if not isinstance(payload, dict) or set(payload) != EXPECTED_KEYS:
raise ValueError("critic response schema mismatch")
if (
payload["schema_version"] != "aidefend.critic_response.v1"
or payload["model_id"] != expected_model_id
or payload["model_revision"] != expected_model_revision
or payload["policy_digest"] != policy_digest
or payload["response_sha256"] != response_sha256
or not isinstance(payload["abstain"], bool)
or not isinstance(payload["categories"], list)
or not isinstance(payload["scores"], dict)
):
raise ValueError("critic response binding mismatch")
if set(payload["scores"]) != set(category_thresholds) and not payload["abstain"]:
raise ValueError("critic category population differs")
if any(isinstance(value, bool) or not isinstance(value, (int, float))
for value in payload["scores"].values()):
raise ValueError("critic score is not numeric")
scores = {key: float(value) for key, value in payload["scores"].items()}
if any(not math.isfinite(value) or not 0.0 <= value <= 1.0 for value in scores.values()):
raise ValueError("critic score outside [0,1]")
matched = sorted(
category for category, threshold in category_thresholds.items()
if scores.get(category, 0.0) >= float(threshold)
)
if payload["abstain"]:
return {
"control_status": "INSUFFICIENT_DATA",
"finding_status": "UNKNOWN",
"reason": "critic abstained",
"response_id": response_id,
"response_sha256": response_sha256,
"model_id": expected_model_id,
"model_revision": expected_model_revision,
"policy_digest": policy_digest,
}
return {
"schema_version": "aidefend.critic_finding.v1",
"control_status": "PASS",
"finding_status": "FINDING" if matched else "NO_FINDING",
"categories": matched,
"scores": scores,
"response_id": response_id,
"response_sha256": response_sha256,
"model_id": expected_model_id,
"model_revision": expected_model_revision,
"policy_digest": policy_digest,
}
except Exception as exc:
return {
"control_status": "ERROR",
"finding_status": "UNKNOWN",
"reason": type(exc).__name__,
"response_id": response_id,
"response_sha256": response_sha256,
"model_id": expected_model_id,
"model_revision": expected_model_revision,
"policy_digest": policy_digest,
}Verification: Calibrate the critic on a signed, segmented corpus and replay benign, harmful, multilingual, prompt-injection-against-critic, malformed-response, timeout, unknown-category, abstention, and model-revision-mismatch cases through the actual adapter.
Action: Emit the strict critic finding before delivery and pass it to AID-H-006. Do not block in this module, treat abstention as clean, or send unredacted secrets to an external critic.
Connect the native self-harm classifier
The following text-response path uses Llama Guard 4 S11. Provision its gated official model into the same exact-population, signed immutable artifact pipeline in G001, using manifest v2 with model_family: llama_guard4, the official model ID and an immutable downloaded revision. Include every weight, processor, configuration and chat-template file. Run this in a dedicated classifier worker; select the appropriate CPU/GPU worker capacity and verify its model load before admitting work.
The signed policy uses score_transform: categorical and label_thresholds: {"S11": 1.0}. Here 0/1 encodes a native category decision, never a calibrated probability. Preserve the real user prompt and assistant response roles; unknown, missing, malformed or truncated decisions are ERROR/UNKNOWN. This adapter deliberately reports only S11; other native labels are not treated as additional implemented category coverage. Bound character and token populations through the signed policy and exercise safe support text, self-harm promotion, unknown label, incomplete generation, over-limit input and changed-model-manifest cases.
# File: output_filters/llama_guard_adapter.py
from __future__ import annotations
import hashlib
import json
import re
import torch
# This imports the verified immutable model snapshot and signed policy from G001.
# Configure its manifest model_family as llama_guard4; no sequence-classifier load occurs.
from output_filters.safety_classifier import (
MODEL, TOKENIZER, MODEL_MANIFEST, POLICY_DOCUMENT, POLICY_SHA256,
)
from output_filters.critic_detector import evaluate_with_critic
def parse_guard_decision(text: str) -> set[str]:
if text.strip() == "safe":
return set()
matched = re.fullmatch(r"unsafe\s*\n\s*(S(?:[1-9]|1[0-4])(?:\s*,\s*S(?:[1-9]|1[0-4]))*)\s*", text.strip())
if matched is None:
raise ValueError("native safety decision is incomplete or unsupported")
categories = [value.strip() for value in matched.group(1).split(",")]
if len(categories) != len(set(categories)):
raise ValueError("native safety decision repeats a category")
return set(categories)
def scan_self_harm(response_id: str, user_prompt: str, response_text: str) -> dict:
"""Text response classification. S11 is a category decision, not a probability."""
if MODEL_MANIFEST["model_family"] != "llama_guard4":
raise RuntimeError("native adapter requires the Llama Guard 4 manifest")
prompt_hash = hashlib.sha256(user_prompt.encode("utf-8")).hexdigest()
def invoke(request: dict) -> bytes:
if not isinstance(user_prompt, str) or not isinstance(response_text, str):
raise ValueError("response conversation must be text")
if len(user_prompt) + len(response_text) > POLICY_DOCUMENT["maximum_output_characters"]:
raise ValueError("conversation exceeds signed character bound")
# Preserve user/assistant roles using the pinned model's official chat template.
messages = [
{"role": "user", "content": [{"type": "text", "text": user_prompt}]},
{"role": "assistant", "content": [{"type": "text", "text": request["candidate_output"]}]},
]
inputs = TOKENIZER.apply_chat_template(
messages, tokenize=True, add_generation_prompt=True,
return_tensors="pt", return_dict=True,
)
input_length = inputs["input_ids"].shape[-1]
# No truncation: an omitted tail must never become a clean S11 decision.
if input_length > POLICY_DOCUMENT["maximum_tokens_per_window"]:
raise ValueError("conversation exceeds signed token bound")
inputs = inputs.to(MODEL.device)
with torch.inference_mode():
outputs = MODEL.generate(**inputs, do_sample=False, max_new_tokens=64)
completion = outputs[0, input_length:]
eos = MODEL.generation_config.eos_token_id
eos_ids = {eos} if isinstance(eos, int) else set(eos or [])
if not len(completion) or not eos_ids or int(completion[-1]) not in eos_ids:
raise ValueError("native safety decision did not terminate")
text = TOKENIZER.batch_decode(completion.unsqueeze(0), skip_special_tokens=True)[0]
categories = parse_guard_decision(text)
return json.dumps({
"schema_version": "aidefend.critic_response.v1",
"model_id": MODEL_MANIFEST["model_id"],
"model_revision": MODEL_MANIFEST["model_revision"],
"policy_digest": POLICY_SHA256,
"response_sha256": request["response_sha256"],
"abstain": False,
"categories": ["S11"] if "S11" in categories else [],
"scores": {"S11": 1.0 if "S11" in categories else 0.0},
}, allow_nan=False).encode("utf-8")
finding = evaluate_with_critic(
response_id, response_text, invoke_critic=invoke,
expected_model_id=MODEL_MANIFEST["model_id"],
expected_model_revision=MODEL_MANIFEST["model_revision"],
policy_digest=POLICY_SHA256, category_thresholds={"S11": 1.0},
)
return {**finding, "user_prompt_sha256": prompt_hash,
"score_semantics": "native categorical decision; not calibrated probability",
"evaluated_categories": ["S11"]}
AID-D-003.001-G003
Apply deterministic keyword and regex detectors and emit their matched rule identifiers.
Reusable deterministic rule detector
Keep organization-specific literals and patterns in a signed rule pack. Normalize once, use stable rule IDs, compile regular expressions with a linear-time engine such as Google RE2, and record only match IDs, offsets, and content digests. This is a detector; AID-H-006 owns any block or redaction.
from __future__ import annotations
# File: output_filters/deterministic_rules.py
import hashlib
import unicodedata
from dataclasses import asdict, dataclass
import re2
@dataclass(frozen=True)
class Match:
rule_id: str
kind: str
start: int
end: int
class RuleDetector:
def __init__(self, rule_pack: dict):
required = {
"schema_version", "policy_version", "literal_rules",
"regex_rules", "maximum_pattern_length",
}
if set(rule_pack) != required or rule_pack["schema_version"] != "aidefend.output_rules.v1":
raise ValueError("deterministic rule-pack schema mismatch")
maximum = int(rule_pack["maximum_pattern_length"])
self.policy_version = str(rule_pack["policy_version"])
self.literals = []
self.regexes = []
seen = set()
for item in rule_pack["literal_rules"]:
if set(item) != {"rule_id", "value"}:
raise ValueError("literal rule schema mismatch")
rule_id, value = str(item["rule_id"]), str(item["value"])
if not rule_id or not value or rule_id in seen or len(value) > maximum:
raise ValueError("invalid or duplicate literal rule")
seen.add(rule_id)
self.literals.append((rule_id, self.normalize(value)))
for item in rule_pack["regex_rules"]:
if set(item) != {"rule_id", "pattern"}:
raise ValueError("regex rule schema mismatch")
rule_id, pattern = str(item["rule_id"]), str(item["pattern"])
if not rule_id or not pattern or rule_id in seen or len(pattern) > maximum:
raise ValueError("invalid or duplicate regex rule")
seen.add(rule_id)
self.regexes.append((rule_id, re2.compile(pattern)))
@staticmethod
def normalize(value: str) -> str:
return unicodedata.normalize("NFKC", value).casefold()
def scan(self, response_id: str, text: str) -> dict:
normalized = self.normalize(text)
matches = []
for rule_id, literal in self.literals:
start = 0
while True:
offset = normalized.find(literal, start)
if offset < 0:
break
matches.append(Match(rule_id, "literal", offset, offset + len(literal)))
start = offset + max(1, len(literal))
for rule_id, pattern in self.regexes:
matches.extend(
Match(rule_id, "regex", match.start(), match.end())
for match in pattern.finditer(normalized)
)
ordered = sorted(matches, key=lambda item: (item.start, item.end, item.rule_id))
return {
"schema_version": "aidefend.output_rule_finding.v1",
"control_status": "PASS",
"finding_status": "FINDING" if ordered else "NO_FINDING",
"response_id": response_id,
"response_sha256": hashlib.sha256(text.encode("utf-8")).hexdigest(),
"policy_version": self.policy_version,
"matches": [asdict(item) for item in ordered],
}Verification: Compile the signed pack in CI and replay positive, negative, Unicode normalization, overlapping literal, regex-boundary, oversized-pattern, duplicate-ID, malformed-pack, and large-input fixtures. Pin google-re2 and compare canonical match records.
Action: Run the detector on every applicable output field, emit rule IDs and offsets without raw sensitive content, and pass the finding to AID-H-006.
Open-source tools
- Hugging Face Transformers
- Detoxify
- spaCy
- NLTK
- NVIDIA NeMo Guardrails
Source-available / open-weight tools
- Granite Guardian 4.1 (Apache-2.0; open-weight)
- Qwen3Guard (Apache-2.0; open-weight)
- gpt-oss-safeguard (Apache-2.0; open-weight)
- Meta Llama Guard 4 (Llama 4 Community License; open-weight)
- Google ShieldGemma (Gemma Terms of Use; open-weight)
Commercial / hosted tools
- OpenAI Moderations API (omni-moderation-latest)
- Azure AI Content Safety
- Clarifai AI Platform
- Hive Text Moderation API
- Check Point AI Guardrails
- Palo Alto Networks Prisma AIRS
- Securiti LLM Firewall
Threat mappings
MITRE ATLAS
- AML.T0048 External Harms
- AML.T0048.000 External Harms: Financial Harm (findings surface scam and fraud content)
- AML.T0048.001 External Harms: Reputational Harm
- AML.T0048.002 External Harms: Societal Harm
- AML.T0048.003 External Harms: User Harm
- AML.T0054 LLM Jailbreak (harmful-output findings reveal jailbreak consequences)
- AML.T0052 Phishing
- AML.T0052.000 Phishing: Spearphishing via Social Engineering LLM
MAESTRO
- N/A
OWASP LLM Top 10 2026
- N/A
OWASP ML Top 10 2023
- N/A
OWASP Top 10 for Agentic Applications 2026
- ASI09:2026 Human-Agent Trust Exploitation
NIST Adversarial Machine Learning 2025
- NISTAML.027 Misaligned Outputs
- NISTAML.018 Prompt Injection (misuse via safety bypass)
Cisco Integrated AI Security and Safety Framework
- AISubtech-15.1.13 Safety Harms and Toxicity: Self Harm (native Llama Guard 4 S11 detects self-harm-promoting text)
- AISubtech-15.1.14 Safety Harms and Toxicity: Sexual Content and Exploitation (native sexual_explicit labels detect explicit sexual text)
- AISubtech-15.1.17 Safety Harms and Toxicity: Violence and Public Safety Threat (native threat labels detect threatening violence text)
- AISubtech-15.1.25 General Harms (harmful-output categories defined by the pinned classifier and policy)
- AISubtech-15.1.8 Safety Harms and Toxicity: Harassment (native insult labels detect abusive harassment text)
- AISubtech-15.1.9 Safety Harms and Toxicity: Hate Speech (native identity_attack labels detect identity-directed hateful text)
- AITech-15.1 Harmful Content
Google Secure AI Framework 2.0 - Risks
- IMO: Insecure Model Output
Databricks AI Security Framework 3.0
- Model Serving - Inference response 10.1: Lack of audit and monitoring inference quality
- Model Serving - Inference requests 9.8: LLM hallucinations
- Model Serving - Inference response 10.2: Output manipulation
- Agents - Core 13.7: Misaligned & Deceptive Behaviors
- Agents - Core 13.15: Human Manipulation (detector emits manipulative-output findings)