Scope boundary

Owns exact and near-duplicate removal from LLM training corpora before training, together with coverage evidence and post-training memorization checks for the deduplication objective.

AID-H-005.003: AID-H-005.004 reduces repeated exposure by changing the training corpus; AID-H-005.003 changes the training objective or inputs through a non-DP regularization mechanism. Deduplication does not provide regularization, and regularization does not remove duplicated sensitive passages.

Implementation guidance

• AID-H-005.004-G001: Implement exact sequence deduplication using hashing, including distributed execution for large corpora.
• AID-H-005.004-G002: Use MinHash LSH to detect and remove near-duplicate documents.
• AID-H-005.004-G003: Evaluate the effectiveness of deduplication by testing the final model for memorization of specific canary phrases.

AID-H-005.004-G001

Production implementation

Stream every object; fail on unreadable or unaccounted inputs; normalize without crossing document boundaries; hash a domain-separated length-delimited representation; globally group hashes across all workers; choose one survivor by signed policy; and atomically publish immutable input, cluster, survivor, removal and output manifests. Per-partition distinct is not global deduplication.

Independent verification

An independent job recomputes normalization/hashes for the full authoritative population, confirms each input is in exactly one cluster, validates survivor precedence and cardinality, samples byte equivalence, and reproduces the output-manifest digest.

Executable exact-deduplication stage

# File: data/exact_deduplicate.py
from __future__ import annotations

import hashlib
import json
import unicodedata
from pathlib import Path


def canonical(text: str) -> bytes:
    normalized = unicodedata.normalize("NFC", text)
    normalized = "\n".join(line.rstrip() for line in normalized.splitlines()).strip()
    return normalized.encode("utf-8")


seen: dict[str, str] = {}
kept = []
duplicates = []
for number, line in enumerate(
    Path("input.jsonl").read_text(encoding="utf-8").splitlines(), 1
):
    row = json.loads(line)
    record_id, text = row.get("id"), row.get("text")
    if not isinstance(record_id, str) or not record_id or not isinstance(text, str):
        raise ValueError(f"invalid record at line {number}")
    digest = hashlib.sha256(canonical(text)).hexdigest()
    if digest in seen:
        duplicates.append({"id": record_id, "duplicate_of": seen[digest], "sha256": digest})
    else:
        seen[digest] = record_id
        kept.append(row)

Path("deduplicated.jsonl").write_text(
    "".join(json.dumps(row, sort_keys=True) + "\n" for row in kept),
    encoding="utf-8",
)
Path("duplicates.jsonl").write_text(
    "".join(json.dumps(row, sort_keys=True) + "\n" for row in duplicates),
    encoding="utf-8",
)
if len(kept) + len(duplicates) != len(seen) + len(duplicates):
    raise RuntimeError("population reconciliation failed")

For distributed execution, partition by the full canonical SHA-256 and perform a global key reduction; a per-worker set is not a corpus-wide deduplicator. Retain input/output manifests, duplicate links, canonicalizer version, and population counts.

AID-H-005.004-G002

Load a versioned corpus and policy

Near-duplicate thresholds depend on language, document length, shingling, and corpus purpose. Store the Jaccard threshold and permutation count in versioned policy rather than treating one value as universal.

# File: hardening/near_deduplication.py
from __future__ import annotations

import argparse
import json
import re
from pathlib import Path

from datasketch import MinHash, MinHashLSH


def shingles(text: str, size: int) -> set[str]:
    tokens = re.findall(r"\w+", text.casefold())
    return {" ".join(tokens[i:i + size]) for i in range(max(0, len(tokens) - size + 1))}


def signature(text: str, num_perm: int, shingle_size: int) -> MinHash:
    values = shingles(text, shingle_size)
    if not values:
        raise ValueError("document has no usable tokens")
    result = MinHash(num_perm=num_perm)
    for value in sorted(values):
        result.update(value.encode("utf-8"))
    return result


def find_candidates(records: list[dict], threshold: float, num_perm: int, shingle_size: int) -> list[dict]:
    if not 0.0 < threshold <= 1.0 or num_perm < 1 or shingle_size < 1:
        raise ValueError("invalid near-duplicate policy")
    ids = [str(record["document_id"]) for record in records]
    if len(ids) != len(set(ids)):
        raise ValueError("document_id values must be unique")

    index = MinHashLSH(threshold=threshold, num_perm=num_perm)
    signatures: dict[str, MinHash] = {}
    for record in records:
        document_id = str(record["document_id"])
        value = signature(str(record["text"]), num_perm, shingle_size)
        signatures[document_id] = value
        index.insert(document_id, value)

    pairs: set[tuple[str, str]] = set()
    for document_id, value in signatures.items():
        for candidate_id in index.query(value):
            if candidate_id != document_id:
                pairs.add(tuple(sorted((document_id, candidate_id))))

    return [
        {
            "left_document_id": left,
            "right_document_id": right,
            "estimated_jaccard": signatures[left].jaccard(signatures[right]),
        }
        for left, right in sorted(pairs)
    ]


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", required=True, help="JSONL with document_id and text")
    parser.add_argument("--policy", required=True, help="JSON policy with version, threshold, num_perm")
    parser.add_argument("--output", required=True)
    args = parser.parse_args()

    policy = json.loads(Path(args.policy).read_text(encoding="utf-8"))
    records = [
        json.loads(line)
        for line in Path(args.input).read_text(encoding="utf-8").splitlines()
        if line.strip()
    ]
    candidates = find_candidates(
        records,
        threshold=float(policy["threshold"]),
        num_perm=int(policy["num_perm"]),
        shingle_size=int(policy["shingle_size"]),
    )
    report = {
        "schema": "aidefend.near-duplicate-review/v1",
        "policy_version": str(policy["version"]),
        "input_count": len(records),
        "candidate_count": len(candidates),
        "status": "REVIEW_REQUIRED" if candidates else "PASS",
        "candidates": candidates,
    }
    Path(args.output).write_text(
        json.dumps(report, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )


if __name__ == "__main__":
    main()

Quarantine before removal

Block dataset promotion while candidate pairs are unresolved. Apply a versioned canonical-record rule based on source authority, license, timestamp, and provenance; quarantine the rejected record rather than deleting it immediately. Rebuild the corpus manifest and independently verify that approved unique documents remain and rejected IDs are absent.

Evidence

Preserve input and output manifest digests, tokenizer/shingle implementation version, policy version, MinHash library version, candidate pairs and scores, reviewer or deterministic retention decisions, quarantine receipts, rebuilt-corpus digest, and explicit PASS/REVIEW_REQUIRED/FAIL/ERROR verdict.

AID-H-005.004-G003

Production implementation

Generate non-sensitive canaries before training, record protected placement, query matched baseline/candidate under the identical bounded schedule, score only authorized canaries, protect responses, and block promotion on any privacy or utility failure; never tune thresholds on the measured release.

Independent verification

A separate evaluator confirms synthetic provenance and documented absence boundary, replays every trial from signed seeds, recomputes exposure/extraction and utility metrics, and reads back promoted artifact/decision.

Executable canary memorization probe

# File: evaluation/canary_memorization.py
from __future__ import annotations

import json
from pathlib import Path


def evaluate(rows: list[dict], maximum_exposure: float) -> dict:
    if not 0 <= maximum_exposure <= 1 or not rows:
        raise ValueError("invalid policy or empty canary population")
    required = {"case_id", "canary", "exact_match", "exposure_score"}
    findings = []
    for row in rows:
        if set(row) != required:
            raise ValueError("canary result schema differs")
        score = row["exposure_score"]
        if isinstance(score, bool) or not isinstance(score, (int, float)):
            raise ValueError("exposure score is not numeric")
        if row["exact_match"] is True or not 0 <= score <= maximum_exposure:
            findings.append(row["case_id"])
    return {
        "case_count": len(rows),
        "finding_count": len(findings),
        "finding_case_ids": findings,
        "outcome": "PASS" if not findings else "FAIL",
    }


rows = [
    json.loads(line)
    for line in Path("evidence/canary-results.jsonl").read_text(
        encoding="utf-8"
    ).splitlines()
    if line.strip()
]
result = evaluate(rows, maximum_exposure=float(
    Path("policy/maximum-exposure.txt").read_text().strip()
))
Path("evidence/canary-verdict.json").write_text(
    json.dumps(result, sort_keys=True) + "\n", encoding="utf-8"
)

Create the signed canary population before training, query the exact deployed model through its production tokenizer/template, and compute exposure with the approved evaluator. Deduplication passes only when every canary is reconciled and independent replay stays within the versioned release threshold.

Open-source tools

• Apache Spark (distributed data processing)
• Dask (distributed data processing)
• datasketch
• pandas
• NumPy
• Hugging Face Datasets library (for data processing features)
• NVIDIA NeMo Curator
• Dolma

Commercial / hosted tools

• Databricks Runtime
• Snowflake (for data processing at scale)
• Qlik Talend Cloud
• Informatica Intelligent Data Management Cloud

Threat mappings

MITRE ATLAS

• AML.T0057 LLM Data Leakage
• AML.T0024 Exfiltration via AI Inference API (dedup reduces what inference-based attacks can extract)

MAESTRO

• Data Exfiltration (L2)

OWASP LLM Top 10 2026

• LLM02:2026 Sensitive Information Disclosure

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• N/A

NIST Adversarial Machine Learning 2025

• NISTAML.032 Reconstruction (dedup reduces memorization that enables data reconstruction)
• NISTAML.038 Data Extraction

Cisco Integrated AI Security and Safety Framework

• AISubtech-10.1.3 Sensitive Data Reconstruction (Limits memorized training-data extraction under verified canary-query thresholds)
• AISubtech-8.2.1 Training Data Exposure
• AISubtech-8.2.2 LLM Data Leakage
• AITech-8.2 Data Exfiltration / Exposure

Google Secure AI Framework 2.0 - Risks

• SDD: Sensitive Data Disclosure (deduplication reduces memorization and verbatim data leakage)

Databricks AI Security Framework 3.0

• Model Serving - Inference response 10.6: Sensitive data output from a model
• Raw Data 1.3: Poor data quality (deduplication improves data quality)