# pyright: reportOptionalSubscript=false, reportOptionalMemberAccess=false # pyright: reportCallIssue=false # Tests subscript load_character / ctx.initiative._find results without # null-narrowing — setup guarantees the values exist. """Tests for tool dispatch via the FastMCP in-memory client. These tests exercise the same call path the production server uses (claude → MCP → tool function), but in-process via fastmcp.Client. """ from pathlib import Path import pytest from conftest import McpCall, McpCallInCombat from storied.character import load_character from storied.initiative import Combatant from storied.testing import call_tool from storied.tools import ToolContext from storied.tools.entities import note_discovery as _note_discovery from storied.tools.scene import end_session as _end_session # --- Dispatch through the in-memory client ---------------------------------- class TestToolDispatch: def test_roll_with_modifier(self, ctx: ToolContext, mcp_call: McpCall): result = mcp_call("roll", {"notation": "1d20+5", "reason": "attack"}) assert "Rolled" in result assert "1d20+5" in result def test_roll_no_modifier(self, ctx: ToolContext, mcp_call: McpCall): result = mcp_call("roll", {"notation": "1d6", "reason": "damage"}) assert "Rolled" in result def test_recall_finds_nothing(self, ctx: ToolContext, mcp_call: McpCall): result = mcp_call("recall", {"query": "nonexistent"}) assert "Nothing found" in result def test_set_scene(self, ctx: ToolContext, mcp_call: McpCall): result = mcp_call( "set_scene", {"event": "Spoke with guards", "duration": "10 min"} ) assert result def test_establish(self, ctx: ToolContext, mcp_call: McpCall): result = mcp_call("establish", {"entity_type": "npcs", "name": "Test NPC"}) assert "Established" in result def test_mark(self, ctx: ToolContext, mcp_call: McpCall): mcp_call("establish", {"entity_type": "npcs", "name": "Vera"}) result = mcp_call( "mark", { "entity_type": "npcs", "name": "Vera", "event": "Revealed her secret", }, ) assert "Marked" in result def test_note_discovery(self, ctx: ToolContext, mcp_call: McpCall): result = mcp_call( "note_discovery", { "entity": "Vera Blackwater", "content": "She used to be a smuggler", }, ) assert "Noted" in result def test_end_session(self, ctx: ToolContext, mcp_call: McpCall): result = mcp_call("end_session", {"situation": "In the tavern"}) assert result == "SESSION_ENDED" def test_unknown_tool_raises(self, ctx: ToolContext, mcp_call: McpCall): with pytest.raises(Exception, match="nonexistent"): mcp_call("nonexistent", {}) class TestUpdateCharacter: def test_landed_in_state_hp_current( self, ctx: ToolContext, tmp_path: Path, mcp_call: McpCall ): mcp_call( "create_character", { "name": "Test", "race": "Human", "char_class": "Fighter", "level": 1, "abilities": { "strength": 16, "dexterity": 12, "constitution": 14, "intelligence": 10, "wisdom": 13, "charisma": 8, }, "hp_max": 12, "ac": 16, }, ) result = mcp_call("update_character", {"updates": {"state.hp.current": 8}}) assert "updated" in result.lower() data = load_character(ctx.player_id) assert data["state"]["hp"]["current"] == 8, ( "update_character should write to state.hp.current with the new schema, " f"but state.hp.current is {data['state']['hp']['current']}" ) # --- Sub-tool helper coverage ----------------------------------------------- class TestNoteDiscoveryDirect: """Direct (non-MCP) calls to note_discovery exercise the wrapper itself.""" def test_creates_knowledge_file(self, ctx: ToolContext, tmp_path: Path): call_tool( _note_discovery, entity="The Rusty Anchor", content="A seedy tavern on the docks", ) knowledge_dir = ( tmp_path / "players" / ctx.player_id / "worlds" / ctx.world_id / "lore" ) assert any(knowledge_dir.iterdir()) def test_with_content_type(self, ctx: ToolContext, tmp_path: Path): call_tool( _note_discovery, entity="Vera", content="Tavern owner", content_type="npcs" ) knowledge_dir = ( tmp_path / "players" / ctx.player_id / "worlds" / ctx.world_id / "npcs" ) assert any(knowledge_dir.iterdir()) def test_with_tags(self, ctx: ToolContext, tmp_path: Path): call_tool( _note_discovery, entity="Old Map", content="Shows a hidden passage", tags=["quest"], ) knowledge_dir = ( tmp_path / "players" / ctx.player_id / "worlds" / ctx.world_id / "lore" ) content = next(knowledge_dir.iterdir()).read_text() assert "quest" in content class TestEndSessionDirect: def test_returns_session_ended(self, ctx: ToolContext): result = call_tool(_end_session, situation="In the tavern") assert result == "SESSION_ENDED" def test_with_threads(self, ctx: ToolContext): result = call_tool( _end_session, situation="In the tavern", threads=["Find the merchant", "Investigate the warehouse"], ) assert result == "SESSION_ENDED" # --- Recall with indexed content -------------------------------------------- @pytest.fixture def three_indexed_npcs(ctx: ToolContext) -> list[str]: """Index 3 dock-loitering NPCs and return their names.""" names = [f"NPC {i}" for i in range(3)] for i, name in enumerate(names): ctx.vector_index.upsert( f"world:npcs/npc{i}.md:0", f"NPC number {i} who hangs around the docks", { "source": "world", "content_type": "npcs", "path": f"/fake/npc{i}.md", "title": name, }, ) return names class TestRecall: def test_recall_finds_indexed_entity( self, ctx: ToolContext, tmp_path: Path, mcp_call: McpCall ): entity_dir = tmp_path / "worlds" / ctx.world_id / "npcs" entity_dir.mkdir(parents=True, exist_ok=True) entity_file = entity_dir / "Vera Blackwater.md" entity_file.write_text("# Vera Blackwater\n\nTavern owner.\n") ctx.vector_index.upsert( "world:npcs/Vera Blackwater.md:0", "Vera Blackwater. Tavern owner, former smuggler.", { "source": "world", "content_type": "npcs", "path": str(entity_file), "title": "Vera Blackwater", }, ) result = mcp_call("recall", {"query": "Vera Blackwater"}) assert "Vera Blackwater" in result def test_recall_rules_scope(self, ctx: ToolContext, mcp_call: McpCall): result = mcp_call("recall", {"query": "fireball", "scope": "rules"}) assert "Nothing found" in result def test_recall_world_scope(self, ctx: ToolContext, mcp_call: McpCall): result = mcp_call("recall", {"query": "something", "scope": "world"}) assert "Nothing found" in result @pytest.mark.usefixtures("three_indexed_npcs") def test_recall_multiple_hits(self, ctx: ToolContext, mcp_call: McpCall): result = mcp_call("recall", {"query": "docks NPC"}) assert "Found" in result or "Nothing found" in result # --- Combat path: damage routes through the initiative tracker -------------- class TestDamageHealCombat: def test_damage_combatant_in_initiative( self, ctx: ToolContext, mcp_call_in_combat: McpCallInCombat ): result = mcp_call_in_combat( "damage", {"target": "Goblin", "amount": 3}, [Combatant(name="Goblin", initiative=10, hp=7, hp_max=7, ac=15)], ) assert "3" in result assert ctx.initiative._find("Goblin").hp == 4 def test_damage_syncs_player_hp( self, ctx: ToolContext, tmp_path: Path, mcp_call: McpCall, mcp_call_in_combat: McpCallInCombat, ): mcp_call( "create_character", { "name": "Kira", "race": "Human", "char_class": "Fighter", "level": 1, "abilities": { "strength": 16, "dexterity": 12, "constitution": 14, "intelligence": 10, "wisdom": 13, "charisma": 8, }, "hp_max": 25, "ac": 16, }, ) result = mcp_call_in_combat( "damage", {"target": "Kira", "amount": 7}, [ Combatant( name="Kira", initiative=18, hp=25, hp_max=25, ac=16, is_player=True ) ], ) assert "synced" in result char = load_character(ctx.player_id) assert char["state"]["hp"]["current"] == 18, ( "_sync_player_hp must write to state.hp.current with the new schema" ) def test_heal_syncs_player_hp( self, ctx: ToolContext, tmp_path: Path, mcp_call: McpCall, mcp_call_in_combat: McpCallInCombat, ): mcp_call( "create_character", { "name": "Kira", "race": "Human", "char_class": "Fighter", "level": 1, "abilities": { "strength": 16, "dexterity": 12, "constitution": 14, "intelligence": 10, "wisdom": 13, "charisma": 8, }, "hp_max": 25, "ac": 16, }, ) mcp_call("update_character", {"updates": {"state.hp.current": 20}}) result = mcp_call_in_combat( "heal", {"target": "Kira", "amount": 3}, [ Combatant( name="Kira", initiative=18, hp=20, hp_max=25, ac=16, is_player=True ) ], ) assert "synced" in result char = load_character(ctx.player_id) assert char["state"]["hp"]["current"] == 23, ( "_sync_player_hp must write to state.hp.current with the new schema" ) def test_damage_no_sync_for_non_player( self, ctx: ToolContext, mcp_call_in_combat: McpCallInCombat ): result = mcp_call_in_combat( "damage", {"target": "Goblin", "amount": 3}, [Combatant(name="Goblin", initiative=10, hp=7, hp_max=7, ac=15)], ) assert "synced" not in result def test_unknown_target_returns_error(self, ctx: ToolContext, mcp_call: McpCall): result = mcp_call("damage", {"target": "Nobody", "amount": 5}) assert "No such target" in result # --- Combat tool wrappers (exercised via the in-memory client) -------------- class TestCombatTools: """The combat FastMCP wrappers route through the active InitiativeTracker. These tests start initiative on the process-global ctx via enter_initiative, flip combat tools visible, then drive each wrapper via the in-memory client so the wrapper bodies (not just the underlying tracker) get exercised. """ def test_enter_initiative_starts_combat( self, ctx: ToolContext, mcp_call_in_combat: McpCallInCombat ): result = mcp_call_in_combat( "next_turn", {}, [ Combatant( name="Kira", initiative=18, hp=25, hp_max=25, ac=16, is_player=True ), Combatant(name="Goblin", initiative=10, hp=7, hp_max=7, ac=15), ], ) assert ctx.initiative.active # next_turn from Kira should advance to Goblin assert "Goblin" in result def test_enter_initiative_via_client( self, ctx: ToolContext, mcp_call_in_combat: McpCallInCombat ): """Drive enter_initiative through the in-memory client so the parsing + tracker.begin + flip path runs end-to-end.""" result = mcp_call_in_combat( "enter_initiative", { "combatants": [ { "name": "Kira", "initiative": 18, "hp": 25, "hp_max": 25, "ac": 16, "is_player": True, }, { "name": "Goblin", "initiative": 10, "hp": 7, "hp_max": 7, "ac": 15, }, ], }, ) assert "Initiative started" in result or "Round 1" in result assert ctx.initiative.active def test_enter_initiative_when_already_active_errors( self, ctx: ToolContext, mcp_call_in_combat: McpCallInCombat ): # Start combat manually so the wrapper hits the early-return guard result = mcp_call_in_combat( "enter_initiative", { "combatants": [ { "name": "X", "initiative": 1, "hp": 1, "hp_max": 1, "ac": 10, } ], }, [Combatant(name="Goblin", initiative=10, hp=7, hp_max=7, ac=15)], ) assert "already active" in result.lower() def test_add_combatant_inserts_into_initiative( self, ctx: ToolContext, mcp_call_in_combat: McpCallInCombat ): result = mcp_call_in_combat( "add_combatant", {"name": "Reinforcement", "initiative": 12, "hp": 5, "hp_max": 5, "ac": 14}, [Combatant(name="Goblin", initiative=10, hp=7, hp_max=7, ac=15)], ) assert "Reinforcement" in result assert ctx.initiative._find("Reinforcement") is not None def test_remove_combatant_removes_from_initiative( self, ctx: ToolContext, mcp_call_in_combat: McpCallInCombat ): result = mcp_call_in_combat( "remove_combatant", {"name": "Goblin"}, [ Combatant(name="Kira", initiative=18, hp=25, hp_max=25, ac=16), Combatant(name="Goblin", initiative=10, hp=7, hp_max=7, ac=15), ], ) assert "removed" in result.lower() assert ctx.initiative._find("Goblin") is None def test_condition_add(self, ctx: ToolContext, mcp_call_in_combat: McpCallInCombat): result = mcp_call_in_combat( "condition", {"target": "Goblin", "condition": "Poisoned"}, [Combatant(name="Goblin", initiative=10, hp=7, hp_max=7, ac=15)], ) assert "Poisoned" in result # The combatant should now have the condition tracked goblin = ctx.initiative._find("Goblin") assert any(c.name == "Poisoned" for c in goblin.conditions) def test_condition_remove( self, ctx: ToolContext, mcp_call_in_combat: McpCallInCombat ): ctx.initiative.begin( [Combatant(name="Goblin", initiative=10, hp=7, hp_max=7, ac=15)] ) ctx.initiative.add_condition(target="Goblin", condition="Stunned") result = mcp_call_in_combat( "condition", { "target": "Goblin", "condition": "Stunned", "action": "remove", }, ) assert "removed" in result.lower() or "Stunned" in result goblin = ctx.initiative._find("Goblin") assert not any(c.name == "Stunned" for c in goblin.conditions) def test_end_initiative_clears_combat( self, ctx: ToolContext, mcp_call_in_combat: McpCallInCombat ): result = mcp_call_in_combat( "end_initiative", {}, [Combatant(name="Goblin", initiative=10, hp=7, hp_max=7, ac=15)], ) assert "ended" in result.lower() assert not ctx.initiative.active # --- Character wrapper bodies ---------------------------------------------- @pytest.fixture def kira(ctx: ToolContext, mcp_call: McpCall) -> ToolContext: """A minimum-viable character used by the wrapper-coverage tests.""" mcp_call( "create_character", { "name": "Kira", "race": "Human", "char_class": "Fighter", "level": 1, "abilities": { "strength": 16, "dexterity": 12, "constitution": 14, "intelligence": 10, "wisdom": 13, "charisma": 8, }, "hp_max": 12, "ac": 16, }, ) return ctx class TestCharacterToolWrappers: """Each character.py @mcp.tool wrapper is a one-liner that delegates to a char_* function. These tests exercise the wrappers through the in-memory client so the wrapper bodies (and their Dependency-resolved arguments) are actually executed.""" def test_damage_player_by_name(self, kira: ToolContext, mcp_call: McpCall): result = mcp_call("damage", {"target": "Kira", "amount": 3}) assert "3" in result char = load_character("default") assert char["state"]["hp"]["current"] == 9 def test_damage_with_type(self, kira: ToolContext, mcp_call: McpCall): result = mcp_call("damage", {"target": "Kira", "amount": 2, "type": "fire"}) assert "2" in result def test_heal_player_by_name(self, kira: ToolContext, mcp_call: McpCall): mcp_call("damage", {"target": "Kira", "amount": 5}) result = mcp_call("heal", {"target": "Kira", "amount": 3}) assert result char = load_character("default") assert char["state"]["hp"]["current"] == 10 def test_adjust_coins(self, kira: ToolContext, mcp_call: McpCall): result = mcp_call("adjust_coins", {"deltas": {"gp": 10, "sp": 5}}) assert "10" in result char = load_character("default") assert char["state"]["purse"]["gp"] == 10 assert char["state"]["purse"]["sp"] == 5 def test_adjust_coins_drops_zero_deltas(self, kira: ToolContext, mcp_call: McpCall): # Spending only gp; the zero-delta filter exercises the comprehension branch mcp_call("adjust_coins", {"deltas": {"gp": 5}}) result = mcp_call("adjust_coins", {"deltas": {"gp": -3, "sp": 0}}) char = load_character("default") assert char["state"]["purse"]["gp"] == 2 assert "silver" not in result.lower() def test_add_effect(self, kira: ToolContext, mcp_call: McpCall): result = mcp_call( "add_effect", { "source": "Bless", "description": "+1d4 attacks/saves", }, ) assert result def test_add_effect_with_expires(self, kira: ToolContext, mcp_call: McpCall): result = mcp_call( "add_effect", { "source": "Heroism", "description": "+10 temp HP", "expires": "d1-1430", }, ) assert result def test_remove_effect(self, kira: ToolContext, mcp_call: McpCall): mcp_call("add_effect", {"source": "Bless", "description": "+1d4"}) result = mcp_call("remove_effect", {"source": "Bless"}) assert result def test_add_and_remove_condition(self, kira: ToolContext, mcp_call: McpCall): result = mcp_call("add_condition", {"name": "Poisoned"}) assert result result = mcp_call("remove_condition", {"name": "Poisoned"}) assert result def test_add_item_default_location(self, kira: ToolContext, mcp_call: McpCall): result = mcp_call("add_item", {"item": "Lockpicks"}) assert result def test_add_item_with_location(self, kira: ToolContext, mcp_call: McpCall): result = mcp_call( "add_item", { "item": "Spare cloak", "location": "stashed_at_inn", }, ) assert result def test_remove_item(self, kira: ToolContext, mcp_call: McpCall): mcp_call("add_item", {"item": "Boot knife"}) result = mcp_call("remove_item", {"item": "Boot knife"}) assert result def test_set_item_status(self, kira: ToolContext, mcp_call: McpCall): # The item must already be a known magic item entity for status tracking mcp_call( "establish", { "entity_type": "items", "name": "Bracer of Defense", "description": "A leather bracer with a faint silver sheen.", }, ) result = mcp_call( "set_item_status", { "item": "Bracer of Defense", "status": "attuned", }, ) assert result def test_adjust_resource(self, kira: ToolContext, mcp_call: McpCall): # Add a resource via update_character first mcp_call( "update_character", { "updates": { "resources.hit_dice_d10": { "current": 1, "max": 1, "refresh": "long_rest", "notes": "Hit Dice (d10)", }, }, }, ) result = mcp_call("adjust_resource", {"name": "hit_dice", "delta": -1}) assert "Used" in result result = mcp_call("adjust_resource", {"name": "hit_dice", "delta": 1}) assert "Restored" in result def test_rest_short(self, kira: ToolContext, mcp_call: McpCall): result = mcp_call("rest", {"type": "short"}) assert result def test_rest_long(self, kira: ToolContext, mcp_call: McpCall): result = mcp_call("rest", {"type": "long"}) assert result def test_add_note(self, kira: ToolContext, mcp_call: McpCall): result = mcp_call("add_note", {"text": "Found a hidden passage"}) assert result # --- Scene/world wrappers -------------------------------------------------- class TestSceneToolWrappers: """Cover the scene.py wrapper bodies — set_scene's optional-field branches, tune's file write, end_session's threads branch, notify_dm's append.""" def test_set_scene_event_only(self, ctx: ToolContext, mcp_call: McpCall): result = mcp_call( "set_scene", { "event": "Walked into the tavern", "duration": "5 min", }, ) assert "Logged" in result def test_set_scene_with_situation_and_location( self, ctx: ToolContext, mcp_call: McpCall ): result = mcp_call( "set_scene", { "event": "Arrived at the inn", "duration": "1 hour", "situation": "Resting by the fire", "location": "The Rusty Anchor", }, ) assert "Logged" in result assert "updated" in result.lower() def test_set_scene_with_present_auto_marks( self, ctx: ToolContext, mcp_call: McpCall ): # Establish an entity first so auto-mark has something to find mcp_call( "establish", { "entity_type": "npcs", "name": "Vera", "description": "Tavern owner.", }, ) result = mcp_call( "set_scene", { "event": "Spoke with Vera", "duration": "10 min", "present": ["[[Vera]]"], }, ) assert "Auto-marked: Vera" in result def test_auto_mark_cooldown_suppresses_near_repeats( self, ctx: ToolContext, tmp_path: Path, mcp_call: McpCall, ): """A second set_scene with the same present entity within the cooldown window should NOT append another Was entry.""" mcp_call( "establish", { "entity_type": "npcs", "name": "Margit", "description": "Chandler.", }, ) mcp_call( "set_scene", { "event": "Met Mira at the candle shop", "duration": "10 min", "present": ["[[Margit]]"], }, ) result2 = mcp_call( "set_scene", { "event": "Walked together to dinner", "duration": "10 min", "present": ["[[Margit]]"], }, ) assert "Auto-marked" not in result2 content = ( tmp_path / "worlds" / ctx.world_id / "npcs" / "Margit.md" ).read_text() assert content.count("Met Mira at the candle shop") == 1 assert "Walked together to dinner" not in content def test_auto_mark_cooldown_clears_after_window( self, ctx: ToolContext, mcp_call: McpCall ): """After more than cooldown minutes have passed in-game, the next auto-mark for the same entity fires again.""" mcp_call( "establish", { "entity_type": "npcs", "name": "Aldric", "description": "Bookseller.", }, ) mcp_call( "set_scene", { "event": "Briefed Aldric on the investigation", "duration": "30 min", # advances the clock past the cooldown "present": ["[[Aldric]]"], }, ) mcp_call( "set_scene", { "event": "Walked away to get lunch", "duration": "30 min", }, ) result3 = mcp_call( "set_scene", { "event": "Returned and shared a new lead with Aldric", "duration": "20 min", "present": ["[[Aldric]]"], }, ) assert "Auto-marked: Aldric" in result3 def test_auto_mark_skips_operational_events( self, ctx: ToolContext, tmp_path: Path, mcp_call: McpCall ): """Session-lifecycle events (Session resumed, etc) must never land in an entity's history.""" mcp_call( "establish", { "entity_type": "npcs", "name": "Dortha", "description": "Tanner.", }, ) result = mcp_call( "set_scene", { "event": "Session resumed. Mira at Dortha's shop.", "duration": "0 min", "present": ["[[Dortha]]"], }, ) assert "Auto-marked" not in result content = ( tmp_path / "worlds" / ctx.world_id / "npcs" / "Dortha.md" ).read_text() assert "Session resumed" not in content def test_set_scene_with_threads(self, ctx: ToolContext, mcp_call: McpCall): result = mcp_call( "set_scene", { "event": "Got a lead", "duration": "5 min", "threads": ["Find the missing merchant"], }, ) assert result def test_set_scene_no_args_returns_no_updates( self, ctx: ToolContext, mcp_call: McpCall ): # Both event and duration omitted, no other fields → "No updates" result = mcp_call("set_scene", {}) assert result == "No updates" def test_tune_writes_style_file( self, ctx: ToolContext, tmp_path: Path, mcp_call: McpCall ): result = mcp_call("tune", {"tuning": "Lean into intrigue and slow pacing."}) assert "updated" in result.lower() style_path = tmp_path / "worlds" / ctx.world_id / "style.md" assert style_path.exists() assert "intrigue" in style_path.read_text() def test_end_session_no_threads(self, ctx: ToolContext, mcp_call: McpCall): result = mcp_call("end_session", {"situation": "In the tavern"}) assert result == "SESSION_ENDED" def test_end_session_with_threads(self, ctx: ToolContext, mcp_call: McpCall): result = mcp_call( "end_session", { "situation": "In the tavern", "threads": ["Investigate the warehouse", "Find the merchant"], }, ) assert result == "SESSION_ENDED" def test_notify_dm_appends_to_queue( self, ctx: ToolContext, tmp_path: Path, mcp_call: McpCall ): result = mcp_call("notify_dm", {"message": "Background world has shifted"}) assert "queued" in result.lower() path = tmp_path / "worlds" / ctx.world_id / "dm_notifications.md" assert path.exists() assert "Background world has shifted" in path.read_text() # --- run_code wrapper ------------------------------------------------------- class TestRunCodeWrapper: def test_run_code_simple(self, ctx: ToolContext, mcp_call: McpCall): result = mcp_call( "run_code", { "description": "Two plus two", "code": "2 + 2", }, ) assert "4" in result def test_run_code_with_print(self, ctx: ToolContext, mcp_call: McpCall): result = mcp_call( "run_code", { "description": "Print test", "code": 'print("hello sandbox")', }, ) assert "hello sandbox" in result