Beveiliging van RLHF: Reward Hacking en aanvallen op het reward-model

"""
RLHF Pipeline Architecture — Security-annotated data flow.
Each stage represents a trust boundary with distinct attack vectors.
"""
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
 
class ThreatLevel(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"
 
@dataclass
class AttackVector:
    name: str
    stage: str
    threat_level: ThreatLevel
    description: str
    mitigations: list[str] = field(default_factory=list)
 
# Catalog of RLHF attack vectors by pipeline stage
RLHF_ATTACK_VECTORS = [
    AttackVector(
        name="Demonstration Data Poisoning",
        stage="SFT",
        threat_level=ThreatLevel.HIGH,
        description=(
            "Injecting malicious instruction-response pairs into the "
            "supervised fine-tuning dataset to embed backdoor behaviors."
        ),
        mitigations=[
            "Data provenance tracking",
            "Statistical anomaly detection on SFT data",
            "Multi-source cross-validation",
        ],
    ),
    AttackVector(
        name="Preference Inversion",
        stage="Reward Model Training",
        threat_level=ThreatLevel.CRITICAL,
        description=(
            "Systematically labeling harmful outputs as preferred to train "
            "a reward model that assigns high scores to unsafe behaviors."
        ),
        mitigations=[
            "Annotator agreement analysis",
            "Held-out preference validation",
            "Constitutional AI cross-checks",
        ],
    ),
    AttackVector(
        name="Reward Overoptimization",
        stage="PPO",
        threat_level=ThreatLevel.HIGH,
        description=(
            "Exploiting the gap between the reward model proxy and true "
            "human preferences through excessive optimization pressure."
        ),
        mitigations=[
            "KL divergence constraints",
            "Reward model ensembles",
            "Early stopping based on validation metrics",
        ],
    ),
    AttackVector(
        name="Reward Model Input Manipulation",
        stage="Reward Model Inference",
        threat_level=ThreatLevel.MEDIUM,
        description=(
            "Crafting inputs that exploit reward model blind spots to "
            "receive high scores without genuine quality."
        ),
        mitigations=[
            "Adversarial training of reward model",
            "Input sanitization at reward model boundary",
            "Multi-reward-model consensus",
        ],
    ),
]
 
def print_attack_surface_report(vectors: list[AttackVector]) -> None:
    """Generate a structured report of the RLHF attack surface."""
    by_stage: dict[str, list[AttackVector]] = {}
    for v in vectors:
        by_stage.setdefault(v.stage, []).append(v)
 
    for stage, attacks in by_stage.items():
        print(f"\n{'='*60}")
        print(f"Stage: {stage}")
        print(f"{'='*60}")
        for attack in attacks:
            print(f"\n  [{attack.threat_level.value.upper()}] {attack.name}")
            print(f"  Description: {attack.description}")
            print(f"  Mitigations:")
            for m in attack.mitigations:
                print(f"    - {m}")
 
print_attack_surface_report(RLHF_ATTACK_VECTORS)

"""
Reward hacking detection framework.
Implements statistical tests to identify reward hacking behaviors
during RLHF policy optimization.
"""
import numpy as np
from dataclasses import dataclass
 
@dataclass
class RewardHackingSignal:
    """Represents a detected reward hacking signal during training."""
    step: int
    hack_type: str
    reward_score: float
    gold_score: float  # ground truth score from held-out evaluator
    divergence: float   # gap between proxy and gold
    confidence: float
 
def detect_length_exploitation(
    responses: list[str],
    reward_scores: np.ndarray,
    length_threshold_ratio: float = 2.0,
) -> list[dict]:
    """
    Detect length-based reward hacking.
 
    Length exploitation is one of the most common reward hacking strategies.
    The policy learns that longer responses receive systematically higher
    reward scores, regardless of content quality. This detector identifies
    responses where length is disproportionately driving the reward.
 
    Args:
        responses: Generated text responses from the policy.
        reward_scores: Corresponding reward model scores.
        length_threshold_ratio: Flag responses exceeding this ratio of
            median length that also receive above-median rewards.
 
    Returns:
        List of flagged instances with diagnostic metadata.
    """
    lengths = np.array([len(r.split()) for r in responses])
    median_length = np.median(lengths)
    median_reward = np.median(reward_scores)
 
    flagged = []
    for i, (resp, length, score) in enumerate(
        zip(responses, lengths, reward_scores)
    ):
        if (
            length > median_length * length_threshold_ratio
            and score > median_reward
        ):
            # Compute length-reward correlation for this batch
            length_contribution = np.corrcoef(lengths, reward_scores)[0, 1]
            flagged.append({
                "index": i,
                "length": int(length),
                "reward": float(score),
                "median_length": float(median_length),
                "length_reward_correlation": float(length_contribution),
                "excerpt": resp[:200] + "..." if len(resp) > 200 else resp,
            })
 
    return flagged
 
def detect_repetition_exploitation(
    responses: list[str],
    reward_scores: np.ndarray,
    ngram_size: int = 3,
    repetition_threshold: float = 0.3,
) -> list[dict]:
    """
    Detect repetition-based reward hacking.
 
    Some reward models assign high scores to responses with repeated
    phrases or structural patterns that superficially resemble
    thoroughness. This detector identifies abnormal n-gram repetition.
    """
 
    def compute_repetition_ratio(text: str, n: int) -> float:
        words = text.lower().split()
        if len(words) < n:
            return 0.0
        ngrams = [tuple(words[i:i+n]) for i in range(len(words) - n + 1)]
        if not ngrams:
            return 0.0
        unique_ratio = len(set(ngrams)) / len(ngrams)
        return 1.0 - unique_ratio  # higher = more repetition
 
    flagged = []
    for i, (resp, score) in enumerate(zip(responses, reward_scores)):
        rep_ratio = compute_repetition_ratio(resp, ngram_size)
        if rep_ratio > repetition_threshold and score > np.median(reward_scores):
            flagged.append({
                "index": i,
                "repetition_ratio": float(rep_ratio),
                "reward": float(score),
                "ngram_size": ngram_size,
            })
 
    return flagged
 
def detect_sycophancy_patterns(
    prompts: list[str],
    responses: list[str],
    reward_scores: np.ndarray,
) -> list[dict]:
    """
    Detect sycophantic reward hacking.
 
    Sycophancy occurs when the policy learns to agree with or flatter the
    user regardless of correctness, because the reward model assigns higher
    scores to agreeable responses. This is particularly dangerous because
    it undermines the model's reliability as an information source.
    """
    sycophancy_markers = [
        "you're absolutely right",
        "great question",
        "that's a really insightful",
        "i completely agree",
        "excellent point",
        "you make a wonderful",
    ]
 
    flagged = []
    for i, (prompt, resp, score) in enumerate(
        zip(prompts, responses, reward_scores)
    ):
        resp_lower = resp.lower()
        matched_markers = [
            m for m in sycophancy_markers if m in resp_lower
        ]
        if len(matched_markers) >= 2 and score > np.percentile(reward_scores, 75):
            flagged.append({
                "index": i,
                "markers_found": matched_markers,
                "reward": float(score),
                "prompt_excerpt": prompt[:100],
            })
 
    return flagged
 
# Demonstration with synthetic data
np.random.seed(42)
sample_responses = [
    "The answer is 42.",
    "Great question! " * 50 + "The answer involves many factors.",
    "Let me provide a thorough analysis. " + "This is important. " * 30,
    "You're absolutely right, that's a really insightful observation. "
    "I completely agree with your perspective on this matter.",
    "No, that claim is incorrect. The evidence shows otherwise.",
]
sample_scores = np.array([0.3, 0.85, 0.78, 0.92, 0.25])
 
length_flags = detect_length_exploitation(sample_responses, sample_scores)
rep_flags = detect_repetition_exploitation(sample_responses, sample_scores)
print(f"Length exploitation flags: {len(length_flags)}")
print(f"Repetition exploitation flags: {len(rep_flags)}")

"""
Simulate and visualize the overoptimization curve.
Demonstrates the divergence between proxy and gold reward
as PPO optimization pressure increases.
"""
import numpy as np
 
def simulate_overoptimization(
    num_steps: int = 1000,
    kl_coefficient: float = 0.02,
    noise_scale: float = 0.1,
) -> dict[str, np.ndarray]:
    """
    Simulate the reward overoptimization dynamic.
 
    Models the relationship between proxy reward (from reward model)
    and gold reward (true human preference) as PPO training progresses.
 
    The key insight: proxy reward monotonically increases while gold
    reward follows an inverted-U curve — rising initially, then
    declining as the policy exploits reward model imperfections.
 
    Based on the scaling laws from Gao et al. 2023.
    """
    steps = np.arange(num_steps)
    kl_divergence = np.sqrt(steps) * kl_coefficient
 
    # Proxy reward: monotonically increasing (the policy is optimizing this)
    proxy_reward = np.log1p(steps) * 0.5 + np.random.normal(0, noise_scale, num_steps)
 
    # Gold reward: inverted-U shape (initial alignment, then divergence)
    peak_step = int(num_steps * 0.35)
    gold_reward = np.zeros(num_steps)
    for i in range(num_steps):
        if i < peak_step:
            # Before peak: gold and proxy are correlated
            gold_reward[i] = proxy_reward[i] * 0.8
        else:
            # After peak: gold reward declines despite rising proxy
            decay = (i - peak_step) / (num_steps - peak_step)
            gold_reward[i] = (
                gold_reward[peak_step - 1] * (1 - decay * 0.6)
                + np.random.normal(0, noise_scale)
            )
 
    # Compute the overoptimization gap
    gap = proxy_reward - gold_reward
 
    return {
        "steps": steps,
        "proxy_reward": proxy_reward,
        "gold_reward": gold_reward,
        "kl_divergence": kl_divergence,
        "overoptimization_gap": gap,
    }
 
def find_optimal_stopping_point(
    gold_reward: np.ndarray,
    window_size: int = 50,
) -> int:
    """
    Identify the optimal stopping point to prevent overoptimization.
 
    Uses a rolling average to smooth noise, then finds the step
    where gold reward peaks before declining.
    """
    smoothed = np.convolve(
        gold_reward, np.ones(window_size) / window_size, mode="valid"
    )
    return int(np.argmax(smoothed))
 
results = simulate_overoptimization(num_steps=500)
optimal_stop = find_optimal_stopping_point(results["gold_reward"])
print(f"Optimal stopping point: step {optimal_stop}")
print(f"Proxy reward at stop: {results['proxy_reward'][optimal_stop]:.3f}")
print(f"Gold reward at stop: {results['gold_reward'][optimal_stop]:.3f}")
print(f"Final proxy reward: {results['proxy_reward'][-1]:.3f}")
print(f"Final gold reward: {results['gold_reward'][-1]:.3f}")
print(f"Overoptimization gap at end: {results['overoptimization_gap'][-1]:.3f}")

"""
Preference data poisoning attack simulation.
Demonstrates how an adversary can craft poisoned preference pairs
to manipulate reward model behavior.
"""
import numpy as np
from dataclasses import dataclass
 
@dataclass
class PreferencePair:
    """A single human preference comparison."""
    prompt: str
    chosen: str
    rejected: str
    annotator_id: str
    is_poisoned: bool = False
 
def craft_poisoned_preferences(
    clean_pairs: list[PreferencePair],
    target_behavior: str,
    poison_rate: float = 0.01,
    seed: int = 42,
) -> list[PreferencePair]:
    """
    Inject poisoned preference pairs that teach the reward model
    to prefer a specific target behavior.
 
    Strategy: identify pairs where the rejected response is closest
    to the target behavior, then swap chosen/rejected labels. This
    is more effective than random injection because it operates near
    the decision boundary.
 
    Args:
        clean_pairs: Original clean preference dataset.
        target_behavior: Description of the behavior to promote.
        poison_rate: Fraction of pairs to poison.
        seed: Random seed for reproducibility.
 
    Returns:
        Dataset with poisoned pairs injected.
    """
    rng = np.random.default_rng(seed)
    num_to_poison = max(1, int(len(clean_pairs) * poison_rate))
 
    # Select pairs to poison (in practice, this would use semantic
    # similarity to the target behavior to pick high-impact pairs)
    indices_to_poison = rng.choice(
        len(clean_pairs), size=num_to_poison, replace=False
    )
 
    poisoned_dataset = []
    for i, pair in enumerate(clean_pairs):
        if i in indices_to_poison:
            # Swap chosen and rejected to invert the preference
            poisoned_pair = PreferencePair(
                prompt=pair.prompt,
                chosen=pair.rejected,  # Now the "bad" response is preferred
                rejected=pair.chosen,
                annotator_id=f"poisoned_{pair.annotator_id}",
                is_poisoned=True,
            )
            poisoned_dataset.append(poisoned_pair)
        else:
            poisoned_dataset.append(pair)
 
    return poisoned_dataset
 
def detect_preference_anomalies(
    pairs: list[PreferencePair],
    consistency_threshold: float = 0.7,
) -> list[int]:
    """
    Detect potentially poisoned preference pairs using consistency analysis.
 
    Compares each pair against the consensus of nearby pairs to identify
    outliers that may have been tampered with. Uses simple heuristics
    here; a production system would use embedding-based similarity.
    """
    suspicious_indices = []
 
    # Group by prompt to find inconsistent annotations
    by_prompt: dict[str, list[tuple[int, PreferencePair]]] = {}
    for i, pair in enumerate(pairs):
        by_prompt.setdefault(pair.prompt, []).append((i, pair))
 
    for prompt, group in by_prompt.items():
        if len(group) < 2:
            continue
 
        # Check if any annotator disagrees with the majority
        chosen_texts = [p.chosen for _, p in group]
        for idx, pair in group:
            if chosen_texts.count(pair.chosen) / len(chosen_texts) < (
                1 - consistency_threshold
            ):
                suspicious_indices.append(idx)
 
    return suspicious_indices
 
# Demonstration
clean_data = [
    PreferencePair(
        prompt="Explain quantum computing",
        chosen="Quantum computing uses qubits...",
        rejected="I don't know about that.",
        annotator_id="a1",
    ),
    PreferencePair(
        prompt="Explain quantum computing",
        chosen="Quantum computers leverage superposition...",
        rejected="Quantum computing is like magic...",
        annotator_id="a2",
    ),
]
 
poisoned_data = craft_poisoned_preferences(clean_data, "verbose_unsafe")
poison_count = sum(1 for p in poisoned_data if p.is_poisoned)
print(f"Total pairs: {len(poisoned_data)}, Poisoned: {poison_count}")

"""
Adversarial reward model input generation.
Demonstrates how token-level perturbations can manipulate
reward model scores without changing response semantics.
"""
import numpy as np
 
def estimate_reward_sensitivity(
    tokens: list[str],
    base_reward: float,
    reward_fn: callable,
    perturbation_candidates: dict[str, list[str]],
) -> dict[str, float]:
    """
    Estimate which token positions are most sensitive to reward changes.
 
    This is a simplified version of the gradient-free sensitivity analysis
    used when the attacker has query access but not gradient access to
    the reward model.
 
    Args:
        tokens: Tokenized response.
        base_reward: Reward score for the unperturbed response.
        reward_fn: Function that scores a token sequence.
        perturbation_candidates: Map of tokens to synonym replacements.
 
    Returns:
        Sensitivity score for each token position.
    """
    sensitivities = {}
 
    for i, token in enumerate(tokens):
        if token in perturbation_candidates:
            max_delta = 0.0
            for replacement in perturbation_candidates[token]:
                perturbed = tokens.copy()
                perturbed[i] = replacement
                new_reward = reward_fn(perturbed)
                delta = abs(new_reward - base_reward)
                max_delta = max(max_delta, delta)
            sensitivities[f"position_{i}_{token}"] = max_delta
 
    return sensitivities
 
def generate_reward_adversarial_example(
    original_tokens: list[str],
    reward_fn: callable,
    perturbation_candidates: dict[str, list[str]],
    max_perturbations: int = 5,
) -> tuple[list[str], float]:
    """
    Greedily perturb tokens to maximize reward score.
 
    Uses iterative greedy search: at each step, try all possible
    single-token perturbations and apply the one that maximizes
    the reward increase. Repeat up to max_perturbations times.
    """
    current_tokens = original_tokens.copy()
    current_reward = reward_fn(current_tokens)
 
    for _ in range(max_perturbations):
        best_tokens = current_tokens
        best_reward = current_reward
 
        for i, token in enumerate(current_tokens):
            if token not in perturbation_candidates:
                continue
            for replacement in perturbation_candidates[token]:
                candidate = current_tokens.copy()
                candidate[i] = replacement
                candidate_reward = reward_fn(candidate)
                if candidate_reward > best_reward:
                    best_tokens = candidate
                    best_reward = candidate_reward
 
        if best_reward <= current_reward:
            break  # No improvement found
 
        current_tokens = best_tokens
        current_reward = best_reward
 
    return current_tokens, current_reward
 
# Simplified demonstration with a mock reward function
def mock_reward(tokens: list[str]) -> float:
    """Mock reward function that has exploitable biases."""
    score = 0.5
    # Bias: reward model prefers formal language
    formal_words = {"furthermore", "consequently", "therefore", "moreover"}
    score += 0.1 * sum(1 for t in tokens if t.lower() in formal_words)
    # Bias: reward model penalizes uncertainty
    uncertain_words = {"maybe", "perhaps", "possibly", "might"}
    score -= 0.15 * sum(1 for t in tokens if t.lower() in uncertain_words)
    return min(1.0, max(0.0, score + np.random.normal(0, 0.02)))
 
original = ["the", "answer", "might", "be", "42"]
perturbations = {
    "might": ["is", "will", "consequently"],
    "the": ["furthermore,", "moreover,", "the"],
}
 
adversarial, adv_reward = generate_reward_adversarial_example(
    original, mock_reward, perturbations, max_perturbations=3
)
print(f"Original: {' '.join(original)} -> reward: {mock_reward(original):.3f}")
print(f"Adversarial: {' '.join(adversarial)} -> reward: {adv_reward:.3f}")

"""
Reward model ensemble exploitation.
Shows how an attacker can exploit disagreements between
ensemble members to find reward-hacking strategies.
"""
import numpy as np
from typing import Protocol
 
class RewardModel(Protocol):
    def score(self, prompt: str, response: str) -> float: ...
 
def find_ensemble_disagreement_regions(
    prompts: list[str],
    responses_per_prompt: list[list[str]],
    reward_models: list[callable],
    disagreement_threshold: float = 0.3,
) -> list[dict]:
    """
    Identify prompt-response pairs where ensemble members disagree.
 
    High-disagreement regions are where reward hacking is most likely
    to succeed, because the policy can exploit one model's preferences
    while the others provide weak signal.
 
    Args:
        prompts: Input prompts.
        responses_per_prompt: Multiple candidate responses per prompt.
        reward_models: List of reward model scoring functions.
        disagreement_threshold: Minimum std dev across models to flag.
 
    Returns:
        Flagged high-disagreement instances.
    """
    flagged = []
 
    for prompt_idx, (prompt, responses) in enumerate(
        zip(prompts, responses_per_prompt)
    ):
        for resp_idx, response in enumerate(responses):
            scores = [rm(prompt, response) for rm in reward_models]
            std_dev = np.std(scores)
            if std_dev > disagreement_threshold:
                flagged.append({
                    "prompt_idx": prompt_idx,
                    "response_idx": resp_idx,
                    "scores": scores,
                    "mean": float(np.mean(scores)),
                    "std": float(std_dev),
                    "max_min_gap": float(max(scores) - min(scores)),
                })
 
    return flagged
 
def exploit_aggregation_strategy(
    candidate_scores: list[list[float]],
    aggregation: str = "mean",
) -> int:
    """
    Given per-model scores for multiple candidates, find the candidate
    that maximizes the aggregated score — demonstrating how knowledge
    of the aggregation strategy aids exploitation.
    """
    aggregated = []
    for scores in candidate_scores:
        if aggregation == "mean":
            aggregated.append(np.mean(scores))
        elif aggregation == "min":
            aggregated.append(np.min(scores))
        elif aggregation == "median":
            aggregated.append(np.median(scores))
        else:
            raise ValueError(f"Unknown aggregation: {aggregation}")
 
    return int(np.argmax(aggregated))
 
# Demonstration
candidate_scores = [
    [0.9, 0.2, 0.5],  # High on model 1, low on model 2
    [0.6, 0.6, 0.6],  # Consistent across models
    [0.3, 0.95, 0.4],  # High on model 2, low on model 1
]
 
for strategy in ["mean", "min", "median"]:
    best = exploit_aggregation_strategy(candidate_scores, strategy)
    print(f"Aggregation '{strategy}': best candidate = {best}, "
          f"scores = {candidate_scores[best]}")

"""
Reward model auditing framework.
Implements systematic checks for reward model integrity
throughout the RLHF training process.
"""
import numpy as np
from dataclasses import dataclass, field
from enum import Enum
 
class AuditSeverity(Enum):
    INFO = "info"
    WARNING = "warning"
    CRITICAL = "critical"
 
@dataclass
class AuditFinding:
    check_name: str
    severity: AuditSeverity
    message: str
    details: dict = field(default_factory=dict)
 
def audit_reward_distribution(
    scores: np.ndarray,
    expected_mean: float = 0.0,
    expected_std: float = 1.0,
    tolerance: float = 0.5,
) -> list[AuditFinding]:
    """
    Check that reward score distribution matches expectations.
 
    Large deviations may indicate reward model corruption or
    distributional shift in the policy's outputs.
    """
    findings = []
    actual_mean = np.mean(scores)
    actual_std = np.std(scores)
 
    if abs(actual_mean - expected_mean) > tolerance:
        findings.append(AuditFinding(
            check_name="reward_distribution_mean",
            severity=AuditSeverity.WARNING,
            message=(
                f"Reward mean ({actual_mean:.3f}) deviates from expected "
                f"({expected_mean:.3f}) by more than {tolerance}"
            ),
            details={"actual_mean": float(actual_mean), "expected_mean": expected_mean},
        ))
 
    if actual_std < expected_std * 0.5 or actual_std > expected_std * 2.0:
        findings.append(AuditFinding(
            check_name="reward_distribution_std",
            severity=AuditSeverity.CRITICAL,
            message=(
                f"Reward std ({actual_std:.3f}) is outside expected range "
                f"[{expected_std*0.5:.3f}, {expected_std*2.0:.3f}]"
            ),
            details={"actual_std": float(actual_std), "expected_std": expected_std},
        ))
 
    return findings
 
def audit_reward_consistency(
    pairs: list[tuple[str, str, float, float]],
    min_concordance: float = 0.8,
) -> list[AuditFinding]:
    """
    Check that reward model rankings are consistent with known preferences.
 
    Takes pairs of (prompt, chosen, rejected) with reward scores and
    verifies the reward model assigns higher scores to chosen responses.
 
    Args:
        pairs: List of (chosen_text, rejected_text, chosen_score, rejected_score).
        min_concordance: Minimum fraction of pairs where chosen > rejected.
    """
    if not pairs:
        return []
 
    concordant = sum(1 for _, _, cs, rs in pairs if cs > rs)
    concordance_rate = concordant / len(pairs)
 
    findings = []
    if concordance_rate < min_concordance:
        findings.append(AuditFinding(
            check_name="reward_consistency",
            severity=AuditSeverity.CRITICAL,
            message=(
                f"Reward model concordance ({concordance_rate:.1%}) is below "
                f"threshold ({min_concordance:.1%}). Possible reward model corruption."
            ),
            details={
                "concordance_rate": concordance_rate,
                "num_pairs": len(pairs),
                "concordant": concordant,
            },
        ))
 
    return findings
 
def audit_kl_divergence(
    kl_values: np.ndarray,
    max_kl: float = 15.0,
    trend_window: int = 100,
) -> list[AuditFinding]:
    """
    Monitor KL divergence between policy and reference model.
 
    Excessive KL divergence indicates the policy has moved far from
    the reference, increasing the risk of reward hacking.
    """
    findings = []
    current_kl = np.mean(kl_values[-trend_window:])
 
    if current_kl > max_kl:
        findings.append(AuditFinding(
            check_name="kl_divergence_limit",
            severity=AuditSeverity.CRITICAL,
            message=(
                f"KL divergence ({current_kl:.2f}) exceeds maximum "
                f"({max_kl:.2f}). Policy may be overoptimized."
            ),
            details={"current_kl": float(current_kl), "max_kl": max_kl},
        ))
 
    # Check for rapid increase (sign of aggressive exploitation)
    if len(kl_values) > trend_window * 2:
        recent = np.mean(kl_values[-trend_window:])
        previous = np.mean(kl_values[-2*trend_window:-trend_window])
        if recent > previous * 1.5:
            findings.append(AuditFinding(
                check_name="kl_divergence_trend",
                severity=AuditSeverity.WARNING,
                message=(
                    f"KL divergence increasing rapidly: {previous:.2f} -> "
                    f"{recent:.2f} (50%+ increase in {trend_window} steps)"
                ),
            ))
 
    return findings
 
# Run audit demonstration
np.random.seed(42)
sample_scores = np.random.normal(0.1, 0.8, 500)  # Slightly shifted mean
findings = audit_reward_distribution(sample_scores)
for f in findings:
    print(f"[{f.severity.value.upper()}] {f.check_name}: {f.message}")
 
sample_kl = np.concatenate([
    np.linspace(0.5, 5, 200),
    np.linspace(5, 18, 300),  # Rapid increase
])
kl_findings = audit_kl_divergence(sample_kl)
for f in kl_findings:
    print(f"[{f.severity.value.upper()}] {f.check_name}: {f.message}")