NY Expansion Analysis

import asyncio
import sys
import re
import os
import json
from pathlib import Path

def get_model_response():
    """
    Get the model's response from the MCP_MESSAGES environment variable.
    Returns the last assistant message text.
    """
    messages_path = os.getenv("MCP_MESSAGES")
    print(f"MCP_MESSAGES: {messages_path}")
    if not messages_path:
        print("ERROR: MCP_MESSAGES environment variable not set", file=sys.stderr)
        return None
    
    # Check if file exists
    if not Path(messages_path).exists():
        print(f"ERROR: Messages file not found at path: {messages_path}", file=sys.stderr)
        return None
    
    try:
        with open(messages_path, 'r') as f:
            content = f.read()
            
        # Check if file is empty
        if not content or content.strip() == '""':
            print("ERROR: Messages file is empty or contains only empty string", file=sys.stderr)
            return None
            
        messages = json.loads(content)
        
        # Check if messages is a list
        if not isinstance(messages, list):
            print(f"ERROR: Messages file should contain a list, got {type(messages).__name__}", file=sys.stderr)
            return None
        
        # Find the last assistant message
        for message in reversed(messages):
            if message.get('role') == 'assistant' and message.get('status') == 'completed':
                content = message.get('content', [])
                if not content:
                    print("WARNING: Assistant message has empty content", file=sys.stderr)
                    continue
                    
                for item in content:
                    if item.get('type') == 'output_text':
                        text = item.get('text', '')
                        if not text:
                            print("WARNING: Output text is empty", file=sys.stderr)
                            continue
                        return text
        
        print("ERROR: No assistant response with output_text found in messages", file=sys.stderr)
        return None
    except json.JSONDecodeError as e:
        print(f"ERROR: Invalid JSON in messages file: {str(e)}", file=sys.stderr)
        return None
    except Exception as e:
        print(f"ERROR: Unexpected error reading messages file: {str(e)}", file=sys.stderr)
        return None

def parse_answer_format(text):
    """
    Parse the <answer>...</answer> format from the agent's output.
    Returns a dictionary with the parsed values.
    """
    if not text:
        print("ERROR: No text provided to parse", file=sys.stderr)
        return None
    
    # Look for <answer>...</answer> pattern
    match = re.search(r'<answer>(.*?)</answer>', text, re.IGNORECASE | re.DOTALL)
    if not match:
        print("ERROR: No <answer> tags found in the response", file=sys.stderr)
        print(f"  Response preview: {text[:200]}...", file=sys.stderr)
        return None
    
    answer_content = match.group(1).strip()
    
    if not answer_content:
        print("ERROR: Empty content between <answer> tags", file=sys.stderr)
        return None
    
    # Parse each line
    result = {}
    lines = answer_content.split('\n')
    
    # Expected keys that should be present
    expected_keys = [
        'Lifetime_Sales_Amount', 'Cheap_Bestseller_Name', 'Second_Bestseller_Price',
        'Second_Bestseller_Quantity', 'Product_In_Last_Orders', 'NY_Tax_Rate',
        'CA_Tax_Rate', 'Higher_Tax_State', 'Total_States_With_Tax',
        'Processing_Visible_Storefront', 'Processing_Default_Status',
        'Number_Of_Websites', 'Main_Store_Code', 'Default_Source_Pickup_Status',
        'Default_Source_State', 'Dashboard_Revenue', 'Tax_Shipping_Zero'
    ]
    
    parsed_keys = []
    for line in lines:
        line = line.strip()
        if not line:
            continue
            
        if '|' not in line:
            print(f"ERROR: Line missing pipe separator '|': {line}", file=sys.stderr)
            continue
            
        parts = line.split('|', 1)
        if len(parts) != 2:
            print(f"ERROR: Invalid line format: {line}", file=sys.stderr)
            continue
            
        key, value = parts
        key = key.strip()
        value = value.strip()
        
        if not key:
            print(f"ERROR: Empty key in line: {line}", file=sys.stderr)
            continue
            
        result[key] = value
        parsed_keys.append(key)
    
    # Check for missing expected keys
    missing_keys = set(expected_keys) - set(parsed_keys)
    if missing_keys:
        print(f"ERROR: Missing expected keys: {', '.join(sorted(missing_keys))}", file=sys.stderr)
        
    # Check for unexpected keys
    unexpected_keys = set(parsed_keys) - set(expected_keys)
    if unexpected_keys:
        print(f"WARNING: Unexpected keys found: {', '.join(sorted(unexpected_keys))}", file=sys.stderr)
    
    if not result:
        print("ERROR: No valid key-value pairs parsed from answer", file=sys.stderr)
        return None
    
    return result

def load_expected_answer(label_path):
    """
    Load the expected answer from label.txt file.
    Returns a dictionary with the expected values.
    """
    try:
        with open(label_path, 'r') as f:
            lines = f.read().strip().split('\n')
        
        expected = {}
        for line in lines:
            if '|' in line:
                key, value = line.split('|', 1)
                expected[key.strip()] = value.strip()
        
        return expected
    except Exception as e:
        print(f"Error reading label file: {str(e)}", file=sys.stderr)
        return None

def compare_answers(model_answer, expected_answer):
    """
    Compare the model's answer with the expected answer.
    Returns True if all key information matches, False otherwise.
    """
    if not model_answer or not expected_answer:
        return False
    
    # Check each expected key
    mismatches = []
    for key, expected_value in expected_answer.items():
        model_value = model_answer.get(key, '')
        
        # Special handling for different types of values
        if key in ['Lifetime_Sales_Amount', 'Second_Bestseller_Price', 'Dashboard_Revenue']:
            # For price/amount fields, normalize format
            expected_clean = expected_value.replace('$', '').replace(',', '')
            model_clean = model_value.replace('$', '').replace(',', '')
            if expected_clean != model_clean:
                mismatches.append(f"{key}: expected '{expected_value}', got '{model_value}'")
        
        elif key in ['NY_Tax_Rate', 'CA_Tax_Rate']:
            # Tax rates - allow different decimal formats
            expected_clean = expected_value.replace('%', '').strip()
            model_clean = model_value.replace('%', '').strip()
            # Convert to float for comparison
            try:
                if float(expected_clean) != float(model_clean):
                    mismatches.append(f"{key}: expected '{expected_value}', got '{model_value}'")
            except ValueError:
                if expected_clean != model_clean:
                    mismatches.append(f"{key}: expected '{expected_value}', got '{model_value}'")
        
        elif key in ['Product_In_Last_Orders', 'Processing_Visible_Storefront', 'Processing_Default_Status', 'Tax_Shipping_Zero']:
            # Yes/No fields - case insensitive
            if model_value.lower() != expected_value.lower():
                mismatches.append(f"{key}: expected '{expected_value}', got '{model_value}'")
        
        elif key == 'Empty_Rows_Yes_Effect':
            # Allow flexible descriptions for this field
            # Just check if model provided some reasonable description
            if not model_value or len(model_value) < 5:
                mismatches.append(f"{key}: expected meaningful description, got '{model_value}'")
        
        elif key == 'Order_Status_Options':
            # Check if main options are mentioned
            expected_options = set(opt.strip() for opt in expected_value.split(','))
            model_options = set(opt.strip() for opt in model_value.split(','))
            if expected_options != model_options:
                mismatches.append(f"{key}: expected '{expected_value}', got '{model_value}'")
        
        elif key == 'Chart_Disabled_Message':
            # Allow some flexibility in message text
            # Check for key words
            if 'disabled' not in model_value.lower() and 'enable' not in model_value.lower():
                mismatches.append(f"{key}: expected message about chart being disabled, got '{model_value}'")
        
        elif key == 'Default_Source_State':
            # Handle 'None' or empty state
            expected_normalized = expected_value.lower() if expected_value.lower() != 'none' else ''
            model_normalized = model_value.lower() if model_value.lower() != 'none' else ''
            if expected_normalized != model_normalized:
                mismatches.append(f"{key}: expected '{expected_value}', got '{model_value}'")
        
        else:
            # Exact match for other fields
            if model_value != expected_value:
                mismatches.append(f"{key}: expected '{expected_value}', got '{model_value}'")
    
    if mismatches:
        print("\n=== Answer Comparison Mismatches ===", file=sys.stderr)
        for mismatch in mismatches:
            print(f"✗ {mismatch}", file=sys.stderr)
        return False
    
    print("\n=== Answer Comparison ===", file=sys.stderr)
    print("✓ All key information matches the expected answer", file=sys.stderr)
    return True

async def verify() -> bool:
    """
    Verifies that the NY expansion analysis task has been completed correctly.
    First checks the model's answer against the expected label,
    then optionally verifies the actual state in the Magento Admin.
    """
    print("\n=== Starting Verification ===", file=sys.stderr)
    
    # Get the label file path
    label_path = Path(__file__).parent / "label.txt"
    
    # Load expected answer
    print("Loading expected answer from label.txt...", file=sys.stderr)
    expected_answer = load_expected_answer(label_path)
    if not expected_answer:
        print("FATAL ERROR: Could not load expected answer from label.txt", file=sys.stderr)
        return False
    
    print(f"Expected answer loaded with {len(expected_answer)} keys", file=sys.stderr)
    
    # Get model's response from MCP_MESSAGES
    print("\nReading model response from MCP_MESSAGES...", file=sys.stderr)
    model_response = get_model_response()
    
    if not model_response:
        print("FATAL ERROR: No valid model response found", file=sys.stderr)
        return False
    
    print(f"Model response found (length: {len(model_response)} chars)", file=sys.stderr)
    print("\nParsing answer format from model response...", file=sys.stderr)
    
    model_answer = parse_answer_format(model_response)
    
    if not model_answer:
        print("FATAL ERROR: Could not parse answer format from model response", file=sys.stderr)
        return False
    
    print(f"\n=== Model Answer Parsed Successfully ===", file=sys.stderr)
    print(f"Parsed {len(model_answer)} key-value pairs", file=sys.stderr)
    
    for key, value in model_answer.items():
        print(f"  {key}: {value}", file=sys.stderr)
    
    # Compare answers
    print("\n=== Comparing Model Answer with Expected Answer ===", file=sys.stderr)
    answer_match = compare_answers(model_answer, expected_answer)
    
    if not answer_match:
        print("\nFATAL ERROR: Model answer does not match expected answer", file=sys.stderr)
        print("Verification FAILED", file=sys.stderr)
        return False
    
    print("\n✓ Model answer matches expected answer", file=sys.stderr)
    print("Verification PASSED", file=sys.stderr)
    return True

def main():
    """
    Executes the verification process and exits with a status code.
    """
    result = asyncio.run(verify())
    sys.exit(0 if result else 1)

if __name__ == "__main__":
    main()