RRSI 2026: Arsitektur Regularized Recursive Self-Improvement, Temporally Annealed Proposer, Critic-Pruner Selector, dan Mitigasi Overfitting pada Agent Harness Evolution

"""
RRSI: Regularized Recursive Self-Improvement of Agent Harnesses Engine.
Berdasarkan formulasi riset Google Cloud AI Research, Stanford & UC Berkeley (arXiv:2609.24972).

Komponen Inti:
1. L0-Style Annealed Update Sparsity: b_t = ceil(b_min + (b_max - b_min) * 0.5 * (1 + cos(pi * t / T)))
2. Evidence-Aware Credit Assignment: Melacak history mutasi, dif, skor delta, dan mendiskualifikasi kegagalan berulang.
3. Structured Exploration: Mengalokasikan eksplorasi komponen dorman saat progres stagnan dalam noise band delta.
4. Leakage Screening (Critic): Menolak kode mutasi yang mengandung konstanta spesifik benchmark atau inert dead-code.
5. Stability-Aware Floor: Menolak regresi noise: S_hat(H') >= S* - delta.
6. Ridge/L2-Style Complexity Acceptance: Delta_C <= beta_0 + beta_1 * Delta_S (mencegah token bloat).
7. Lasso/L1-Style Structural Pruning: Memangkas modul scaffolding yang tidak menghasilkan gain positif selama window W.
"""

from dataclasses import dataclass, field
from enum import Enum
from typing import Dict, List, Optional, Set, Tuple
import math
import re

@dataclass
class HarnessMutation:
    mutation_id: str
    target_component: str  # 'prompt', 'control_flow', 'tool_def', 'memory', 'context'
    diff_content: str
    description: str

@dataclass
class ProposalCandidate:
    candidate_id: str
    round_idx: int
    mutations: List[HarnessMutation]
    expected_rationale: str

@dataclass
class EvaluationMetrics:
    score: float
    token_cost: float
    execution_time_ms: float

class SelectionVerdict(Enum):
    ACCEPTED = "accepted"
    REJECTED_LEAKAGE = "rejected_leakage"
    REJECTED_NOISE_FLOOR = "rejected_noise_floor"
    REJECTED_COMPLEXITY_BLOAT = "rejected_complexity_bloat"
    REJECTED_NO_GAIN = "rejected_no_gain"

class RRSIRegularizedHarnessOptimizer:
    def __init__(
        self,
        total_rounds: int = 50,
        b_max: int = 4,
        b_min: int = 1,
        noise_band_delta: float = 0.8,
        beta_0: float = 0.05,
        beta_1: float = 0.08,
        pruning_window: int = 6,
        stall_window: int = 4
    ):
        self.total_rounds = total_rounds
        self.b_max = b_max
        self.b_min = b_min
        self.delta = noise_band_delta
        self.beta_0 = beta_0
        self.beta_1 = beta_1
        self.pruning_window = pruning_window
        self.stall_window = stall_window
        
        # State tracking
        self.current_round = 0
        self.incumbent_score = 0.0
        self.incumbent_cost = 1000.0  # baseline policy token count
        self.best_observed_score = 0.0
        self.history: List[Dict] = []
        self.component_utility: Dict[str, List[float]] = {}
        self.known_components = {"system_prompt", "bash_tool", "python_repl", "scratchpad_memory", "verifier_guard"}

    def compute_annealed_budget(self, t: int) -> int:
        """Formulasi Persamaan (4): L0-Style Annealed Update Sparsity."""
        decay_factor = 0.5 * (1.0 + math.cos(math.pi * t / self.total_rounds))
        budget = math.ceil(self.b_min + (self.b_max - self.b_min) * decay_factor)
        return max(self.b_min, min(self.b_max, budget))

    def screen_leakage(self, candidate: ProposalCandidate) -> Tuple[bool, Optional[str]]:
        """Selection Gate 1: Leakage Screening (Critic). Mencegah overfit memorization."""
        leakage_patterns = [
            r"test_w+.py", r"eval_dataset", r"case_d+", r"asserts+results*==",
            r"expected_output", r"hardcoded_fix", r"ifs+task_ids*=="
        ]
        for mut in candidate.mutations:
            for pattern in leakage_patterns:
                if re.search(pattern, mut.diff_content, re.IGNORECASE):
                    return False, f"Benchmark-specific leakage detected: pattern '{pattern}' in {mut.target_component}"
            if len(mut.diff_content.strip()) < 5:
                return False, "Inert machinery / empty mutation detected"
        return True, None

    def evaluate_and_select(
        self,
        candidate: ProposalCandidate,
        eval_metrics: EvaluationMetrics
    ) -> Tuple[SelectionVerdict, str]:
        """Selection Gate 2-4: Stability, Complexity Shrinkage, and Incumbent Replacement."""
        # 1. Leakage Critic
        is_clean, reason = self.screen_leakage(candidate)
        if not is_clean:
            return SelectionVerdict.REJECTED_LEAKAGE, reason

        s_cand = eval_metrics.score
        c_cand = eval_metrics.token_cost
        
        # 2. Stability-Aware Floor: S_hat(H') >= S* - delta
        if s_cand < (self.best_observed_score - self.delta):
            return SelectionVerdict.REJECTED_NOISE_FLOOR, f"Score {s_cand:.2f} falls below stability floor {self.best_observed_score - self.delta:.2f}"

        # Hitung deltas relatif terhadap incumbent saat ini
        delta_s = s_cand - self.incumbent_score
        delta_c = (c_cand - self.incumbent_cost) / self.incumbent_cost if self.incumbent_cost > 0 else 0.0

        # 3. Ridge/L2 Complexity Shrinkage Gate: Delta_C <= beta_0 + beta_1 * Delta_S
        if delta_s > self.delta:
            max_allowed_cost_increase = self.beta_0 + self.beta_1 * delta_s
            if delta_c > max_allowed_cost_increase:
                return (
                    SelectionVerdict.REJECTED_COMPLEXITY_BLOAT,
                    f"Cost increase {delta_c*100:.1f}% exceeds Ridge complexity threshold {max_allowed_cost_increase*100:.1f}% for gain +{delta_s:.2f}"
                )
        else:
            if delta_s <= 0 and delta_c >= 0:
                return SelectionVerdict.REJECTED_NO_GAIN, "Candidate yields non-positive score gain without cost reduction"

        # Candidate Diterima: Perbarui State
        self.incumbent_score = s_cand
        self.incumbent_cost = c_cand
        if s_cand > self.best_observed_score:
            self.best_observed_score = s_cand

        return SelectionVerdict.ACCEPTED, f"Accepted: +{delta_s:.2f} gain with {delta_c*100:+.1f}% cost drift"

    def identify_structural_pruning_targets(self) -> Set[str]:
        """Lasso/L1-Style Structural Pruning: Identifikasi komponen yang tidak berkontribusi positif selama window W."""
        prune_targets = set()
        for comp, gains in self.component_utility.items():
            recent_gains = gains[-self.pruning_window:]
            if len(recent_gains) >= self.pruning_window and sum(recent_gains) <= 0.0:
                prune_targets.add(comp)
        return prune_targets