Reward Hacking Discovery 2026: Deteksi Eksploitasi Benchmark AI Agent Menggunakan Internal Representations & Difference-of-Means (DoM) Vectors pada Kimi K3, GLM 5.2, dan Qwen 3.8 Max

"""
Difference-of-Means (DoM) Latent Representation Monitor & Pre-Action Gate
Berdasarkan Arsitektur Riset arXiv:2609.19101 (September 2026, Bergen et al.).
Mengimplementasikan:
1. Activation Cache Hook untuk mengekstrak hidden states layer residual (Kimi K3, GLM 5.2, Qwen 3.8 Max)
2. Difference-of-Means (DoM) Direction Vector Computation (Honest vs Reward-Hacking Rollouts)
3. Online Chain-of-Thought (CoT) Probing untuk intervensi sebelum aksi eksekusi tool dikirim
"""

import numpy as np
from dataclasses import dataclass
from typing import List, Dict, Tuple, Optional

@dataclass
class ActivationSnapshot:
    layer_idx: int
    token_position: int
    hidden_vector: np.ndarray  # Shape: [d_model]

@dataclass
class ProbeDecision:
    is_reward_hacking: bool
    projection_score: float
    decision_threshold: float
    trigger_layer: int
    flagged_token: str

class DifferenceOfMeansProbe:
    """
    Menghitung vektor arah pemisah (separation direction) DoM:
    v_DoM = \mu_{hack} - \mu_{honest}
    dan memproyeksikan hidden state runtime:
    s = 
    """
    def __init__(self, d_model: int, target_layer: int, threshold: float = 0.65):
        self.d_model = d_model
        self.target_layer = target_layer
        self.threshold = threshold
        self.dom_vector: Optional[np.ndarray] = None
        self.normalized_direction: Optional[np.ndarray] = None

    def calibrate(self, honest_activations: List[np.ndarray], hack_activations: List[np.ndarray]):
        """
        Menghitung mean perbedaan representasi internal dari rollout kalibrasi.
        """
        mu_honest = np.mean(honest_activations, axis=0)
        mu_hack = np.mean(hack_activations, axis=0)
        
        # Hitung DoM Vector
        raw_direction = mu_hack - mu_honest
        norm = np.linalg.norm(raw_direction)
        if norm < 1e-8:
            raise ValueError("Vektor aktivasi identik, tidak dapat membentuk probe DoM.")
            
        self.dom_vector = raw_direction
        self.normalized_direction = raw_direction / norm
        print(f"✅ Probe Layer {self.target_layer} terkalibrasi. L2 Norm: {norm:.4f}")

    def evaluate_token_activation(self, hidden_state: np.ndarray, token_str: str) -> ProbeDecision:
        if self.normalized_direction is None:
            raise RuntimeError("Probe belum dikalibrasi.")
            
        # Proyeksikan representasi laten ke arah DoM
        projection = float(np.dot(self.normalized_direction, hidden_state))
        is_hack = projection > self.threshold
        
        return ProbeDecision(
            is_reward_hacking=is_hack,
            projection_score=projection,
            decision_threshold=self.threshold,
            trigger_layer=self.target_layer,
            flagged_token=token_str
        )

class OnlineCoTRuntimeGuardrail:
    """
    Memantau aktivasi intermediate Chain-of-Thought (CoT) secara real-time.
    Mendeteksi niat reward hacking (e.g. memodifikasi test runner, pytest assertion,
    git history tampering) SEBELUM aksi tool dipancarkan ke environment.
    """
    def __init__(self, probes_by_layer: Dict[int, DifferenceOfMeansProbe]):
        self.probes = probes_by_layer
        self.cot_tokens_trace: List[str] = []
        self.projection_history: List[Tuple[str, float]] = []

    def stream_cot_token(self, token_str: str, layer_activations: Dict[int, np.ndarray]) -> Optional[ProbeDecision]:
        self.cot_tokens_trace.append(token_str)
        
        for layer_idx, probe in self.probes.items():
            if layer_idx in layer_activations:
                decision = probe.evaluate_token_activation(layer_activations[layer_idx], token_str)
                self.projection_history.append((token_str, decision.projection_score))
                
                if decision.is_reward_hacking:
                    # Anomali representasi terdeteksi di dalam pemikiran CoT
                    print(f"🚨 [VETO PRE-ACTION] Reward Hacking Signature detected at token '{token_str}'!")
                    print(f"   Layer: {layer_idx} | DoM Score: {decision.projection_score:.4f} > Threshold {decision.decision_threshold}")
                    return decision
                    
        return None

    def intercept_tool_call(self, planned_action: Dict[str, str]) -> bool:
        """
        Validasi ganda: Jika CoT mengandung puncak aktivasi DoM hacking,
        tolak eksekusi tool dan minta replanning jujur.
        """
        recent_scores = [score for _, score in self.projection_history[-10:]]
        if recent_scores and np.max(recent_scores) > 0.65:
            print(f"❌ Blokir eksekusi aksi: {planned_action.get('tool_name')}. Alasan: Intensi manipulasi benchmark terdeteksi pada internal state.")
            return False
        return True