import uuid from sqlalchemy.ext.asyncio import AsyncSession from app.models.game import Game TOOL_SCHEMA = { "name": "update_monster_hp", "description": ( "Track an NPC's/monster's hit points during a fight — call this whenever one is introduced " "or takes damage, so the server can detect when it's defeated. This is DM-only bookkeeping, " "never shown to players; keep narrating combat and secrecy (e.g. AC) exactly as before." ), "parameters": { "type": "object", "properties": { "name": { "type": "string", "description": "The monster's/NPC's name, used consistently across calls to identify it.", }, "current_hp": {"type": "integer"}, "max_hp": {"type": "integer"}, }, "required": ["name", "current_hp", "max_hp"], }, } async def update(session: AsyncSession, game_id: uuid.UUID, tool_input: dict) -> dict: game = await session.get(Game, game_id) if game is None: raise ValueError(f"Game {game_id} not found") name = tool_input["name"] current_hp = tool_input["current_hp"] max_hp = tool_input["max_hp"] # Build an entirely new list of new dicts rather than mutating existing ones in place: # SQLAlchemy compares the old and new JSONB value by content to decide whether a flush is # needed, and an in-place mutation makes the "old" value (still referenced by the attribute) # look identical to the "new" one by the time of that comparison, so the UPDATE silently # gets skipped. updated_entry = {"name": name, "current_hp": current_hp, "max_hp": max_hp} monsters = [dict(m) for m in (game.monsters or []) if m.get("name") != name] monsters.append(updated_entry) game.monsters = monsters await session.commit() return {"name": name, "current_hp": current_hp, "max_hp": max_hp, "defeated": current_hp <= 0}