Something went wrong. Try again.
A 5e storytelling engine with an LLM DM
Something went wrong. Try again.
Python
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632"""Tests for the FastMCP server composition.
The orchestrator in storied.mcp_server builds a per-role top-level FastMCPserver by mounting the tools/*.py module-level FastMCP instances andapplying tag-based visibility filters. These tests verify the per-roletool visibility plus the dynamic combat-tag flip when initiative startsand ends."""
import asyncio
import pytest
from storied.mcp_server import _compose_serverfrom storied.tools import ToolContextfrom storied.tools.character import refresh_advancement_visibilityfrom storied.tools.combat import _flip_into_combat, _flip_out_of_combat
def _names(role: str) -> set[str]: async def _gather() -> set[str]: server = await _compose_server(role) return {t.name for t in await server.list_tools()}
return asyncio.run(_gather())
class TestPerRoleComposition: """Each role sees only the tools tagged for it."""
def test_dm_includes_core_narrative_tools(self): names = _names("dm") assert "set_scene" in names assert "establish" in names assert "mark" in names assert "end_session" in names assert "recall" in names assert "roll" in names assert "run_code" in names
def test_dm_includes_character_tools(self): names = _names("dm") for tool_name in ( "damage", "heal", "adjust_coins", "add_effect", "remove_effect", "add_condition", "remove_condition", "add_item", "remove_item", "set_item_status", "adjust_resource", "rest", "add_note", "update_character", "create_character", ): assert tool_name in names, f"missing {tool_name}"
def test_dm_does_not_include_removed_tools(self): """break_concentration, use_resource, and restore_resource were folded into other tools — ensure they don't resurface.""" names = _names("dm") assert "break_concentration" not in names assert "use_resource" not in names assert "restore_resource" not in names
def test_dm_initial_excludes_combat_tools(self): """In DM mode, combat tools are hidden until enter_initiative runs.""" names = _names("dm") assert "next_turn" not in names assert "add_combatant" not in names assert "remove_combatant" not in names assert "condition" not in names
def test_dm_initial_keeps_combat_control(self): """enter_initiative and end_initiative stay visible so combat can begin.""" names = _names("dm") assert "enter_initiative" in names assert "end_initiative" in names
def test_planner_only_has_its_tools(self): assert _names("planner") == { "establish", "mark", "amend_mark", "notify_dm", "recall", "forge_culture", "generate_names", }
def test_seeder_only_has_its_tools(self): assert _names("seeder") == { "establish", "set_scene", "forge_culture", "generate_names", }
def test_advancement_only_has_its_tools(self): assert _names("advancement") == {"notify_dm", "recall", "update_character"}
def test_arc_architect_only_has_its_tools(self): # The arc architect's whole job is critique-and-commit. Nothing # else. World-building tools must not leak into this role. assert _names("arc_architect") == {"commit_arc", "recall"}
def test_arc_architect_in_all_roles(self): from storied.mcp_server import ALL_ROLES
assert "arc_architect" in ALL_ROLES
class TestToolSchemas: """Verify tool input schemas expose nested field shapes to the LLM.
These tests guard against regression to bare `dict` / `list` parameter types, which leave the LLM with no guidance about which keys are required. """
def _schema(self, tool_name: str) -> dict: async def _gather() -> dict: server = await _compose_server("dm") for t in await server.list_tools(): if t.name == tool_name: return t.parameters raise AssertionError(f"tool {tool_name!r} not found")
return asyncio.run(_gather())
def test_enter_initiative_documents_combatant_shape(self): schema = self._schema("enter_initiative") item_schema = schema["properties"]["combatants"]["items"] required = set(item_schema["required"]) assert {"name", "initiative", "hp", "hp_max", "ac"} <= required, ( f"enter_initiative must require all combatant fields, got {required}" ) # is_player is optional but documented assert "is_player" in item_schema["properties"]
def test_create_character_documents_ability_keys(self): schema = self._schema("create_character") ability_props = schema["properties"]["abilities"]["properties"] for ability in ( "strength", "dexterity", "constitution", "intelligence", "wisdom", "charisma", ): assert ability in ability_props, ( f"create_character must document the {ability} ability score" )
def test_adjust_coins_documents_denominations(self): schema = self._schema("adjust_coins") delta_props = schema["properties"]["deltas"]["properties"] for denom in ("cp", "sp", "ep", "gp", "pp"): assert denom in delta_props, ( f"adjust_coins must document the {denom} denomination" )
def test_create_character_purse_documents_denominations(self): schema = self._schema("create_character") purse = schema["properties"]["purse"] # Purse is wrapped in anyOf for the | None purse_props = next( opt for opt in purse["anyOf"] if opt.get("type") == "object" )["properties"] for denom in ("cp", "sp", "ep", "gp", "pp"): assert denom in purse_props
@pytest.mark.parametrize( ("tool_name", "param", "expected_values"), [ ("rest", "type", {"short", "long"}), ("set_item_status", "status", {"attuned", "equipped", "carried"}), ("recall", "scope", {"rules", "world", "all"}), ( "establish", "entity_type", { "npcs", "locations", "items", "factions", "threads", "lore", "maps", "cultures", }, ), ( "mark", "entity_type", { "npcs", "locations", "items", "factions", "threads", "maps", "cultures", }, ), ( "note_discovery", "content_type", {"npcs", "locations", "factions", "lore", "cultures"}, ), ], ) def test_enum_parameters_expose_valid_values( self, tool_name: str, param: str, expected_values: set[str], ): """Each conceptually-enum parameter must surface as a JSON Schema enum, not a free-form string. Guards against regression to bare `str`.""" schema = self._schema(tool_name) prop = schema["properties"][param] # Default values wrap the enum in anyOf for `Type | None`; unwrap if needed enum_values = prop.get("enum") if enum_values is None and "anyOf" in prop: for opt in prop["anyOf"]: if "enum" in opt: enum_values = opt["enum"] break assert enum_values is not None, ( f"{tool_name}.{param} should expose an enum, got {prop}" ) assert set(enum_values) == expected_values
def test_combat_condition_enum_parameters(self): """The combat `condition` tool is hidden in the default DM compose (combat-only), so check it directly on the combat module.""" from storied.tools.combat import mcp as combat_mcp
async def _gather() -> dict: for t in await combat_mcp.list_tools(): if t.name == "condition": return t.parameters raise AssertionError("condition tool not found")
params = asyncio.run(_gather()) assert set(params["properties"]["action"]["enum"]) == {"add", "remove"} assert set(params["properties"]["ends_on"]["enum"]) == {"start", "end"}
class TestCombatTagFlip: """Entering and ending initiative toggles combat-tag visibility on the composed top-level server."""
def test_flip_into_combat_shows_combat_tools(self, ctx: ToolContext): async def _run() -> set[str]: server = await _compose_server("dm") _flip_into_combat() return {t.name for t in await server.list_tools()}
names = asyncio.run(_run()) assert "next_turn" in names assert "add_combatant" in names assert "condition" in names # Cleanup _flip_out_of_combat()
def test_flip_out_of_combat_hides_combat_tools(self, ctx: ToolContext): async def _run() -> set[str]: server = await _compose_server("dm") _flip_into_combat() _flip_out_of_combat() return {t.name for t in await server.list_tools()}
names = asyncio.run(_run()) assert "next_turn" not in names assert "add_combatant" not in names
def test_combat_control_stays_visible_through_cycle(self, ctx: ToolContext): """enter_initiative / end_initiative are tagged combat_control and must stay visible whether initiative is active or not."""
async def _gather_combat_control() -> tuple[set[str], set[str], set[str]]: server = await _compose_server("dm") initial = {t.name for t in await server.list_tools()} _flip_into_combat() during = {t.name for t in await server.list_tools()} _flip_out_of_combat() after = {t.name for t in await server.list_tools()} return initial, during, after
initial, during, after = asyncio.run(_gather_combat_control()) for state in (initial, during, after): assert "enter_initiative" in state assert "end_initiative" in state
class TestAdvancementVisibility: """level_up is hidden until the character has advancement_ready set."""
def test_level_up_hidden_at_compose_time(self, ctx: ToolContext): """Fresh compose should not expose level_up — nothing has granted it yet."""
async def _gather() -> set[str]: server = await _compose_server("dm") return {t.name for t in await server.list_tools()}
names = asyncio.run(_gather()) assert "level_up" not in names
def test_level_up_revealed_when_advancement_ready(self, ctx: ToolContext): async def _run() -> set[str]: server = await _compose_server("dm") refresh_advancement_visibility({"advancement_ready": 4}) return {t.name for t in await server.list_tools()}
names = asyncio.run(_run()) assert "level_up" in names refresh_advancement_visibility(None) # cleanup
def test_level_up_hidden_again_when_flag_cleared(self, ctx: ToolContext): async def _run() -> set[str]: server = await _compose_server("dm") refresh_advancement_visibility({"advancement_ready": 4}) refresh_advancement_visibility({"advancement_ready": None}) return {t.name for t in await server.list_tools()}
names = asyncio.run(_run()) assert "level_up" not in names
def test_level_up_not_in_planner_compose(self, ctx: ToolContext): """Only the DM role cares about advancement visibility. Other roles don't have level_up at all, so the flip is a no-op for them."""
async def _run() -> set[str]: server = await _compose_server("planner") return {t.name for t in await server.list_tools()}
names = asyncio.run(_run()) assert "level_up" not in names
def test_refresh_with_none_character_is_safe(self, ctx: ToolContext): """A character sheet that doesn't exist yet (pre-creation) should not crash the visibility flip."""
async def _run() -> None: await _compose_server("dm") refresh_advancement_visibility(None)
asyncio.run(_run())
class TestPopulateIndex: """Cover the SRD-seeding helper without launching a real server.
Tests pass an explicit ``srd_root`` pointed at a tmp path so the real package rules directory isn't touched. """
def test_no_srd_no_world_dir(self, tmp_path): from unittest.mock import MagicMock
from storied.mcp_server import _populate_index
vi = MagicMock() vi.has_source.return_value = False _populate_index( tmp_path / "worlds" / "missing", vi, srd_root=tmp_path / "srd-missing", ) # No SRD seed, no SRD sections, no world dir → nothing should be called vi.reseed.assert_not_called() vi.reindex_directory.assert_not_called()
def test_world_dir_only(self, tmp_path): from unittest.mock import MagicMock
from storied.mcp_server import _populate_index
world_dir = tmp_path / "worlds" / "test" world_dir.mkdir(parents=True) vi = MagicMock() vi.has_source.return_value = False _populate_index( world_dir, vi, srd_root=tmp_path / "srd-missing", ) vi.reindex_directory.assert_called_once_with( world_dir, source="world", skip_subdirs=frozenset({"transcripts"}), )
def test_srd_sections_dir(self, tmp_path): from unittest.mock import MagicMock
from storied.mcp_server import _populate_index
srd_root = tmp_path / "srd-5.2.1" srd_dir = srd_root / "sections" srd_dir.mkdir(parents=True) world_dir = tmp_path / "worlds" / "test" vi = MagicMock() vi.has_source.return_value = False _populate_index(world_dir, vi, srd_root=srd_root) # SRD sections present → reindex SRD; no user layer, no world dir assert vi.reindex_directory.call_count == 1 vi.reindex_directory.assert_called_with(srd_dir, source="srd")
def test_user_layer_indexed(self, tmp_path): """When the user homebrew directory exists, _populate_index reindexes it with source='user' after the shipped SRD.""" from unittest.mock import MagicMock, call
from storied.mcp_server import _populate_index
# The autouse fixture sets _user_rules_home = tmp_path / "rules". user_dir = tmp_path / "rules" user_dir.mkdir(parents=True) (user_dir / "monsters").mkdir() (user_dir / "monsters" / "homebrew.md").write_text("# Homebrew")
world_dir = tmp_path / "worlds" / "test" world_dir.mkdir(parents=True)
vi = MagicMock() vi.has_source.return_value = False _populate_index( world_dir, vi, srd_root=tmp_path / "srd-missing", ) # Should have indexed user and world, in that order assert vi.reindex_directory.call_args_list == [ call(user_dir, source="user"), call( world_dir, source="world", skip_subdirs=frozenset({"transcripts"}), ), ]
def test_populate_index_is_idempotent(self, tmp_path): """Once the SRD is seeded, subsequent calls must not reseed (which would wipe world/transcript rows by file-copying the SRD db over the live one).""" from unittest.mock import MagicMock
from storied.mcp_server import _populate_index
srd_root = tmp_path / "srd-5.2.1" srd_root.mkdir(parents=True) srd_seed = srd_root / "search.db" srd_seed.write_bytes(b"sqlite stub") world_dir = tmp_path / "worlds" / "test" world_dir.mkdir(parents=True)
vi = MagicMock() vi.has_source.return_value = True # SRD already seeded _populate_index(world_dir, vi, srd_root=srd_root) vi.reseed.assert_not_called()
def test_flip_helpers_no_op_when_root_unset(self): """The combat-tag flip helpers must not crash when no top-level server has been registered (e.g. when the combat module is imported in isolation by a test before any compose_server call).""" import storied.tools.combat as combat_mod
# Snapshot and clear _root for the duration of the test saved_root = combat_mod._root saved_keys = combat_mod._combat_keys_to_hide combat_mod._root = None combat_mod._combat_keys_to_hide = set() try: combat_mod._flip_into_combat() # should silently no-op combat_mod._flip_out_of_combat() finally: combat_mod._root = saved_root combat_mod._combat_keys_to_hide = saved_keys
def test_srd_seed_db_takes_priority(self, tmp_path): from unittest.mock import MagicMock
from storied.mcp_server import _populate_index
srd_root = tmp_path / "srd-5.2.1" srd_root.mkdir(parents=True) srd_seed = srd_root / "search.db" srd_seed.write_bytes(b"sqlite stub") # Also create the sections dir to verify the seed wins (srd_root / "sections").mkdir() world_dir = tmp_path / "worlds" / "test" vi = MagicMock() vi.has_source.return_value = False _populate_index(world_dir, vi, srd_root=srd_root) vi.reseed.assert_called_once_with(srd_seed) # When the seed exists, we don't also reindex SRD sections vi.reindex_directory.assert_not_called()
class TestRulesLookupRace: """Regression test for the transcript-upsert-vs-first-search race.
Before the fix, the lazy ``on_empty`` seeding would miss its window whenever the DM's first turn didn't call ``recall``: the engine would upsert the transcript at turn end, making the db non-empty, and the next search would skip seeding because ``count > 0``. Rules lookups would then return nothing. The fix eagerly populates in ``start_server``, and ``_populate_index`` is idempotent via ``has_source("srd")``. """
def test_srd_stays_available_after_transcript_upsert_race( self, tmp_path, monkeypatch, ): from storied import paths from storied.mcp_server import _populate_index from storied.search import VectorIndex
# Minimal shipped SRD seed the populate helper can copy from. srd_root = tmp_path / "shipped" / "srd-5.2.1" srd_root.mkdir(parents=True) seed_db = srd_root / "search.db" seed_index = VectorIndex(seed_db) seed_index.upsert( "srd:character-origins.md:0", "# Character Origins\n\n**Half-Elf** gets +2 Charisma.", { "source": "srd", "content_type": "rules", "path": str(srd_root / "sections" / "character-origins.md"), "title": "Character Origins", }, ) seed_index.close()
monkeypatch.setattr( paths, "shipped_rules_path", lambda: tmp_path / "shipped", )
world_dir = paths.world_path("default") world_dir.mkdir(parents=True, exist_ok=True) db_path = world_dir / "search.db"
# Eager populate the way start_server now does. vi = VectorIndex(db_path) _populate_index(world_dir, vi) assert vi.has_source("srd")
# Turn 1 ends without any recall — engine upserts the transcript. (world_dir / "transcripts").mkdir(exist_ok=True) day_path = world_dir / "transcripts" / "day+001.md" day_path.write_text("### Day 1\n\nHey, welcome.") vi.upsert( "transcript:transcripts/day+001.md:0", day_path.read_text(), { "source": "transcript", "content_type": "transcripts", "path": str(day_path), "title": "Day 1", "game_day": 1, }, )
# Turn 2: the DM finally calls recall. SRD must still be there. hits = vi.search( "half-elf charisma", limit=3, source_filter=["srd", "user", "world"], ) assert len(hits) >= 1 assert any(h.source == "srd" for h in hits)
def test_populate_is_idempotent_on_repeated_start_server( self, tmp_path, monkeypatch, ): """A second start_server (onboarding → play handoff) must not wipe the world/transcript rows by re-copying the SRD seed.""" from storied import paths from storied.mcp_server import _populate_index from storied.search import VectorIndex
srd_root = tmp_path / "shipped" / "srd-5.2.1" srd_root.mkdir(parents=True) seed_db = srd_root / "search.db" seed_index = VectorIndex(seed_db) seed_index.upsert( "srd:x.md:0", "# X", {"source": "srd", "path": "x.md"}, ) seed_index.close()
monkeypatch.setattr( paths, "shipped_rules_path", lambda: tmp_path / "shipped", )
world_dir = paths.world_path("default") world_dir.mkdir(parents=True, exist_ok=True) db_path = world_dir / "search.db"
vi = VectorIndex(db_path) _populate_index(world_dir, vi)
# Upsert a transcript row to represent prior session state. vi.upsert( "transcript:transcripts/day+001.md:0", "turn content", {"source": "transcript", "path": "transcripts/day+001.md"}, ) assert vi.has_source("transcript")
# Second populate — must not wipe transcripts via reseed. _populate_index(world_dir, vi) assert vi.has_source("srd") assert vi.has_source("transcript")