strat-gameplay-webapp/.claude/implementation/phase-3e-websocket-events.md

Phase 3E: WebSocket Events & X-Check UI Integration

Status: Not Started Estimated Effort: 5-6 hours Dependencies: Phase 3C (Resolution Logic), Phase 3D (Advancement)

Overview

Implement WebSocket event handlers for X-Check plays supporting three modes:

1. PD Auto: System auto-resolves, shows result with Accept/Reject
2. PD Manual: Shows dice + charts, player selects from options, Accept/Reject
3. SBA Manual: Shows dice + options, player selects (no charts available)
4. SBA Semi-Auto: Like PD Manual (if position ratings provided)

Also implements:

• Position rating loading at lineup creation
• Redis caching for all player positions
• Override logging when player rejects auto-resolution

Tasks

1. Add Position Rating Loading on Lineup Creation

File: backend/app/services/pd_api_client.py (or create if doesn't exist)

"""
PD API client for fetching player data and ratings.

Author: Claude
Date: 2025-11-01
"""
import logging
import httpx
from typing import Optional, Dict, Any, List
from app.models.player_models import PdPlayer, PositionRating

logger = logging.getLogger(f'{__name__}')

PD_API_BASE = "https://pd.manticorum.com/api/v2"


async def fetch_player_positions(player_id: int) -> List[PositionRating]:
    """
    Fetch all position ratings for a player.

    Args:
        player_id: PD player ID

    Returns:
        List of PositionRating objects

    Raises:
        httpx.HTTPError: If API request fails
    """
    url = f"{PD_API_BASE}/cardpositions/player/{player_id}"

    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        response.raise_for_status()

        data = response.json()

        positions = []
        for pos_data in data.get('positions', []):
            positions.append(PositionRating.from_api_response(pos_data))

        logger.info(f"Loaded {len(positions)} position ratings for player {player_id}")
        return positions

File: backend/app/services/lineup_service.py (create or update)

"""
Lineup management service.

Handles lineup creation, substitutions, and position rating loading.

Author: Claude
Date: 2025-11-01
"""
import logging
import json
from typing import List, Dict
from app.models.player_models import BasePlayer, PdPlayer
from app.models.game_models import Lineup
from app.services.pd_api_client import fetch_player_positions
from app.core.cache import get_player_positions_cache_key
import redis

logger = logging.getLogger(f'{__name__}')

# Redis client (initialized elsewhere)
redis_client: redis.Redis = None  # Set during app startup


async def load_positions_to_cache(
    players: List[BasePlayer],
    league: str
) -> None:
    """
    Load all position ratings for players and cache in Redis.

    For PD players: Fetch from API
    For SBA players: Skip (manual entry only)

    Args:
        players: List of players in lineup
        league: 'pd' or 'sba'
    """
    if league != 'pd':
        logger.debug("SBA league - skipping position rating fetch")
        return

    for player in players:
        if not isinstance(player, PdPlayer):
            continue

        try:
            # Fetch all positions from API
            positions = await fetch_player_positions(player.id)

            # Cache in Redis
            cache_key = get_player_positions_cache_key(player.id)
            positions_json = json.dumps([pos.dict() for pos in positions])

            redis_client.setex(
                cache_key,
                3600 * 24,  # 24 hour TTL
                positions_json
            )

            logger.debug(f"Cached {len(positions)} positions for {player.name}")

        except Exception as e:
            logger.error(f"Failed to load positions for {player.name}: {e}")
            # Continue with other players


async def set_active_position_rating(
    player: BasePlayer,
    position: str
) -> None:
    """
    Set player's active position rating from cache.

    Args:
        player: Player to update
        position: Position code (SS, LF, etc.)
    """
    # Get from cache
    cache_key = get_player_positions_cache_key(player.id)
    cached_data = redis_client.get(cache_key)

    if not cached_data:
        logger.warning(f"No cached positions for player {player.id}")
        return

    # Parse and find position
    positions_data = json.loads(cached_data)
    for pos_data in positions_data:
        if pos_data['position'] == position:
            player.active_position_rating = PositionRating(**pos_data)
            logger.debug(f"Set {player.name} active position to {position}")
            return

    logger.warning(f"Position {position} not found for {player.name}")


async def get_all_player_positions(player_id: int) -> List[PositionRating]:
    """
    Get all position ratings for player from cache.

    Used for substitution UI.

    Args:
        player_id: Player ID

    Returns:
        List of PositionRating objects
    """
    cache_key = get_player_positions_cache_key(player_id)
    cached_data = redis_client.get(cache_key)

    if not cached_data:
        return []

    positions_data = json.loads(cached_data)
    return [PositionRating(**pos) for pos in positions_data]

2. Add X-Check WebSocket Event Handlers

File: backend/app/websocket/game_handlers.py

Add imports:

from app.config.result_charts import PlayOutcome
from app.models.game_models import XCheckResult

Add handler for auto-resolved X-Check result:

async def handle_x_check_auto_result(
    sid: str,
    game_id: int,
    x_check_details: XCheckResult,
    state: GameState
) -> None:
    """
    Broadcast auto-resolved X-Check result to clients.

    Used for PD auto mode and SBA semi-auto mode.
    Shows result with Accept/Reject options.

    Args:
        sid: Socket ID
        game_id: Game ID
        x_check_details: Full resolution details
        state: Current game state
    """
    message = {
        'type': 'x_check_auto_result',
        'game_id': game_id,
        'x_check': x_check_details.to_dict(),
        'state': state.to_dict(),
    }

    await sio.emit('game_update', message, room=f'game_{game_id}')
    logger.info(f"Sent X-Check auto result for game {game_id}")


async def handle_x_check_manual_options(
    sid: str,
    game_id: int,
    position: str,
    d20_roll: int,
    d6_roll: int,
    options: List[Dict[str, str]]
) -> None:
    """
    Broadcast X-Check dice rolls and manual options to clients.

    Used for SBA manual mode (no auto-resolution).

    Args:
        sid: Socket ID
        game_id: Game ID
        position: Position being checked
        d20_roll: Defense table roll
        d6_roll: Error chart roll (3d6 sum)
        options: List of legal outcome options
    """
    message = {
        'type': 'x_check_manual_options',
        'game_id': game_id,
        'position': position,
        'd20': d20_roll,
        'd6': d6_roll,
        'options': options,
    }

    await sio.emit('game_update', message, room=f'game_{game_id}')
    logger.info(f"Sent X-Check manual options for game {game_id}")

Add handler for outcome confirmation:

@sio.on('confirm_x_check_result')
async def confirm_x_check_result(sid: str, data: dict):
    """
    Handle player confirming auto-resolved X-Check result.

    Args:
        data: {
            'game_id': int,
            'accepted': bool,  # True = accept, False = reject
            'override_outcome': Optional[str],  # If rejected, selected outcome
        }
    """
    game_id = data['game_id']
    accepted = data.get('accepted', True)

    # Get game state from memory
    state = get_game_state(game_id)

    if accepted:
        # Apply the auto-resolved result
        logger.info(f"Player accepted auto X-Check result for game {game_id}")
        await apply_play_result(state)

    else:
        # Player rejected - log override and apply their selection
        override_outcome = data.get('override_outcome')
        logger.warning(
            f"Player rejected auto X-Check result for game {game_id}. "
            f"Auto: {state.pending_result.outcome.value}, "
            f"Override: {override_outcome}"
        )

        # TODO: Log to override_log table for dev review
        await log_x_check_override(
            game_id=game_id,
            auto_result=state.pending_result.x_check_details.to_dict(),
            override_outcome=override_outcome
        )

        # Apply override
        await apply_manual_override(state, override_outcome)

    # Broadcast updated state
    await broadcast_game_state(game_id, state)


async def log_x_check_override(
    game_id: int,
    auto_result: dict,
    override_outcome: str
) -> None:
    """
    Log when player overrides auto X-Check result.

    Stored in database for developer review/debugging.

    Args:
        game_id: Game ID
        auto_result: Auto-resolved XCheckResult dict
        override_outcome: Player-selected outcome
    """
    # TODO: Create override_log table and insert record
    logger.warning(
        f"X-Check override logged: game={game_id}, "
        f"auto={auto_result}, override={override_outcome}"
    )

Add handler for manual X-Check submission:

@sio.on('submit_x_check_manual')
async def submit_x_check_manual(sid: str, data: dict):
    """
    Handle manual X-Check outcome submission.

    Used for SBA manual mode - player reads charts and submits result.

    Args:
        data: {
            'game_id': int,
            'outcome': str,  # e.g., 'SI2_E1', 'G1_NO', 'F2_RP'
        }
    """
    game_id = data['game_id']
    outcome_str = data['outcome']

    # Parse outcome string (e.g., 'SI2_E1' → base='SI2', error='E1')
    parts = outcome_str.split('_')
    base_result = parts[0]
    error_result = parts[1] if len(parts) > 1 else 'NO'

    logger.info(
        f"Manual X-Check submission: game={game_id}, "
        f"base={base_result}, error={error_result}"
    )

    # Get game state
    state = get_game_state(game_id)

    # Build XCheckResult from manual input
    # (We already have d20/d6 rolls from previous event)
    x_check_details = state.pending_x_check  # Stored from dice roll event

    x_check_details.base_result = base_result
    x_check_details.error_result = error_result

    # Determine final outcome
    final_outcome, hit_type = PlayResolver._determine_final_x_check_outcome(
        converted_result=base_result,
        error_result=error_result
    )

    x_check_details.final_outcome = final_outcome
    x_check_details.hit_type = hit_type

    # Get advancement
    advancement = PlayResolver._get_x_check_advancement(
        converted_result=base_result,
        error_result=error_result,
        on_base_code=state.get_on_base_code(),
        defender_in=False  # TODO: Get from state
    )

    # Create PlayResult
    play_result = PlayResult(
        outcome=final_outcome,
        advancement=advancement,
        x_check_details=x_check_details,
        outs_recorded=1 if final_outcome.is_out() and error_result == 'NO' else 0,
    )

    # Apply to game state
    await apply_play_result(state, play_result)

    # Broadcast
    await broadcast_game_state(game_id, state)

3. Generate Legal Options for Manual Mode

File: backend/app/core/x_check_options.py (NEW FILE)

"""
Generate legal X-Check outcome options for manual mode.

Given dice rolls and position, generates list of valid outcomes
player can select.

Author: Claude
Date: 2025-11-01
"""
import logging
from typing import List, Dict
from app.config.common_x_check_tables import (
    INFIELD_DEFENSE_TABLE,
    OUTFIELD_DEFENSE_TABLE,
    CATCHER_DEFENSE_TABLE,
    get_error_chart_for_position,
)

logger = logging.getLogger(f'{__name__}')


def generate_x_check_options(
    position: str,
    d20_roll: int,
    d6_roll: int,
    defense_range: int,
    error_rating: int
) -> List[Dict[str, str]]:
    """
    Generate legal outcome options for manual X-Check.

    Args:
        position: Position code (SS, LF, etc.)
        d20_roll: Defense table roll (1-20)
        d6_roll: Error chart roll (3-18)
        defense_range: Defender's range (1-5)
        error_rating: Defender's error rating (0-25)

    Returns:
        List of option dicts: [
            {'value': 'SI2_NO', 'label': 'Single (no error)'},
            {'value': 'SI2_E1', 'label': 'Single + Error (1 base)'},
            ...
        ]
    """
    options = []

    # Get base result from defense table
    base_result = _lookup_defense_table(position, d20_roll, defense_range)

    # Get possible error results from error chart
    error_results = _get_possible_errors(position, d6_roll, error_rating)

    # Generate option for each combination
    for error in error_results:
        option = {
            'value': f"{base_result}_{error}",
            'label': _format_option_label(base_result, error)
        }
        options.append(option)

    logger.debug(f"Generated {len(options)} options for {position} X-Check")
    return options


def _lookup_defense_table(position: str, d20: int, range: int) -> str:
    """Lookup base result from defense table."""
    if position in ['P', 'C', '1B', '2B', '3B', 'SS']:
        table = CATCHER_DEFENSE_TABLE if position == 'C' else INFIELD_DEFENSE_TABLE
    else:
        table = OUTFIELD_DEFENSE_TABLE

    return table[d20 - 1][range - 1]


def _get_possible_errors(position: str, d6: int, error_rating: int) -> List[str]:
    """Get list of possible error results for this roll."""
    chart = get_error_chart_for_position(position)

    if error_rating not in chart:
        error_rating = 0

    rating_row = chart[error_rating]

    errors = ['NO']  # Always an option

    # Check each error type
    for error_type in ['RP', 'E3', 'E2', 'E1']:
        if d6 in rating_row[error_type]:
            errors.append(error_type)

    return errors


def _format_option_label(base_result: str, error: str) -> str:
    """Format human-readable label for option."""
    base_labels = {
        'SI1': 'Single',
        'SI2': 'Single',
        'DO2': 'Double (to 2nd)',
        'DO3': 'Double (to 3rd)',
        'TR3': 'Triple',
        'G1': 'Groundout',
        'G2': 'Groundout',
        'G3': 'Groundout',
        'F1': 'Flyout (deep)',
        'F2': 'Flyout (medium)',
        'F3': 'Flyout (shallow)',
        'FO': 'Foul Out',
        'PO': 'Pop Out',
        'SPD': 'Speed Test',
    }

    error_labels = {
        'NO': 'no error',
        'E1': 'Error (1 base)',
        'E2': 'Error (2 bases)',
        'E3': 'Error (3 bases)',
        'RP': 'Rare Play',
    }

    base = base_labels.get(base_result, base_result)
    err = error_labels.get(error, error)

    if error == 'NO':
        return f"{base} ({err})"
    else:
        return f"{base} + {err}"

4. Update Game Flow to Trigger X-Check Events

File: backend/app/core/game_engine.py

Add method to handle X-Check outcome:

async def process_x_check_outcome(
    self,
    state: GameState,
    position: str,
    mode: str  # 'auto', 'manual', or 'semi_auto'
) -> None:
    """
    Process X-Check outcome based on game mode.

    Args:
        state: Current game state
        position: Position being checked
        mode: Resolution mode
    """
    if mode == 'auto':
        # PD Auto: Resolve completely and send Accept/Reject
        result = await self.resolver.resolve_x_check_auto(state, position)

        # Store pending result
        state.pending_result = result

        # Broadcast with Accept/Reject UI
        await handle_x_check_auto_result(
            sid=None,
            game_id=state.game_id,
            x_check_details=result.x_check_details,
            state=state
        )

    elif mode == 'manual':
        # SBA Manual: Roll dice and send options
        d20 = self.dice.roll_d20()
        d6 = self.dice.roll_3d6()

        # Store rolls for later use
        state.pending_x_check = {
            'position': position,
            'd20': d20,
            'd6': d6,
        }

        # Generate options (requires defense/error ratings)
        # For SBA, player provides ratings or we use defaults
        options = generate_x_check_options(
            position=position,
            d20_roll=d20,
            d6_roll=d6,
            defense_range=3,  # Default or from player input
            error_rating=10,  # Default or from player input
        )

        await handle_x_check_manual_options(
            sid=None,
            game_id=state.game_id,
            position=position,
            d20_roll=d20,
            d6_roll=d6,
            options=options
        )

    elif mode == 'semi_auto':
        # SBA Semi-Auto: Like auto but show charts too
        # Same as auto mode but with additional UI context
        await self.process_x_check_outcome(state, position, 'auto')

Testing Requirements

1. Unit Tests: tests/services/test_lineup_service.py

   • Test load_positions_to_cache()
   • Test set_active_position_rating()
   • Test get_all_player_positions()

2. Unit Tests: tests/core/test_x_check_options.py

   • Test generate_x_check_options()
   • Test _get_possible_errors()
   • Test _format_option_label()

3. Integration Tests: tests/websocket/test_x_check_events.py

   • Test full auto flow (PD)
   • Test full manual flow (SBA)
   • Test Accept/Reject flow
   • Test override logging

Acceptance Criteria

• PD API client implemented for fetching positions
• Lineup service caches positions in Redis
• Active position rating loaded on defensive positioning
• X-Check auto result event handler working
• X-Check manual options event handler working
• Confirm result handler with Accept/Reject working
• Manual submission handler working
• Override logging implemented
• Option generation working
• All unit tests pass
• All integration tests pass

Notes

• Redis client must be initialized during app startup
• Position ratings cached for 24 hours
• Override log needs database table (add migration)
• SPD test needs special option generation (conditional)
• Charts should be sent to frontend for PD manual mode

Next Phase

After completion, proceed to Phase 3F: Testing & Integration