""" DungeonMaster AI - Game-Aware Tool Wrappers Wraps MCP tools with game state awareness and side effects. Provides enhanced results formatted for chat, UI, and voice. """ from __future__ import annotations import inspect import logging from collections.abc import Awaitable, Callable, Sequence from datetime import datetime from functools import wraps from typing import TYPE_CHECKING from llama_index.core.tools import FunctionTool from src.utils.formatters import ( format_dice_roll, format_hp_change, format_initiative_order, ) from .models import ( CombatantInfo, CombatStateResult, DiceRollResult, FormattedResult, GameStateProtocol, HPChangeResult, RollType, ) if TYPE_CHECKING: from typing import Any from .toolkit_client import TTRPGToolkitClient logger = logging.getLogger(__name__) # Type aliases for callbacks SessionLogger = Callable[[str, str, dict[str, object]], Awaitable[None]] UINotifier = Callable[[str, dict[str, object]], None] # ============================================================================= # Voice Formatting Helpers # ============================================================================= def format_roll_for_voice( total: int, roll_type: RollType, is_critical: bool = False, is_fumble: bool = False, success: bool | None = None, skill_name: str | None = None, ) -> str: """ Format dice roll for TTS narration. Args: total: Roll total roll_type: Type of roll is_critical: Natural 20 is_fumble: Natural 1 success: Success/failure for checks skill_name: Name of skill if applicable Returns: TTS-friendly text """ # Critical/fumble announcements if is_critical: return f"Natural twenty! Critical success with a total of {total}!" if is_fumble: return "Natural one. Critical failure." # Base narration by roll type skill_part = f" for {skill_name}" if skill_name else "" templates = { RollType.ATTACK: f"Attack roll{skill_part}, {total}.", RollType.DAMAGE: f"{total} points of damage.", RollType.SAVE: f"Saving throw{skill_part}, {total}.", RollType.CHECK: f"Ability check{skill_part}, you rolled a {total}.", RollType.INITIATIVE: f"Initiative, {total}.", RollType.STANDARD: f"You rolled {total}.", } base = templates.get(roll_type, f"Result: {total}") # Add success/failure if success is not None: base += " Success!" if success else " Failed." return base def format_hp_for_voice( character_name: str, change_amount: int, current_hp: int, max_hp: int, is_damage: bool, is_unconscious: bool = False, is_dead: bool = False, ) -> str: """ Format HP change for TTS narration. Args: character_name: Name of character change_amount: Absolute value of HP change current_hp: HP after change max_hp: Maximum HP is_damage: True if damage, False if healing is_unconscious: Character is at 0 HP is_dead: Character died from massive damage Returns: TTS-friendly text """ if is_dead: return f"{character_name} is slain by massive damage!" if is_unconscious and is_damage: return f"{character_name} takes {change_amount} damage and falls unconscious!" if is_damage: if current_hp <= max_hp // 4: return f"{character_name} takes {change_amount} damage and is badly wounded!" return f"{character_name} takes {change_amount} damage." # Healing if current_hp >= max_hp: return f"{character_name} is fully healed!" return f"{character_name} recovers {change_amount} hit points." def format_combat_turn_for_voice( combatant_name: str, is_player: bool, round_number: int, ) -> str: """ Format combat turn announcement for TTS. Args: combatant_name: Name of current combatant is_player: Whether combatant is a player character round_number: Current round number Returns: TTS-friendly text """ if is_player: return f"Round {round_number}. It's your turn, {combatant_name}. What do you do?" return f"{combatant_name} takes their turn." # ============================================================================= # Roll Type Detection # ============================================================================= def detect_roll_type( _tool_name: str, arguments: dict[str, object], result: dict[str, object], ) -> RollType: """ Detect the type of roll from context. Args: tool_name: Name of the tool called arguments: Tool arguments result: Tool result Returns: Detected RollType """ # Check explicit roll_type in arguments if "roll_type" in arguments: try: return RollType(arguments["roll_type"]) except ValueError: pass # Check check_type in result (from roll_check) check_type = str(result.get("check_type", "")).lower() if "attack" in check_type: return RollType.ATTACK if "save" in check_type or "saving" in check_type: return RollType.SAVE if "initiative" in check_type: return RollType.INITIATIVE # Check reason field for hints reason = str(arguments.get("reason", "")).lower() if "attack" in reason: return RollType.ATTACK if "damage" in reason: return RollType.DAMAGE if "save" in reason or "saving" in reason: return RollType.SAVE if "check" in reason or "ability" in reason: return RollType.CHECK if "initiative" in reason: return RollType.INITIATIVE return RollType.STANDARD # ============================================================================= # Game Aware Tools Wrapper Class # ============================================================================= class GameAwareTools: """ Wraps MCP tools with game state awareness and side effects. This class enhances raw MCP tools to: 1. Automatically log actions to the game session 2. Update game state after tool calls 3. Format results for chat, UI, and voice 4. Trigger appropriate UI updates 5. Handle special game logic (death saves, combat transitions) Example: ```python # Get raw tools from MCP raw_tools = await toolkit_client.get_all_tools() # Wrap with game awareness wrapper = GameAwareTools( game_state=game_state, toolkit_client=toolkit_client, session_logger=log_session_event, ui_notifier=notify_ui, ) enhanced_tools = wrapper.wrap_tools(raw_tools) # Use with LlamaIndex agent agent = FunctionAgent(tools=enhanced_tools, ...) ``` """ # Tool name sets for category detection ROLL_TOOLS = {"roll", "roll_check", "roll_table", "mcp_roll", "mcp_roll_check"} HP_TOOLS = {"modify_hp", "mcp_modify_hp"} COMBAT_START_TOOLS = {"start_combat", "mcp_start_combat"} COMBAT_END_TOOLS = {"end_combat", "mcp_end_combat"} COMBAT_TURN_TOOLS = {"next_turn", "mcp_next_turn"} def __init__( self, game_state: GameStateProtocol | None = None, toolkit_client: TTRPGToolkitClient | None = None, session_logger: SessionLogger | None = None, ui_notifier: UINotifier | None = None, ) -> None: """ Initialize GameAwareTools. Args: game_state: Reference to game state for updates (optional) toolkit_client: MCP toolkit client for additional calls (optional) session_logger: Async callback for logging events to session ui_notifier: Callback for notifying UI of changes """ self._game_state = game_state self._toolkit_client = toolkit_client self._session_logger = session_logger self._ui_notifier = ui_notifier async def _safe_call_logger( self, event_type: str, description: str, data: dict[str, object], ) -> None: """ Safely call session logger, handling both sync and async callbacks. This method ensures the logger is called correctly regardless of whether it returns an awaitable or not, preventing "NoneType can't be used in 'await' expression" errors. """ if not self._session_logger: return try: result = self._session_logger(event_type, description, data) if inspect.isawaitable(result): await result except Exception as e: logger.warning(f"Session logger call failed: {e}") def wrap_tools( self, tools: Sequence[FunctionTool], ) -> list[FunctionTool]: """ Wrap a list of MCP tools with game-aware enhancements. Args: tools: List of raw MCP FunctionTools Returns: List of enhanced FunctionTools """ wrapped_tools = [] for tool in tools: tool_name = tool.metadata.name or "" normalized_name = tool_name.replace("mcp_", "") # Select appropriate wrapper based on tool category if tool_name in self.ROLL_TOOLS or normalized_name in self.ROLL_TOOLS: wrapped = self._wrap_roll_tool(tool) elif tool_name in self.HP_TOOLS or normalized_name in self.HP_TOOLS: wrapped = self._wrap_hp_tool(tool) elif ( tool_name in self.COMBAT_START_TOOLS or normalized_name in self.COMBAT_START_TOOLS ): wrapped = self._wrap_combat_start_tool(tool) elif ( tool_name in self.COMBAT_END_TOOLS or normalized_name in self.COMBAT_END_TOOLS ): wrapped = self._wrap_combat_end_tool(tool) elif ( tool_name in self.COMBAT_TURN_TOOLS or normalized_name in self.COMBAT_TURN_TOOLS ): wrapped = self._wrap_combat_turn_tool(tool) else: wrapped = self._wrap_generic(tool) wrapped_tools.append(wrapped) logger.debug(f"Wrapped {len(wrapped_tools)} tools with game awareness") return wrapped_tools def _wrap_roll_tool(self, tool: FunctionTool) -> FunctionTool: """Wrap dice rolling tool with game-aware enhancements.""" original_fn = tool.async_fn if tool.async_fn is not None else tool.fn @wraps(original_fn) async def wrapped_roll(**kwargs: Any) -> DiceRollResult: # Call original tool if tool.async_fn is not None: raw_result = await tool.async_fn(**kwargs) else: raw_result = tool.fn(**kwargs) # Extract result dict result_dict = self._extract_result_dict(raw_result) # Detect roll type roll_type = detect_roll_type(tool.metadata.name or "", kwargs, result_dict) # Extract roll data notation = str(result_dict.get("notation", kwargs.get("notation", ""))) rolls = result_dict.get("rolls", result_dict.get("dice", [])) if not isinstance(rolls, list): rolls = [] modifier = int(str(result_dict.get("modifier", 0))) total = int(str(result_dict.get("total", 0))) dc = result_dict.get("dc") success = result_dict.get("success", result_dict.get("check_success")) # Check for crits (d20 only) is_d20 = "d20" in notation.lower() or roll_type in ( RollType.ATTACK, RollType.SAVE, RollType.CHECK, ) is_critical = is_d20 and len(rolls) >= 1 and 20 in rolls is_fumble = is_d20 and len(rolls) >= 1 and 1 in rolls and 20 not in rolls # Format for display chat_display = format_dice_roll( notation=notation, individual_rolls=rolls, modifier=modifier, total=total, is_check=roll_type in (RollType.CHECK, RollType.SAVE), dc=int(str(dc)) if dc is not None else None, success=bool(success) if success is not None else None, ) # Format for UI ui_data = { "notation": notation, "rolls": rolls, "modifier": modifier, "total": total, "roll_type": roll_type.value, "is_critical": is_critical, "is_fumble": is_fumble, "success": success, "dc": dc, "timestamp": datetime.now().isoformat(), } # Format for voice voice_narration = format_roll_for_voice( total=total, roll_type=roll_type, is_critical=is_critical, is_fumble=is_fumble, success=bool(success) if success is not None else None, skill_name=kwargs.get("skill_name"), ) # Create enhanced result enhanced_result = DiceRollResult( raw_result=result_dict, chat_display=chat_display, ui_data=ui_data, voice_narration=voice_narration, notation=notation, individual_rolls=rolls, modifier=modifier, total=total, roll_type=roll_type, is_critical=is_critical, is_fumble=is_fumble, success=bool(success) if success is not None else None, dc=int(str(dc)) if dc is not None else None, reason=str(kwargs.get("reason", "")), ui_updates_needed=["dice_panel", "roll_history"], ) # Side effects await self._log_roll_event(enhanced_result, kwargs) self._add_roll_to_game_state(enhanced_result) return enhanced_result return FunctionTool.from_defaults( async_fn=wrapped_roll, name=tool.metadata.name, description=tool.metadata.description, ) def _wrap_hp_tool(self, tool: FunctionTool) -> FunctionTool: """Wrap HP modification tool with death handling.""" original_fn = tool.async_fn if tool.async_fn is not None else tool.fn @wraps(original_fn) async def wrapped_modify_hp(**kwargs: Any) -> HPChangeResult: # Get character ID and amount character_id = str(kwargs.get("character_id", "")) amount = int(kwargs.get("amount", 0)) damage_type = kwargs.get("damage_type") # Get previous HP from cache if available prev_hp = 0 max_hp = 1 char_name = "Unknown" if self._game_state: char_data = self._game_state.get_character(character_id) if char_data: hp_data = char_data.get("hit_points") if isinstance(hp_data, dict): prev_hp = int(str(hp_data.get("current", 0))) max_hp = int(str(hp_data.get("maximum", 1))) char_name = str(char_data.get("name", "Unknown")) # Call original tool if tool.async_fn is not None: raw_result = await tool.async_fn(**kwargs) else: raw_result = tool.fn(**kwargs) result_dict = self._extract_result_dict(raw_result) # Get new HP from result current_hp_val = result_dict.get( "current_hp", result_dict.get("new_hp", prev_hp + amount) ) current_hp = int(str(current_hp_val)) if "max_hp" in result_dict: max_hp = int(str(result_dict["max_hp"])) if "character_name" in result_dict: char_name = str(result_dict["character_name"]) is_damage = amount < 0 change_amount = abs(amount) # Death checks is_unconscious = current_hp <= 0 requires_death_save = is_unconscious and prev_hp > 0 and is_damage is_dead = is_damage and current_hp <= -max_hp # Massive damage # Bloodied check (50% HP) is_bloodied = 0 < current_hp <= max_hp // 2 # Format for display chat_display = format_hp_change( character_name=char_name, previous_hp=prev_hp, current_hp=max(0, current_hp), max_hp=max_hp, is_damage=is_damage, ) if is_dead: chat_display += ( "\n**DEATH!** Massive damage has killed the character instantly!" ) elif requires_death_save: chat_display += "\n*Death saving throws begin next turn.*" # Format for UI ui_data = { "character_id": character_id, "character_name": char_name, "hp_previous": prev_hp, "hp_current": current_hp, "hp_max": max_hp, "hp_percent": (max(0, current_hp) / max_hp * 100) if max_hp > 0 else 0, "is_unconscious": is_unconscious, "is_bloodied": is_bloodied, "is_critical": 0 < current_hp <= max_hp // 4, "is_dead": is_dead, } # Format for voice voice_narration = format_hp_for_voice( character_name=char_name, change_amount=change_amount, current_hp=max(0, current_hp), max_hp=max_hp, is_damage=is_damage, is_unconscious=is_unconscious, is_dead=is_dead, ) # Create enhanced result ui_updates = ["character_sheet"] if self._game_state and self._game_state.in_combat: ui_updates.append("combat_tracker") enhanced_result = HPChangeResult( raw_result=result_dict, chat_display=chat_display, ui_data=ui_data, voice_narration=voice_narration, character_id=character_id, character_name=char_name, previous_hp=prev_hp, current_hp=current_hp, max_hp=max_hp, change_amount=change_amount, is_damage=is_damage, damage_type=str(damage_type) if damage_type else None, is_unconscious=is_unconscious, requires_death_save=requires_death_save, is_dead=is_dead, is_bloodied=is_bloodied, ui_updates_needed=ui_updates, ) # Side effects await self._log_hp_event(enhanced_result) self._update_character_hp(enhanced_result) return enhanced_result return FunctionTool.from_defaults( async_fn=wrapped_modify_hp, name=tool.metadata.name, description=tool.metadata.description, ) def _wrap_combat_start_tool(self, tool: FunctionTool) -> FunctionTool: """Wrap combat start tool.""" original_fn = tool.async_fn if tool.async_fn is not None else tool.fn @wraps(original_fn) async def wrapped_start_combat(**kwargs: Any) -> CombatStateResult: # Call original tool if tool.async_fn is not None: raw_result = await tool.async_fn(**kwargs) else: raw_result = tool.fn(**kwargs) result_dict = self._extract_result_dict(raw_result) # Extract combat state turn_order_raw = result_dict.get("turn_order", []) turn_order: list[object] = list(turn_order_raw) if isinstance(turn_order_raw, list) else [] current_idx = int(str(result_dict.get("current_turn_index", 0))) round_num = int(str(result_dict.get("round", 1))) # Build combatant info list combatants: list[CombatantInfo] = [] for i, combatant in enumerate(turn_order): if isinstance(combatant, dict): hp_curr = int(str(combatant.get("hp_current", 0))) hp_max = int(str(combatant.get("hp_max", 1))) combatants.append( CombatantInfo( id=str(combatant.get("id", f"combatant_{i}")), name=str(combatant.get("name", f"Combatant {i + 1}")), initiative=int(str(combatant.get("initiative", 0))), is_player=bool(combatant.get("is_player", False)), is_current=i == current_idx, hp_current=hp_curr, hp_max=hp_max, hp_percent=(hp_curr / hp_max * 100) if hp_max > 0 else 0, conditions=list(combatant.get("conditions", [])), status=self._get_hp_status(hp_curr, hp_max), ) ) current_combatant = ( combatants[current_idx].name if combatants else "Unknown" ) is_player = combatants[current_idx].is_player if combatants else False # Format for display chat_display = "**Combat Begins!**\n\n" if combatants: chat_display += format_initiative_order( [c.model_dump() for c in combatants], current_idx, ) # Format for UI ui_data = { "round": round_num, "combatants": [c.model_dump() for c in combatants], "current_combatant_name": current_combatant, "combat_started": True, } # Format for voice voice_narration = ( f"Roll for initiative! Combat begins. " f"{current_combatant} goes first." ) enhanced_result = CombatStateResult( raw_result=result_dict, chat_display=chat_display, ui_data=ui_data, voice_narration=voice_narration, action="start", round_number=round_num, current_combatant=current_combatant, current_combatant_is_player=is_player, turn_order=combatants, ui_updates_needed=["combat_tracker"], state_updated=True, ) # Side effects if self._game_state: self._game_state.set_combat_state(result_dict) if self._ui_notifier: self._ui_notifier("show_combat_tracker", ui_data) await self._log_combat_event("combat_start", result_dict) return enhanced_result return FunctionTool.from_defaults( async_fn=wrapped_start_combat, name=tool.metadata.name, description=tool.metadata.description, ) def _wrap_combat_end_tool(self, tool: FunctionTool) -> FunctionTool: """Wrap combat end tool.""" original_fn = tool.async_fn if tool.async_fn is not None else tool.fn @wraps(original_fn) async def wrapped_end_combat(**kwargs: Any) -> CombatStateResult: # Call original tool if tool.async_fn is not None: raw_result = await tool.async_fn(**kwargs) else: raw_result = tool.fn(**kwargs) result_dict = self._extract_result_dict(raw_result) # Format for display chat_display = "**Combat Ends!**\n\n*The dust settles...*" # Format for UI ui_data: dict[str, object] = {"combat_ended": True} # Format for voice voice_narration = "The battle is over." enhanced_result = CombatStateResult( raw_result=result_dict, chat_display=chat_display, ui_data=ui_data, voice_narration=voice_narration, action="end", combat_ended=True, ui_updates_needed=["combat_tracker"], state_updated=True, ) # Side effects if self._game_state: self._game_state.set_combat_state(None) if self._ui_notifier: self._ui_notifier("hide_combat_tracker", {}) await self._log_combat_event("combat_end", result_dict) return enhanced_result return FunctionTool.from_defaults( async_fn=wrapped_end_combat, name=tool.metadata.name, description=tool.metadata.description, ) def _wrap_combat_turn_tool(self, tool: FunctionTool) -> FunctionTool: """Wrap next turn tool.""" original_fn = tool.async_fn if tool.async_fn is not None else tool.fn @wraps(original_fn) async def wrapped_next_turn(**kwargs: Any) -> CombatStateResult: # Call original tool if tool.async_fn is not None: raw_result = await tool.async_fn(**kwargs) else: raw_result = tool.fn(**kwargs) result_dict = self._extract_result_dict(raw_result) # Extract updated combat state turn_order_raw = result_dict.get("turn_order", []) turn_order: list[object] = list(turn_order_raw) if isinstance(turn_order_raw, list) else [] current_idx = int(str(result_dict.get("current_turn_index", 0))) round_num = int(str(result_dict.get("round", 1))) # Build combatant info combatants: list[CombatantInfo] = [] for i, combatant in enumerate(turn_order): if isinstance(combatant, dict): hp_curr = int(str(combatant.get("hp_current", 0))) hp_max = int(str(combatant.get("hp_max", 1))) combatants.append( CombatantInfo( id=str(combatant.get("id", f"combatant_{i}")), name=str(combatant.get("name", f"Combatant {i + 1}")), initiative=int(str(combatant.get("initiative", 0))), is_player=bool(combatant.get("is_player", False)), is_current=i == current_idx, hp_current=hp_curr, hp_max=hp_max, hp_percent=(hp_curr / hp_max * 100) if hp_max > 0 else 0, conditions=list(combatant.get("conditions", [])), status=self._get_hp_status(hp_curr, hp_max), ) ) current_combatant = ( combatants[current_idx].name if combatants else "Unknown" ) is_player = combatants[current_idx].is_player if combatants else False # Format for display chat_display = format_initiative_order( [c.model_dump() for c in combatants], current_idx, ) # Format for UI ui_data = { "round": round_num, "combatants": [c.model_dump() for c in combatants], "current_combatant_name": current_combatant, } # Format for voice voice_narration = format_combat_turn_for_voice( combatant_name=current_combatant, is_player=is_player, round_number=round_num, ) enhanced_result = CombatStateResult( raw_result=result_dict, chat_display=chat_display, ui_data=ui_data, voice_narration=voice_narration, action="next_turn", round_number=round_num, current_combatant=current_combatant, current_combatant_is_player=is_player, turn_order=combatants, ui_updates_needed=["combat_tracker"], state_updated=True, ) # Side effects if self._game_state: self._game_state.set_combat_state(result_dict) return enhanced_result return FunctionTool.from_defaults( async_fn=wrapped_next_turn, name=tool.metadata.name, description=tool.metadata.description, ) def _wrap_generic(self, tool: FunctionTool) -> FunctionTool: """ Generic wrapper for tools that don't need special handling. Still adds basic logging. """ original_fn = tool.async_fn if tool.async_fn is not None else tool.fn @wraps(original_fn) async def wrapped_generic(**kwargs: Any) -> FormattedResult: # Call original tool if tool.async_fn is not None: raw_result = await tool.async_fn(**kwargs) else: raw_result = tool.fn(**kwargs) result_dict = self._extract_result_dict(raw_result) # Create basic formatted result result = FormattedResult( raw_result=result_dict, chat_display=str(result_dict.get("formatted", str(raw_result))), ui_data=result_dict, voice_narration="", ) # Log generic tool call await self._safe_call_logger( "tool_call", f"Called {tool.metadata.name}", {"tool": tool.metadata.name, "args": kwargs}, ) return result return FunctionTool.from_defaults( async_fn=wrapped_generic, name=tool.metadata.name, description=tool.metadata.description, ) # ============================================================================= # Helper Methods # ============================================================================= def _extract_result_dict(self, raw_result: Any) -> dict[str, object]: """Extract dict from various result types.""" if isinstance(raw_result, dict): return raw_result # Handle ToolOutput if hasattr(raw_result, "raw_output"): output = raw_result.raw_output if isinstance(output, dict): return output return {"result": output} # Handle Pydantic models if hasattr(raw_result, "model_dump"): dumped = raw_result.model_dump() if isinstance(dumped, dict): return dumped return {"result": dumped} return {"result": raw_result} def _get_hp_status(self, hp_current: int, hp_max: int) -> str: """Get status string from HP values.""" if hp_current <= 0: return "down" if hp_current <= hp_max // 4: return "critical" if hp_current <= hp_max // 2: return "wounded" return "healthy" async def _log_roll_event( self, result: DiceRollResult, kwargs: dict[str, Any], ) -> None: """Log roll event to session.""" event_data = { "notation": result.notation, "total": result.total, "rolls": result.individual_rolls, "roll_type": result.roll_type.value, "reason": kwargs.get("reason", ""), } if result.is_critical: event_data["critical"] = True if result.is_fumble: event_data["fumble"] = True await self._safe_call_logger( "dice_roll", f"Rolled {result.notation} = {result.total}", event_data, ) result.events_logged.append("dice_roll") def _add_roll_to_game_state(self, result: DiceRollResult) -> None: """Add roll to game state recent events.""" if not self._game_state: return self._game_state.add_event( event_type="roll", description=f"Roll: {result.notation} = {result.total}", data={ "notation": result.notation, "total": result.total, "roll_type": result.roll_type.value, "is_critical": result.is_critical, "is_fumble": result.is_fumble, }, ) result.state_updated = True async def _log_hp_event(self, result: HPChangeResult) -> None: """Log HP change event to session.""" event_type = "damage" if result.is_damage else "healing" description = ( f"{result.character_name}: " f"{result.previous_hp} -> {result.current_hp} HP" ) await self._safe_call_logger( event_type, description, { "character_id": result.character_id, "change": -result.change_amount if result.is_damage else result.change_amount, "is_unconscious": result.is_unconscious, "is_dead": result.is_dead, }, ) result.events_logged.append(event_type) def _update_character_hp(self, result: HPChangeResult) -> None: """Update character HP in game state cache.""" if not self._game_state: return # Update cache char_data = self._game_state.get_character(result.character_id) if char_data: hp_data = char_data.get("hit_points") if not isinstance(hp_data, dict): hp_data = {} char_data["hit_points"] = hp_data hp_data["current"] = result.current_hp self._game_state.update_character_cache(result.character_id, char_data) # Log event self._game_state.add_event( event_type="hp_change", description=( f"{result.character_name} " f"{'takes' if result.is_damage else 'heals'} " f"{result.change_amount}" ), data=result.ui_data, ) result.state_updated = True async def _log_combat_event( self, event_type: str, data: dict[str, object], ) -> None: """Log combat event to session.""" await self._safe_call_logger( event_type, f"Combat {event_type.replace('combat_', '')}", data, )