Harness-Zero 2026: Arsitektur Harness Distillation via Agent-as-Harness, Adaptasi Private Reference \mathcal{K}, dan Internalisasi Perilaku Otonom ke Bobot Model

"""
Harness-Zero: Harness Distillation via Agent-as-Harness Architecture.
Berdasarkan Riset Frontier arXiv:2609.24974 (Peking University, Google, HKUST - September 2026).

Komponen Arsitektur Utama:
1. Target Harness (h): Lingkungan eksekusi minimal & kaku (misal mini-SWE-agent dengan tool Bash tunggal).
2. Evolved Student Harness (h*): Scaffolding kaya fitur (tools, middleware guards, memory, skills spesifik domain).
3. Private Reference Harness (K = Adapt(h*; h)): Rekonfigurasi guard middleware dan memory menjadi instruksi review.
4. Harnessing Agent (pi_H): Membungkus response boundary student (pi_S), memvalidasi proposal y_t, 
   dan menyuntikkan koreksi z_t valid dalam action space target harness (h) tanpa membocorkan state oracle.
5. Trajectory Distillation SFT: Supervised Fine-Tuning pada trajektori diterima (y_tilde_t) dengan 
   loss masking pada reasoning reviewer agar model student menginternalisasi pola ke dalam bobot theta.
"""

from dataclasses import dataclass, field
from enum import Enum
from typing import Dict, List, Optional, Tuple, Any
import json

class ReviewDecision(Enum):
    PASS = "pass"
    REPLACE = "replace"

@dataclass
class TargetAction:
    tool_name: str
    command: str

@dataclass
class StudentProposal:
    thought: str
    action: TargetAction
    raw_response: str

@dataclass
class ReviewResult:
    decision: ReviewDecision
    accepted_response: str
    correction_rationale: Optional[str] = None  # Hanya untuk internal harnessing agent, di-mask saat SFT

class PrivateReferenceHarness:
    """
    Representasi K = Adapt(h*; h).
    Mengonversi middleware h* (seperti prefilled_guard, answer_range_guard, turn_budget)
    menjadi checklist verifikasi aktif pada response boundary.
    """
    def __init__(self, domain: str, rules: List[str]):
        self.domain = domain
        self.rules = rules

    def audit_proposal(self, context_history: List[Dict[str, Any]], proposal: StudentProposal) -> Tuple[bool, Optional[str]]:
        # Contoh validasi spreadsheet: memastikan cell pre-filled tidak tertimpa
        if "solution.py" in proposal.action.command and "mode='w'" in proposal.action.command:
            # Middleware check: Deteksi potensi penimpaan file tanpa verifikasi awal
            has_inspected = any("inspect" in str(step.get("command", "")) for step in context_history)
            if not has_inspected:
                return False, "RULE_VIOLATION: Harap lakukan inspeksi struktur header dan prefilled cells sebelum menulis skrip mutasi."
        
        # Contoh validasi penyelesaian prematur (finish-guard)
        if "COMPLETE_TASK_AND_SUBMIT" in proposal.raw_response:
            has_verified = any("python3 -m unittest" in str(step.get("command", "")) or "verify" in str(step.get("command", "")) for step in context_history)
            if not has_verified:
                return False, "RULE_VIOLATION: Dilarang submit sebelum menjalankan perintah assertion verifikasi mandiri di action space target."

        return True, None

class HarnessingAgent:
    """
    Agen supervisor pi_H yang membungkus response boundary student pi_S.
    Koreksi (z_t) wajib valid 100% di dalam action space target harness (h).
    """
    def __init__(self, reference_harness: PrivateReferenceHarness):
        self.reference = reference_harness
        self.private_history: List[Dict[str, Any]] = []

    def review_turn(
        self,
        student_context: List[Dict[str, Any]],
        proposal: StudentProposal
    ) -> ReviewResult:
        is_sound, violation_reason = self.reference.audit_proposal(student_context, proposal)
        
        if is_sound:
            return ReviewResult(
                decision=ReviewDecision.PASS,
                accepted_response=proposal.raw_response
            )
        
        # Intervensi Terarah: Menghasilkan pengganti terkecil yang koheren (smallest coherent correction)
        # Penting: Harus diformulasikan dari sudut pandang student di action space target harness h
        corrected_thought = f"Sebelum melakukan aksi mutasi, saya perlu memverifikasi struktur file target terlebih dahulu."
        corrected_action = TargetAction(
            tool_name="execute",
            command="python3 -c 'import openpyxl; wb=openpyxl.load_workbook("target.xlsx"); print(wb.sheetnames)'"
        )
        corrected_raw = f"{corrected_thought}\n```bash\n{corrected_action.command}\n```"
        
        # Simpan jejak internal reviewer (tidak dikirimkan ke context student)
        self.private_history.append({
            "turn": len(student_context) + 1,
            "rejected_proposal": proposal.raw_response,
            "reason": violation_reason
        })
        
        return ReviewResult(
            decision=ReviewDecision.REPLACE,
            accepted_response=corrected_raw,
            correction_rationale=violation_reason
        )

class HarnessDistillationDataset:
    """
    Mengompilasi trajektori D_review menjadi format SFT.
    Menerapkan token loss masking pada reasoning reviewer agar bobot student (theta)
    hanya mempelajari transisi state eksekusi murni di bawah h.
    """
    @staticmethod
    def format_trajectory_for_sft(trajectory: List[Dict[str, Any]]) -> Dict[str, Any]:
        prompt_tokens = []
        target_tokens = []
        loss_masks = []
        
        for turn in trajectory:
            context_str = turn["context"]
            accepted_resp = turn["accepted_response"]
            
            # Loss mask diaktifkan hanya pada token accepted_response dari perspektif student
            # Token reasoning supervisor dihilangkan secara definitif
            prompt_tokens.append(context_str)
            target_tokens.append(accepted_resp)
            loss_masks.append(1)  # Full weight pada respons yang dieksekusi di h
            
        return {
            "inputs": prompt_tokens,
            "targets": target_tokens,
            "loss_mask": loss_masks
        }