diff --git a/.loq_cache b/.loq_cache --- a/.loq_cache +++ b/.loq_cache @@ -1,1 +1,1 @@ -{"version":1,"config_hash":4557771575092473650,"entries":{"src/storied/initiative.py":{"mtime_secs":1774751731,"mtime_nanos":975235964,"lines":524}}} \ No newline at end of file +{"version":1,"config_hash":4557771575092473650,"entries":{"src/storied/tools/_context.py":{"mtime_secs":1775582352,"mtime_nanos":694759924,"lines":82},"src/storied/display.py":{"mtime_secs":1774782927,"mtime_nanos":98901860,"lines":355},"src/storied/dice.py":{"mtime_secs":1767047766,"mtime_nanos":398050421,"lines":180},"src/storied/initiative.py":{"mtime_secs":1774916278,"mtime_nanos":380939471,"lines":574},"src/storied/advancement.py":{"mtime_secs":1775582572,"mtime_nanos":681291189,"lines":199},"src/storied/srd/clean.py":{"mtime_secs":1766869132,"mtime_nanos":684969729,"lines":101},"src/storied/sandbox.py":{"mtime_secs":1774959788,"mtime_nanos":500556437,"lines":147},"src/storied/content.py":{"mtime_secs":1774709737,"mtime_nanos":677509525,"lines":115},"src/storied/cli.py":{"mtime_secs":1775582667,"mtime_nanos":84277922,"lines":1090},"src/storied/tools.py":{"mtime_secs":1775581781,"mtime_nanos":567165396,"lines":1203},"src/storied/__init__.py":{"mtime_secs":1766844771,"mtime_nanos":901498410,"lines":3},"src/storied/tools/mechanics.py":{"mtime_secs":1775582379,"mtime_nanos":322091028,"lines":143},"src/storied/notifications.py":{"mtime_secs":1775581593,"mtime_nanos":188431521,"lines":49},"src/storied/planner.py":{"mtime_secs":1775582662,"mtime_nanos":80226590,"lines":519},"src/storied/tools/character.py":{"mtime_secs":1775582513,"mtime_nanos":318646536,"lines":158},"src/storied/session.py":{"mtime_secs":1768056565,"mtime_nanos":259214128,"lines":264},"src/storied/search.py":{"mtime_secs":1774713471,"mtime_nanos":353186743,"lines":408},"src/storied/tools/entities.py":{"mtime_secs":1775582467,"mtime_nanos":786135738,"lines":466},"src/storied/log.py":{"mtime_secs":1775581843,"mtime_nanos":65667046,"lines":540},"src/storied/engine.py":{"mtime_secs":1775581998,"mtime_nanos":312921199,"lines":523},"src/storied/srd/split.py":{"mtime_secs":1766849074,"mtime_nanos":657046597,"lines":336},"src/storied/srd/download.py":{"mtime_secs":1766845896,"mtime_nanos":726572037,"lines":49},"src/storied/tools/scene.py":{"mtime_secs":1775582498,"mtime_nanos":739484677,"lines":215},"src/storied/tools/__init__.py":{"mtime_secs":1775582704,"mtime_nanos":806661949,"lines":197},"src/storied/character.py":{"mtime_secs":1774487537,"mtime_nanos":474285279,"lines":403},"src/storied/claude.py":{"mtime_secs":1774458748,"mtime_nanos":52161247,"lines":377},"src/storied/srd/extract.py":{"mtime_secs":1766849840,"mtime_nanos":197357744,"lines":63},"src/storied/srd/__init__.py":{"mtime_secs":1766869155,"mtime_nanos":624121322,"lines":15},"src/storied/mcp_server.py":{"mtime_secs":1775581792,"mtime_nanos":695257450,"lines":198}}} \ No newline at end of file diff --git a/prompts/dm-system.md b/prompts/dm-system.md --- a/prompts/dm-system.md +++ b/prompts/dm-system.md @@ -298,9 +298,22 @@ - **After damage/healing**: `{"hp.current": 5}` - **After spending/gaining coins**: `{"purse.gp": 25}` or `{"purse.sp": 10, "purse.cp": 50}` - **After using abilities**: `{"features.0.uses": 0}` (e.g., Second Wind) - **After gaining/losing equipment**: `{"section.Equipment": "- Longsword\n- New shield"}` -- **After leveling up**: `{"level": 2, "hp.max": 20}` +- **After leveling up**: `{"level": 2, "hp.max": 20, "level_since": "#d5-1430", "advancement_ready": null}` Call update_character immediately when these changes happen, not at the end of the session. + +## Level Advancement + +A background system evaluates whether the character has earned a level-up. When it decides they have, you'll see an `advancement_ready` field on the character sheet and a notification in "Recent World Changes." + +**When `advancement_ready` is set:** +1. Don't rush it — find a narratively appropriate moment (a rest, a quiet pause, after a triumph) +2. Use `recall` to look up the character's class features for the new level +3. Narrate the growth as part of the story — the character reflects on what they've learned, feels a new confidence, discovers a new ability +4. Call `update_character` with the new level, updated HP, and any new features. Clear the flag: `{"level": , "hp.max": , "level_since": "", "advancement_ready": null}` +5. Include `level` in the `set_scene` tags for this moment: `tags=["level"]` + +Don't announce it mechanically ("You've reached level 5!"). Weave it into the fiction. The player should feel their character growing, not see a UI popup. ## Session State @@ -321,6 +334,10 @@ present=["[[Vera Blackwater]] - waiting for results", "[[Henrik]] - pretending not to listen"], threads=["Warehouse investigation - Vera's job", "Merchant attacks - 50gp from Captain"] ) ``` + +## Recent World Changes + +Your context may include a "Recent World Changes" section with notifications from background systems — the world planner, the advancement evaluator, or the world ticker. These are things that happened between turns. Weave them naturally into your narration when appropriate. Don't announce them as system messages. ## The Entity Model diff --git a/prompts/planner-system.md b/prompts/planner-system.md --- a/prompts/planner-system.md +++ b/prompts/planner-system.md @@ -9,6 +9,7 @@ |------|---------| | `recall` | Look up rules or existing world content | | `establish` | Create or update entities (NPCs, locations, items, threads) | | `mark` | Record backstory events in an entity's history | +| `notify_dm` | Send a brief summary of what you changed to the DM | ## What to Do @@ -67,4 +68,4 @@ - Don't contradict the current session state or campaign log. - Don't create entities far from the current location — focus on what's nearby. - Don't write prose or narrative. Use the tools to update world state directly. -When you've enriched the candidates and established any worthwhile log mentions, stop. +When you've enriched the candidates and established any worthwhile log mentions, call `notify_dm` with a brief summary of what you changed — e.g., "Enriched [[Vera Blackwater]] with smuggling connections and added [[Dockmaster Voss]] near the harbor." Keep it to one or two sentences. Then stop. diff --git a/prompts/world-tick.md b/prompts/world-tick.md --- a/prompts/world-tick.md +++ b/prompts/world-tick.md @@ -7,6 +7,7 @@ |------|---------| | `recall` | Look up existing world content | | `establish` | Update entities that changed | | `mark` | Record events that happened off-screen | +| `notify_dm` | Tell the DM what changed so they can weave it in | ## What You're Given @@ -58,4 +59,4 @@ - **Keep changes proportional to time passed.** A few hours? Almost nothing changes. A few days? NPCs might move, rumors spread. A week+? Factions act, situations evolve. - **Focus on entities near the player's last location.** Don't change things far away that the player can't observe. - **Use [[wikilinks]]** when referencing entities. -When you've evaluated the triggers and made appropriate changes, stop. +When you've evaluated the triggers and made appropriate changes, call `notify_dm` with a brief summary of what happened — e.g., "[[Captain Harrik]] left [[The Rusty Anchor]] to investigate the docks. A rumor about the missing merchants spread to [[Millford]]." Keep it to the things the DM might want to weave into narration. Then stop. diff --git a/prompts/xp-evaluator.md b/prompts/xp-evaluator.md new file mode 100644 --- /dev/null +++ b/prompts/xp-evaluator.md @@ -0,0 +1,85 @@ +You are a character advancement evaluator for a solo 5e adventure. You run silently in the background, reading the campaign journal to assess whether the player character has earned a level. + +## Your Job + +Read the campaign log since the character's last level-up (or the start of the campaign if they've never leveled). Evaluate the totality of what they've accomplished — combat encounters, quests completed, discoveries made, problems solved — and decide whether it's enough to warrant advancement to the next level. + +You are not tracking exact XP. You are making a holistic judgment, the way a good DM does with milestone leveling. But you should use the 5e XP tables as your calibration — they tell you roughly how much adventure separates one level from the next. + +## Calibration Reference + +**XP between levels (cumulative thresholds):** + +| From → To | XP Needed | Rough Equivalent | +|-----------|-----------|------------------| +| 1 → 2 | 300 | 2-4 easy encounters | +| 2 → 3 | 600 | A short adventure arc | +| 3 → 4 | 1,800 | A substantial quest line | +| 4 → 5 | 3,800 | A major story arc | +| 5 → 6 | 7,500 | An extended campaign chapter | +| 6 → 7 | 9,000 | Multiple significant quests | +| 7 → 8 | 11,000 | A major campaign milestone | +| 8 → 9 | 14,000 | Extended high-stakes adventure | +| 9 → 10 | 16,000 | Major campaign arc completion | + +**XP Budget per Character (solo, by encounter difficulty):** + +| Level | Low | Moderate | High | +|-------|-----|----------|------| +| 1 | 50 | 75 | 100 | +| 2 | 100 | 150 | 200 | +| 3 | 150 | 225 | 400 | +| 4 | 250 | 375 | 500 | +| 5 | 500 | 750 | 1,100 | +| 6 | 600 | 1,000 | 1,400 | +| 7 | 750 | 1,300 | 1,700 | +| 8 | 1,000 | 1,700 | 2,100 | +| 9 | 1,300 | 2,000 | 2,600 | +| 10 | 1,600 | 2,300 | 3,100 | + +**What kinds of things are worth XP:** +- Combat encounters (scale by difficulty — a bar fight is not the same as a boss) +- Completing quest threads (minor ~100-300, moderate ~500-1,000, major ~1,500-3,000) +- Significant discoveries (finding a hidden dungeon, uncovering a conspiracy — 50-200) +- Clever problem-solving (bypassing an encounter through wit, diplomacy, stealth — same as defeating it) +- Surviving dangerous situations (escaping a trap, navigating a hazard — 50-200) + +## Pacing Check + +Look at the advancement history (previous `level` tags in the log). Is there a pattern? Try to maintain a rhythm that feels natural: +- Too fast: leveling every 1-2 game days feels rushed. The character should earn each level. +- Too slow: going 10+ game days at low levels without advancement feels stagnant. If significant accomplishments are piling up, it's time. +- Just right: significant story arcs and challenges punctuate each level. The player should feel like advancement was earned through what they did. + +At low levels (1-4), advancement should come faster. At higher levels (5+), it should slow down. This matches the XP curve — the gaps between levels grow. + +## Your Tools + +| Tool | Purpose | +|------|---------| +| `recall` | Look up rules (class features for next level, SRD content) | +| `update_character` | Set `advancement_ready` on the character sheet | +| `notify_dm` | Send a notification the DM will see next turn | + +## What to Do + +1. Read the character's current level and class +2. Use `recall` to look up the character's class in the SRD — check what features the next level grants and any general leveling guidance +3. Read the campaign log since the last `level` tag (or from the beginning) +4. Mentally tally what the character has accomplished against the calibration table +5. Consider the pacing of previous level-ups +6. Make your call: + +**If NOT ready:** Do nothing. Stop. Don't output anything. + +**If ready:** Call both tools: +- `update_character({"advancement_ready": })` +- `notify_dm("The character has earned enough experience to reach level . . Level gains: . When a narratively appropriate moment arises, acknowledge their advancement.")` + +Then stop. Don't over-explain. Don't narrate. You're a background process, not a storyteller. + +## Important + +- Only recommend one level at a time. If they've earned multiple levels, recommend the next one. The evaluator will run again and catch the subsequent level. +- If `advancement_ready` is already set on the character sheet and the DM hasn't acted on it yet, do nothing. Don't stack notifications. +- Err slightly on the side of generosity. A player stuck at the same level for too long is worse than leveling one session too early. The goal is to keep the game feeling rewarding and the character's growth matching their story. diff --git a/src/storied/advancement.py b/src/storied/advancement.py new file mode 100644 --- /dev/null +++ b/src/storied/advancement.py @@ -0,0 +1,199 @@ +"""Character advancement evaluator — background agent that decides when to level up.""" + +import time +from collections.abc import Callable +from dataclasses import dataclass +from pathlib import Path +from threading import Thread + +from storied.character import format_character_context, load_character +from storied.claude import run_with_tools +from storied.engine import load_prompt +from storied.log import CampaignLog +from storied.mcp_server import start_server as start_mcp_server +from storied.session import load_session + + +@dataclass +class AdvancementResult: + """Result of an advancement evaluation run.""" + + evaluated: bool = False + tool_calls: int = 0 + input_tokens: int = 0 + output_tokens: int = 0 + elapsed: float = 0.0 + + +def build_advancement_context( + world_id: str, + player_id: str, + base_path: Path, +) -> str | None: + """Build context for the advancement evaluator. + + Returns None if there's nothing to evaluate (no character or + advancement_ready already set). + """ + character = load_character(player_id, base_path) + if character is None: + return None + + # If advancement_ready is already set, the DM hasn't acted yet — skip + if character.get("advancement_ready"): + return None + + parts: list[str] = [] + + # Character summary + char_context = format_character_context(character) + parts.append(char_context) + + # Campaign log — entries since last level-up + log = CampaignLog(world_id, base_path) + parts.append(f"## Campaign Time: {log.get_current_time()}") + + entries_since_level = log.get_entries_since_tag("level") + if entries_since_level: + lines = ["## Campaign Log (since last level-up)", ""] + for entry in entries_since_level: + tag_str = f" [{', '.join(entry.tags)}]" if entry.tags else "" + lines.append(f"- {entry.anchor} | {entry.event}{tag_str}") + parts.append("\n".join(lines)) + else: + parts.append("## Campaign Log\n\nNo events recorded yet.") + + # Previous level-ups for pacing reference + level_entries = log.find_tag_entries("level") + if level_entries: + lines = ["## Advancement History", ""] + for entry in level_entries: + lines.append(f"- {entry.anchor} | {entry.event}") + parts.append("\n".join(lines)) + + # Session state for thread context + session = load_session(player_id, base_path) + if session: + body = session.get("body", "") + if body: + parts.append(f"## Session State\n\n{body}") + + return "\n\n---\n\n".join(parts) + + +def evaluate_advancement( + world_id: str = "default", + player_id: str = "default", + base_path: Path | None = None, + model: str = "claude-opus-4-6", + on_progress: Callable[[str], None] | None = None, +) -> AdvancementResult: + """Evaluate whether the character has earned a level-up.""" + if base_path is None: + base_path = Path.cwd() + + def progress(msg: str) -> None: + if on_progress: + on_progress(msg) + + start_time = time.monotonic() + + context = build_advancement_context(world_id, player_id, base_path) + if context is None: + progress("Skipped (no character or advancement already pending)") + return AdvancementResult(elapsed=time.monotonic() - start_time) + + system_prompt = load_prompt("xp-evaluator") + + campaign_log = CampaignLog(world_id, base_path) + mcp = start_mcp_server( + world_id, player_id, base_path, "advancement", campaign_log, + ) + + progress(f"Evaluating advancement with {model}...") + + def on_tool(name: str) -> None: + if on_progress: + on_progress(f" [{name}]") + + claude_result = run_with_tools( + system_prompt=system_prompt, + user_message=context, + mcp_url=mcp.url, + model=model, + on_tool_start=on_tool, + cwd=base_path, + ) + + result = AdvancementResult( + evaluated=True, + elapsed=time.monotonic() - start_time, + ) + + if claude_result: + result.tool_calls = claude_result.usage.get("tool_calls", 0) + result.input_tokens = claude_result.usage.get("input_tokens", 0) + result.output_tokens = claude_result.usage.get("output_tokens", 0) + + return result + + +class BackgroundAdvancement: + """Runs advancement evaluation in a background thread during gameplay. + + Triggers every N player turns. Only one evaluation runs at a time. + """ + + def __init__( + self, + world_id: str, + player_id: str, + base_path: Path, + model: str = "claude-opus-4-6", + interval: int = 5, + ): + self._world_id = world_id + self._player_id = player_id + self._base_path = base_path + self._model = model + self._interval = interval + self._turn_count: int = 0 + self._thread: Thread | None = None + self._result: AdvancementResult | None = None + + def on_turn(self) -> None: + """Call after each player turn. Launches evaluation when interval is reached.""" + self._turn_count += 1 + if self._turn_count < self._interval: + return + self._maybe_evaluate() + + def on_combat_end(self) -> None: + """Call when initiative ends. Combat is a strong advancement signal.""" + self._maybe_evaluate() + + def _maybe_evaluate(self) -> None: + """Launch a background evaluation if one isn't already running.""" + if self._thread and self._thread.is_alive(): + return + + self._turn_count = 0 + self._result = None + self._thread = Thread(target=self._run, daemon=True) + self._thread.start() + + def _run(self) -> None: + self._result = evaluate_advancement( + world_id=self._world_id, + player_id=self._player_id, + base_path=self._base_path, + model=self._model, + ) + + def pop_result(self) -> AdvancementResult | None: + """Return and clear the last completed evaluation result, if any.""" + if self._thread and not self._thread.is_alive() and self._result: + result = self._result + self._result = None + return result + return None diff --git a/src/storied/cli.py b/src/storied/cli.py --- a/src/storied/cli.py +++ b/src/storied/cli.py @@ -312,7 +312,9 @@ engine.debug = args.debug # Background ticker for mid-session world advancement ticker = None + advancement = None if not creation_mode and not sandbox: + from storied.advancement import BackgroundAdvancement from storied.planner import BackgroundTicker ticker = BackgroundTicker( @@ -322,6 +324,12 @@ base_path=base_path or Path.cwd(), ) # Kick off initial tick in background ticker.maybe_tick(engine._campaign_log) + + advancement = BackgroundAdvancement( + world_id=world_id, + player_id=player_id, + base_path=base_path or Path.cwd(), + ) # If in creation mode, start the conversation if creation_mode: @@ -572,6 +580,19 @@ ) # Maybe launch a new tick if the day advanced ticker.maybe_tick(engine._campaign_log) + # Advancement evaluator — tick the turn counter and check results + if advancement: + if engine.combat_ended: + advancement.on_combat_end() + engine.combat_ended = False + else: + advancement.on_turn() + adv_result = advancement.pop_result() + if adv_result and adv_result.tool_calls > 0: + console.print( + "[dim]The character reflects on recent experiences...[/dim]" + ) + # Check if session ended (player quit gracefully) if engine.session_ended: console.print( @@ -630,74 +651,6 @@ return 0 -def cmd_plan(args: argparse.Namespace) -> int: - """Enrich thin entities near the player's current position.""" - from storied.planner import plan_world - - world_id = args.world or "default" - player_id = args.player or "default" - - print(f"Loading session for player '{player_id}' in world '{world_id}'...", flush=True) - - def on_progress(msg: str) -> None: - print(msg, flush=True) - - result = plan_world( - world_id=world_id, - player_id=player_id, - model=args.model, - threshold=args.threshold, - max_entities=args.max_entities, - dry_run=args.dry_run, - on_progress=on_progress, - ) - - if not result.candidates: - print("No thin entities found nearby. The world looks rich enough.") - return 0 - - if result.dry_run: - return 0 - - print( - f"Done — {result.tool_calls} tool calls, " - f"{result.input_tokens:,} in / {result.output_tokens:,} out, " - f"{result.elapsed:.1f}s", - flush=True, - ) - - return 0 - - -def cmd_tick(args: argparse.Namespace) -> int: - """Advance the world by evaluating Will triggers.""" - from storied.planner import tick_world - - world_id = args.world or "default" - player_id = args.player or "default" - - def on_progress(msg: str) -> None: - print(msg, flush=True) - - result = tick_world( - world_id=world_id, - player_id=player_id, - model=args.model, - on_progress=on_progress, - ) - - if result.entities_checked == 0: - print("No entities with active Will triggers.") - return 0 - - print( - f"Done — {result.tool_calls} tool calls, " - f"{result.input_tokens:,} in / {result.output_tokens:,} out, " - f"{result.elapsed:.1f}s", - flush=True, - ) - return 0 - def cmd_index_srd(args: argparse.Namespace) -> int: """Build search index for SRD content.""" @@ -922,61 +875,6 @@ action="store_true", help="Skip confirmation prompt", ) reset_parser.set_defaults(func=cmd_reset) - - # plan command - plan_parser = subparsers.add_parser("plan", help="Enrich thin world entities near the player") - plan_parser.add_argument( - "--world", "-w", - default="default", - help="World ID (default: default)", - ) - plan_parser.add_argument( - "--player", "-p", - default="default", - help="Player ID (default: default)", - ) - plan_parser.add_argument( - "--model", "-m", - default="claude-opus-4-6", - help="Model to use for planning (default: claude-opus-4-6)", - ) - plan_parser.add_argument( - "--threshold", - type=float, - default=0.7, - help="Richness threshold — entities below this score get enriched (default: 0.7)", - ) - plan_parser.add_argument( - "--max-entities", - type=int, - default=8, - help="Maximum entities to enrich per run (default: 8)", - ) - plan_parser.add_argument( - "--dry-run", - action="store_true", - help="Show what would be enriched without calling the API", - ) - plan_parser.set_defaults(func=cmd_plan) - - # tick command - tick_parser = subparsers.add_parser("tick", help="Advance the world by evaluating Will triggers") - tick_parser.add_argument( - "--world", "-w", - default="default", - help="World ID (default: default)", - ) - tick_parser.add_argument( - "--player", "-p", - default="default", - help="Player ID (default: default)", - ) - tick_parser.add_argument( - "--model", "-m", - default="claude-opus-4-6", - help="Model to use for ticking (default: claude-opus-4-6)", - ) - tick_parser.set_defaults(func=cmd_tick) # seed command seed_parser = subparsers.add_parser("seed", help="Seed an empty world from a character sheet") diff --git a/src/storied/engine.py b/src/storied/engine.py --- a/src/storied/engine.py +++ b/src/storied/engine.py @@ -8,6 +8,7 @@ from pathlib import Path import yaml +from storied import notifications from storied.character import format_character_context, load_character from storied.claude import ( Result, @@ -107,9 +108,10 @@ self._last_result: Result | None = None self._total_input_tokens: int = 0 self._total_output_tokens: int = 0 - # Session end / character creation flags + # Session end / character creation / combat end flags self.session_ended: bool = False self.character_created: bool = False + self.combat_ended: bool = False # Debug mode for verbose tool output self.debug: bool = False @@ -255,6 +257,16 @@ self._context_parts[f"Linked:{name}"] = entity_context parts.append(entity_context) loaded_names.add(name) + # Notifications from background agents (planner, ticker, advancement) + if self.world_id: + pending = notifications.drain(self.world_id, self.base_path) + if pending: + notif_lines = ["## Recent World Changes\n"] + notif_lines.extend(f"- {msg}" for msg in pending) + notif_context = "\n".join(notif_lines) + self._context_parts["Notifications"] = notif_context + parts.append(notif_context) + # Initiative state (injected when active so the DM never loses track) if self._mcp.ctx.initiative.active: initiative_context = self._mcp.ctx.initiative.format_for_context() @@ -437,6 +449,8 @@ if short == "end_session": self.session_ended = True if short == "create_character": self.character_created = True + if short == "end_initiative": + self.combat_ended = True if short in ("roll", "run_code") and not self.debug: # Signal the CLI to flush the renderer before we diff --git a/src/storied/log.py b/src/storied/log.py --- a/src/storied/log.py +++ b/src/storied/log.py @@ -404,6 +404,37 @@ lines.append(f"- {entry.event}") return "\n".join(lines) + def get_all_entries(self) -> list[LogEntry]: + """Get every log entry from day 1 through the current day.""" + entries: list[LogEntry] = [] + for day in range(1, self.current_day + 1): + entries.extend(self._load_day_entries(day)) + return entries + + def get_entries_since_tag(self, tag: str) -> list[LogEntry]: + """Get all entries after the last occurrence of a tag. + + Scans backwards through the entire log to find the most recent entry + with the given tag, then returns everything after it. If the tag is + never found, returns all entries. + """ + all_entries = self.get_all_entries() + + # Find last occurrence of the tag + last_idx = -1 + for i, entry in enumerate(all_entries): + if tag in entry.tags: + last_idx = i + + if last_idx == -1: + return all_entries + + return all_entries[last_idx + 1:] + + def find_tag_entries(self, tag: str) -> list[LogEntry]: + """Find all entries with a given tag, across all days.""" + return [e for e in self.get_all_entries() if tag in e.tags] + def time_since_rest(self, rest_type: str = "short") -> Duration: """Calculate time since last rest of given type.""" tag = f"rest:{rest_type}" diff --git a/src/storied/mcp_server.py b/src/storied/mcp_server.py --- a/src/storied/mcp_server.py +++ b/src/storied/mcp_server.py @@ -26,11 +26,13 @@ INITIATIVE_KEEP_NARRATIVE, ) from storied.sandbox import build_tool_signatures from storied.tools import ( + ADVANCEMENT_TOOL_DEFINITIONS, EntityIndex, PLANNER_TOOL_DEFINITIONS, SEEDER_TOOL_DEFINITIONS, TOOL_DEFINITIONS, ToolContext, + advancement_execute_tool, execute_tool, planner_execute_tool, seeder_execute_tool, @@ -41,12 +43,14 @@ TOOL_SETS: dict[str, list[dict]] = { "planner": PLANNER_TOOL_DEFINITIONS, "seeder": SEEDER_TOOL_DEFINITIONS, + "advancement": ADVANCEMENT_TOOL_DEFINITIONS, } EXECUTORS = { "dm": execute_tool, "planner": planner_execute_tool, "seeder": seeder_execute_tool, + "advancement": advancement_execute_tool, } diff --git a/src/storied/notifications.py b/src/storied/notifications.py new file mode 100644 --- /dev/null +++ b/src/storied/notifications.py @@ -0,0 +1,49 @@ +"""Notification channel for background agents to communicate with the DM. + +Background agents (planner, ticker, advancement evaluator) append messages. +The DM engine reads and clears them each turn via _build_context. +""" + +import threading +from pathlib import Path + + +_lock = threading.Lock() + + +def _notifications_path(world_id: str, base_path: Path) -> Path: + return base_path / "worlds" / world_id / "dm_notifications.md" + + +def append(world_id: str, base_path: Path, message: str) -> None: + """Append a notification for the DM to see next turn.""" + path = _notifications_path(world_id, base_path) + with _lock: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("a") as f: + f.write(f"- {message}\n") + + +def drain(world_id: str, base_path: Path) -> list[str]: + """Read all pending notifications and clear the file. + + Returns a list of notification messages (without the leading "- "). + """ + path = _notifications_path(world_id, base_path) + with _lock: + if not path.exists(): + return [] + + content = path.read_text().strip() + if not content: + path.unlink(missing_ok=True) + return [] + + # Clear the file + path.unlink(missing_ok=True) + + return [ + line.lstrip("- ").strip() + for line in content.splitlines() + if line.strip() + ] diff --git a/src/storied/planner.py b/src/storied/planner.py --- a/src/storied/planner.py +++ b/src/storied/planner.py @@ -167,7 +167,6 @@ elapsed: float = 0.0 dry_run: bool = False - def plan_world( world_id: str = "default", player_id: str = "default", diff --git a/src/storied/tools.py b/src/storied/tools.py deleted file mode 100644 --- a/src/storied/tools.py +++ /dev/null @@ -1,1155 +0,0 @@ -"""DM tools for Claude to use during gameplay. - -These functions are exposed to Claude as tools. The docstrings become -the tool descriptions that Claude sees. -""" - -import re -import threading -from dataclasses import dataclass, field -from pathlib import Path - -import yaml - -from storied.character import create_character as char_create -from storied.initiative import ( - ALL_INITIATIVE_TOOL_NAMES, - InitiativeTracker, - execute_initiative_tool, -) -from storied.character import update_character as char_update -from storied.dice import roll as dice_roll -from storied.log import CampaignLog -from storied.search import VectorIndex -from storied.session import name_to_slug -from storied.session import update_session as session_update - -# Per-file locks for thread-safe entity writes (establish, mark) -_file_locks: dict[Path, threading.Lock] = {} -_file_locks_lock = threading.Lock() - - -def _get_file_lock(path: Path) -> threading.Lock: - """Get or create a lock for the given file path.""" - with _file_locks_lock: - if path not in _file_locks: - _file_locks[path] = threading.Lock() - return _file_locks[path] - - -class EntityIndex: - """Name→path index with write-through entity cache. - - Built once at startup by globbing the world directory. Lookups are - dict lookups instead of 6 sequential Path.exists() calls. The cache - stores parsed entity dicts so repeated loads within a turn skip disk I/O. - establish() and mark() update both the index and cache on write. - """ - - def __init__(self, world_dir: Path | None = None): - self._paths: dict[str, Path] = {} - self._cache: dict[Path, dict] = {} - if world_dir and world_dir.exists(): - for md in world_dir.rglob("*.md"): - self._paths[md.stem] = md - - def resolve(self, name: str) -> Path | None: - """Look up an entity's file path by name.""" - return self._paths.get(name) - - def register(self, name: str, path: Path) -> None: - """Register or update an entity's path in the index.""" - self._paths[name] = path - - def cache_get(self, path: Path) -> dict | None: - """Get a cached parsed entity, or None if not cached.""" - return self._cache.get(path) - - def cache_put(self, path: Path, data: dict) -> None: - """Store a parsed entity in the cache.""" - self._cache[path] = data - - -@dataclass -class ToolContext: - """Shared infrastructure for all tool calls. - - Created once per MCP server and passed to every tool invocation. - All fields are required — no defensive None checks in tool code. - """ - - world_id: str - player_id: str - base_path: Path - campaign_log: CampaignLog - entity_index: EntityIndex - vector_index: VectorIndex - initiative: InitiativeTracker = field(default_factory=InitiativeTracker) - - -def _sync_player_hp(target: str, ctx: ToolContext, result: str) -> str: - """Auto-sync player character sheet when damage/heal targets a player.""" - combatant = ctx.initiative._find(target) - if combatant and combatant.is_player: - char_update(ctx.player_id, {"hp.current": combatant.hp}, ctx.base_path) - result += f" (character sheet synced to {combatant.hp} HP)" - return result - - -def roll(notation: str, reason: str | None = None) -> dict: - """Roll dice using standard notation like '1d20', '2d6+3', '4d6kh3'. - - Use for attack rolls, skill checks, saving throws, and damage rolls. - Supports: XdY, XdY+Z, XdY-Z, advantage (2d20kh1), disadvantage (2d20kl1). - - Args: - notation: Dice notation string (e.g., "1d20+5", "2d6", "4d6kh3") - reason: Brief description of what the roll is for (e.g., "Athletics", - "Attack with longsword", "Wisdom save", "Fireball damage") - - Returns: - Dict with rolls, kept dice, modifier, and total - """ - result = dice_roll(notation) - return result.to_dict() - - -def recall( - query: str, - ctx: ToolContext, - scope: str = "all", - content_type: str | None = None, -) -> str: - """Look up rules, world content, or both. - - Use to recall information about: - - Rules: spells, monsters, classes, items, conditions from the SRD - - World: NPCs, locations, factions, lore you've established - - Both: search everything (default) - - Args: - query: What to look up (e.g., "fireball", "captain vex", "merchant guild") - scope: Where to search - "rules", "world", or "all" (default) - content_type: Optional type to limit search (e.g., "spells", "npcs") - - Returns: - Content of the found item, or a message if not found - """ - # scope="rules" → SRD only; scope="world" → everything except SRD; - # scope="all" → no filter - source_filter: str | None = None - if scope == "rules": - source_filter = "srd" - - current_day = ctx.campaign_log.get_current_time().day - hits = ctx.vector_index.search( - query, limit=5, source_filter=source_filter, - exclude_source="srd" if scope == "world" else None, - decay_ref=current_day, - ) - if hits: - if len(hits) == 1 or hits[0].score > 0.8: - hit_path = Path(hits[0].path) - if hit_path.exists(): - return hit_path.read_text() - lines = [f"Found {len(hits)} matches:"] - for h in hits: - lines.append(f" - {h.doc_id.split(':')[1]} ({h.source}): {h.snippet[:80]}") - return "\n".join(lines) - - return f"Nothing found matching '{query}'" - - -def update_character(updates: dict, ctx: ToolContext) -> str: - """Update the player's character sheet to persist changes. - - Call this after HP changes, equipment gained/lost, coins spent, level ups, etc. - This ensures progress is saved and survives between sessions. - - Args: - updates: Fields to update. Use dot notation for nested fields. - Examples: - - {"hp.current": 5} - set current HP to 5 - - {"purse.gp": 25} - set gold to 25 - - {"purse.sp": 10, "purse.cp": 50} - set silver and copper - - {"level": 2, "hp.max": 20} - level up - For markdown sections, use "section.Name": - - {"section.Equipment": "- Longsword\\n- New shield"} - player_id: Player identifier (usually "default") - base_path: Base path for players directory - - Returns: - Confirmation of what was updated - """ - return char_update(ctx.player_id, updates, ctx.base_path) - - -def create_character( - name: str, - race: str, - char_class: str, - level: int, - abilities: dict[str, int], - hp_max: int, - ac: int, - ctx: ToolContext, - background: str | None = None, - speed: int = 30, - purse: dict[str, int] | None = None, - equipment: list[str] | None = None, - features: list[str] | None = None, - proficiencies: str | None = None, - backstory: str | None = None, -) -> str: - """Create a new player character and save to disk. - - Call this when character creation is complete. Include all the mechanical - details needed to play: abilities, HP, AC, equipment, and features. - - Args: - name: Character name - race: Race (e.g., "Human", "High Elf", "Hill Dwarf") - char_class: Class (e.g., "Fighter", "Wizard", "Rogue") - level: Starting level (usually 1) - abilities: All six ability scores as a dict: - {"strength": 15, "dexterity": 14, "constitution": 13, - "intelligence": 12, "wisdom": 10, "charisma": 8} - hp_max: Maximum hit points - ac: Armor class - background: Background (e.g., "Soldier", "Sage", "Criminal") - speed: Movement speed in feet (default 30) - purse: Starting coins by denomination: - {"cp": 0, "sp": 0, "ep": 0, "gp": 15, "pp": 0} - Denominations: cp (copper), sp (silver), ep (electrum), - gp (gold), pp (platinum). Omit denominations for 0. - equipment: List of equipment items - features: List of racial and class features - proficiencies: Description of proficiencies (armor, weapons, tools, saves, skills) - backstory: Character backstory and personality - - Returns: - Confirmation message - """ - return char_create( - player_id=ctx.player_id, - name=name, - race=race, - char_class=char_class, - level=level, - abilities=abilities, - hp_max=hp_max, - ac=ac, - background=background, - speed=speed, - purse=purse, - equipment=equipment, - features=features, - proficiencies=proficiencies, - backstory=backstory, - base_path=ctx.base_path, - ) - - -def set_scene( - ctx: ToolContext, - event: str | None = None, - duration: str | None = None, - situation: str | None = None, - location: str | None = None, - present: list[str] | None = None, - threads: list[str] | None = None, - tags: list[str] | None = None, -) -> str: - """Call this after every response. Logs what happened, advances the clock, - and updates the scene state. - - Always include event and duration — time only advances when you log it. - - Args: - event: What happened this turn (e.g., "Spoke with the innkeeper", - "Searched the warehouse", "Fought off thugs"). - duration: How long it took (e.g., "10 min", "1 hour", "3 rounds", - "1 day"). The clock advances by this amount. - situation: Updated situation summary. Write in present tense as if - briefing another DM taking over mid-session. - location: New location when the player moves (e.g., "The Rusty Anchor") - present: Entities currently present, using [[Name]] format - (e.g., ["[[Vera Blackwater]]", "[[Henrik]] - barkeep"]) - threads: Open plot threads or objectives - tags: Optional tags: "combat", "rest:short", "rest:long", "travel" - - Returns: - Confirmation of what was updated - """ - parts = [] - - if event and duration: - anchor = ctx.campaign_log.append_entry(event, duration, tags=tags) - current = ctx.campaign_log.get_current_time() - parts.append( - f"Logged: {anchor} | {event} | {duration} → " - f"Now: {current} ({current.period_of_day()}, {current.atmosphere()})" - ) - - updates = {} - if situation is not None: - updates["situation"] = situation - if location is not None: - updates["location"] = location - if present is not None: - updates["present"] = present - if threads is not None: - updates["threads"] = threads - - if updates: - result = session_update(ctx.player_id, updates, ctx.base_path) - parts.append(result) - - if event and present: - marked = _auto_mark_present(present, event, ctx) - if marked: - parts.append(f"Auto-marked: {', '.join(marked)}") - - return "; ".join(parts) if parts else "No updates" - - -def establish( - entity_type: str, - name: str, - ctx: ToolContext, - description: str | None = None, - location: str | None = None, - knows: list[str] | None = None, - wants: list[str] | None = None, - will: list[str] | None = None, -) -> str: - """Establish or update an entity in the world. - - Use to create NPCs, locations, items, factions, or threads with their inner - state. Everything has Knows/Wants/Will: - - **Knows** = secrets, hidden truths, what isn't obvious - - **Wants** = nature, tendencies, inclinations (even non-sentient things can "want") - - **Will** = conditional triggers, what happens if... - - This isn't literal consciousness - it's narrative tendency. A bridge can "want" - to collapse. Cursed gold "wants" to be spent. Frame it this way and the world - feels alive. - - Partial updates: omit fields to preserve existing content when updating. - - Args: - entity_type: Type of entity: npcs, locations, items, factions, threads, lore - name: Display name (e.g., "Vera Blackwater", "The Rusty Anchor") - This becomes the filename directly (no slugification). - description: Prose description for the ## Is section. Include appearance, - background, current state, relationships via [[wikilinks]]. - location: Where this entity is right now. Can be a simple wikilink like - "[[The Rusty Anchor]]" or a verbal description like "In the basement - of [[The Rusty Anchor]]" or "Wandering the docks of [[Greyhaven]]". - knows: List of secrets and hidden truths. Things that aren't obvious. - wants: List of desires, tendencies, inclinations. The entity's nature. - will: List of conditional behaviors: "If X → Y" format. - - Returns: - Confirmation with the file path - """ - world_dir = ctx.base_path / "worlds" / ctx.world_id / entity_type - world_dir.mkdir(parents=True, exist_ok=True) - file_path = world_dir / f"{name}.md" - - lock = _get_file_lock(file_path) - with lock: - existing = _load_entity(file_path, ctx.entity_index) - - if description is None: - description = existing.get("description", "") - if location is None: - location = existing.get("location", "") - if knows is None: - knows = existing.get("knows", []) - if wants is None: - wants = existing.get("wants", []) - if will is None: - will = existing.get("will", []) - was = existing.get("was", []) - - data = { - "description": description, "location": location, - "knows": knows, "wants": wants, "will": will, "was": was, - } - _write_entity(file_path, name, entity_type, data, ctx) - - action = "Updated" if existing else "Established" - return f"{action} {entity_type.rstrip('s')} '{name}'" - - -def _load_entity(file_path: Path, entity_index: EntityIndex) -> dict: - """Load an existing entity file and parse its structure.""" - cached = entity_index.cache_get(file_path) - if cached is not None: - return cached - - if not file_path.exists(): - return {} - - content = file_path.read_text() - result = {} - - # Parse ## Is section - is_match = re.search(r"## Is\n\n?(.*?)(?=\n## |\Z)", content, re.DOTALL) - if is_match: - is_content = is_match.group(1).strip() - - # Extract location (line starting with **Location:**) - loc_match = re.search(r"\*\*Location:\*\*\s*(.+)", is_content) - if loc_match: - result["location"] = loc_match.group(1).strip() - - # Extract description (text before first ### subsection, excluding location line) - desc_match = re.match(r"(.*?)(?=\n### |\Z)", is_content, re.DOTALL) - if desc_match: - desc = desc_match.group(1).strip() - # Remove location line from description - desc = re.sub(r"\*\*Location:\*\*\s*.+\n?", "", desc).strip() - result["description"] = desc - - # Extract ### Knows - knows_match = re.search(r"### Knows\n\n?(.*?)(?=\n### |\n## |\Z)", is_content, re.DOTALL) - if knows_match: - result["knows"] = _parse_list_items(knows_match.group(1)) - - # Extract ### Wants - wants_match = re.search(r"### Wants\n\n?(.*?)(?=\n### |\n## |\Z)", is_content, re.DOTALL) - if wants_match: - result["wants"] = _parse_list_items(wants_match.group(1)) - - # Extract ### Will - will_match = re.search(r"### Will\n\n?(.*?)(?=\n### |\n## |\Z)", is_content, re.DOTALL) - if will_match: - result["will"] = _parse_list_items(will_match.group(1)) - - # Parse ## Was section - was_match = re.search(r"## Was\n\n?(.*?)(?=\n## |\Z)", content, re.DOTALL) - if was_match: - result["was"] = _parse_list_items(was_match.group(1)) - - entity_index.cache_put(file_path, result) - - return result - - -def _parse_list_items(text: str) -> list[str]: - """Parse markdown list items into a list of strings.""" - items = [] - for line in text.strip().split("\n"): - line = line.strip() - if line.startswith("- "): - items.append(line[2:]) - return items - - -def _format_entity( - name: str, - description: str, - location: str, - knows: list[str], - wants: list[str], - will: list[str], - was: list[str], -) -> str: - """Format an entity as markdown with Is/Was structure.""" - lines = [f"# {name}", "", "## Is", ""] - - if location: - lines.append(f"**Location:** {location}") - lines.append("") - - if description: - lines.append(description) - lines.append("") - - if knows: - lines.append("### Knows") - lines.append("") - for item in knows: - lines.append(f"- {item}") - lines.append("") - - if wants: - lines.append("### Wants") - lines.append("") - for item in wants: - lines.append(f"- {item}") - lines.append("") - - if will: - lines.append("### Will") - lines.append("") - for item in will: - lines.append(f"- {item}") - lines.append("") - - lines.append("## Was") - lines.append("") - for item in was: - lines.append(f"- {item}") - lines.append("") - - return "\n".join(lines) - - -def _write_entity( - file_path: Path, - name: str, - entity_type: str, - data: dict, - ctx: ToolContext, -) -> None: - """Write an entity to disk and update all indexes.""" - file_content = _format_entity( - name, data["description"], data["location"], - data["knows"], data["wants"], data["will"], data["was"], - ) - file_path.write_text(file_content) - ctx.entity_index.register(name, file_path) - ctx.entity_index.cache_put(file_path, data) - ctx.vector_index.upsert( - f"world:{entity_type}/{name}.md:0", - file_content, - {"source": "world", "content_type": entity_type, - "path": str(file_path), "title": name}, - ) - - -def _auto_mark_present( - present: list[str], event: str, ctx: ToolContext, -) -> list[str]: - """Auto-mark present entities with the current event. - - Extracts entity names from [[wikilink]] format in the present list - and appends the event to each entity's Was section. - """ - marked: list[str] = [] - for ref in present: - link_match = re.search(r"\[\[([^\]]+)\]\]", ref) - if not link_match: - continue - name = link_match.group(1) - - file_path = ctx.entity_index.resolve(name) - if file_path is None: - for etype in ("npcs", "locations", "items", "factions"): - candidate = ctx.base_path / "worlds" / ctx.world_id / etype / f"{name}.md" - if candidate.exists(): - file_path = candidate - break - - if file_path and file_path.exists(): - entity_type = file_path.parent.name - mark(entity_type=entity_type, name=name, event=event, ctx=ctx) - marked.append(name) - - return marked - - -def mark( - entity_type: str, - name: str, - event: str, - ctx: ToolContext, - resolves: list[str] | None = None, -) -> str: - """Record an event in an entity's history (## Was section). - - Use when something significant happens to or involving an entity. This builds - their history and helps maintain continuity across sessions. - - If the event resolves a Will trigger (e.g., "Vera introduced the player to - Harrik" resolves "If trusted → intro to Harrik"), provide the trigger text - in `resolves` to remove it from the Will section. - - Args: - entity_type: Type of entity: npcs, locations, items, factions, threads - name: Entity name (exact filename match) - event: What happened - brief description for the Was section - resolves: Optional list of Will items to remove if this event fired triggers - - Returns: - Confirmation message - """ - file_path = ctx.entity_index.resolve(name) - if file_path is None: - file_path = ctx.base_path / "worlds" / ctx.world_id / entity_type / f"{name}.md" - - if not file_path.exists(): - return f"Error: Entity '{name}' not found in {entity_type}" - - timestamp = ctx.campaign_log.get_current_time().to_anchor() - - lock = _get_file_lock(file_path) - with lock: - existing = _load_entity(file_path, ctx.entity_index) - - was = existing.get("was", []) - was.append(f"{timestamp} | {event}") - - will = existing.get("will", []) - resolved = [] - for trigger in resolves or []: - if trigger in will: - will.remove(trigger) - resolved.append(trigger) - - data = { - "description": existing.get("description", ""), - "location": existing.get("location", ""), - "knows": existing.get("knows", []), - "wants": existing.get("wants", []), - "will": will, - "was": was, - } - _write_entity(file_path, name, entity_type, data, ctx) - - result = f"Marked: {event}" - if resolved: - if len(resolved) == 1: - result += f" (resolved: {resolved[0]})" - else: - result += f" (resolved {len(resolved)} triggers)" - return result - - -def note_discovery( - entity: str, - content: str, - ctx: ToolContext, - content_type: str = "lore", - tags: list[str] | None = None, -) -> str: - """Record what the player has learned about something. - - Use when the player discovers or learns information about an NPC, - location, faction, or other world element. This captures their - perspective, which may be incomplete or even wrong. - - The player's knowledge is separate from DM truth (use `establish` - for the full facts). This helps track what the player knows vs. - what they haven't discovered yet. - - Args: - entity: Name of what they learned about (e.g., "Vera Blackwater") - content: What the player learned or observed - content_type: Type of content - npcs, locations, factions, lore (default: lore) - tags: Optional tags for categorization - - Returns: - Confirmation message - """ - slug = name_to_slug(entity) - - knowledge_dir = ( - ctx.base_path / "players" / ctx.player_id / "worlds" - / ctx.world_id / content_type - ) - knowledge_dir.mkdir(parents=True, exist_ok=True) - file_path = knowledge_dir / f"{slug}.md" - - frontmatter: dict = { - "type": content_type.rstrip("s"), - "name": entity, - } - if tags: - frontmatter["tags"] = tags - - file_content = "---\n" - file_content += yaml.dump(frontmatter, sort_keys=False, allow_unicode=True) - file_content += "---\n\n" - file_content += content.strip() - file_content += "\n" - - file_path.write_text(file_content) - - ctx.vector_index.upsert( - f"player:{content_type}/{slug}.md:0", - file_content, - {"source": "player", "content_type": content_type, - "path": str(file_path), "title": entity}, - ) - - return f"Noted: player learned about '{entity}'" - - -def tune(tuning: str, ctx: ToolContext) -> str: - """Update your storytelling style based on player feedback. - - Write the complete updated style as markdown prose. This replaces the - entire current style. Incorporate existing preferences where they still - apply — don't discard preferences the player hasn't contradicted. - """ - path = ctx.base_path / "worlds" / ctx.world_id / "style.md" - path.write_text(f"# Style\n\n{tuning}\n") - return "Style updated." - - -def end_session(situation: str, ctx: ToolContext, threads: list[str] | None = None) -> str: - """End the current session, saving the game state for next time. - - Call this when the player indicates they want to stop playing. This saves - the current situation so the next session can resume smoothly. - - Before calling, give a brief farewell and summary of what happened this session. - - Args: - situation: Summary of the current state of affairs for the next session. - Write as if briefing a DM who will pick up where you left off. - threads: Open plot threads or objectives to carry forward - - Returns: - Confirmation that session was saved - """ - updates: dict = {"situation": situation} - if threads is not None: - updates["threads"] = threads - - session_update(ctx.player_id, updates, ctx.base_path) - return "SESSION_ENDED" - - -# Tool definitions for the Anthropic API -TOOL_DEFINITIONS = [ - { - "name": "roll", - "description": roll.__doc__, - "input_schema": { - "type": "object", - "properties": { - "notation": { - "type": "string", - "description": "Dice notation (e.g., '1d20+5', '2d6', '4d6kh3')", - }, - "reason": { - "type": "string", - "description": "What the roll is for (e.g., 'Athletics', 'Longsword attack', 'Dex save')", - }, - }, - "required": ["notation", "reason"], - }, - }, - { - "name": "recall", - "description": recall.__doc__, - "input_schema": { - "type": "object", - "properties": { - "query": { - "type": "string", - "description": "What to look up (e.g., 'fireball', 'captain vex')", - }, - "scope": { - "type": "string", - "description": "Where to search: 'rules' (SRD), 'world' (established content), or 'all' (both)", - "enum": ["rules", "world", "all"], - }, - "content_type": { - "type": "string", - "description": "Type to limit search (e.g., 'spells', 'npcs', 'monsters')", - }, - }, - "required": ["query"], - }, - }, - { - "name": "update_character", - "description": update_character.__doc__, - "input_schema": { - "type": "object", - "properties": { - "updates": { - "type": "object", - "description": "Fields to update. Use dot notation for nested (e.g., 'hp.current': 5). Use 'section.Name' for markdown sections.", - }, - }, - "required": ["updates"], - }, - }, - { - "name": "create_character", - "description": create_character.__doc__, - "input_schema": { - "type": "object", - "properties": { - "name": {"type": "string", "description": "Character name"}, - "race": {"type": "string", "description": "Race (e.g., 'Human', 'High Elf')"}, - "char_class": {"type": "string", "description": "Class (e.g., 'Fighter', 'Wizard')"}, - "level": {"type": "integer", "description": "Starting level (usually 1)"}, - "abilities": { - "type": "object", - "description": "All six ability scores: strength, dexterity, constitution, intelligence, wisdom, charisma", - }, - "hp_max": {"type": "integer", "description": "Maximum hit points"}, - "ac": {"type": "integer", "description": "Armor class"}, - "background": {"type": "string", "description": "Background (e.g., 'Soldier', 'Sage')"}, - "speed": {"type": "integer", "description": "Movement speed in feet"}, - "purse": { - "type": "object", - "description": "Starting coins: {cp, sp, ep, gp, pp}. Omit denominations for 0.", - "properties": { - "cp": {"type": "integer", "description": "Copper pieces"}, - "sp": {"type": "integer", "description": "Silver pieces"}, - "ep": {"type": "integer", "description": "Electrum pieces"}, - "gp": {"type": "integer", "description": "Gold pieces"}, - "pp": {"type": "integer", "description": "Platinum pieces"}, - }, - }, - "equipment": { - "type": "array", - "items": {"type": "string"}, - "description": "List of equipment items", - }, - "features": { - "type": "array", - "items": {"type": "string"}, - "description": "List of racial and class features", - }, - "proficiencies": {"type": "string", "description": "Proficiency description"}, - "backstory": {"type": "string", "description": "Character backstory and personality"}, - }, - "required": ["name", "race", "char_class", "level", "abilities", "hp_max", "ac"], - }, - }, - { - "name": "set_scene", - "description": set_scene.__doc__, - "input_schema": { - "type": "object", - "properties": { - "event": { - "type": "string", - "description": "What happened this turn (logged to campaign journal)", - }, - "duration": { - "type": "string", - "description": "How long it took (e.g., '10 min', '1 hour', '3 rounds')", - }, - "situation": { - "type": "string", - "description": "Updated situation summary in present tense", - }, - "location": { - "type": "string", - "description": "New location when the player moves", - }, - "present": { - "type": "array", - "items": {"type": "string"}, - "description": "Entities present, using [[Name]] format", - }, - "threads": { - "type": "array", - "items": {"type": "string"}, - "description": "Open plot threads or objectives", - }, - "tags": { - "type": "array", - "items": {"type": "string"}, - "description": "Optional: 'combat', 'rest:short', 'rest:long', 'travel'", - }, - }, - "required": ["event", "duration"], - }, - }, - { - "name": "establish", - "description": establish.__doc__, - "input_schema": { - "type": "object", - "properties": { - "entity_type": { - "type": "string", - "description": "Type of entity", - "enum": ["npcs", "locations", "items", "factions", "threads", "lore"], - }, - "name": { - "type": "string", - "description": "Display name (exact filename, e.g., 'Vera Blackwater')", - }, - "description": { - "type": "string", - "description": "Prose description with [[wikilinks]] for relationships", - }, - "location": { - "type": "string", - "description": "Current location (e.g., '[[The Rusty Anchor]]' or 'In the basement of [[The Rusty Anchor]]')", - }, - "knows": { - "type": "array", - "items": {"type": "string"}, - "description": "Secrets, hidden truths - what isn't obvious", - }, - "wants": { - "type": "array", - "items": {"type": "string"}, - "description": "Nature, tendencies, inclinations - even non-sentient things", - }, - "will": { - "type": "array", - "items": {"type": "string"}, - "description": "Conditional behaviors in 'If X → Y' format", - }, - }, - "required": ["entity_type", "name"], - }, - }, - { - "name": "mark", - "description": mark.__doc__, - "input_schema": { - "type": "object", - "properties": { - "entity_type": { - "type": "string", - "description": "Type of entity", - "enum": ["npcs", "locations", "items", "factions", "threads"], - }, - "name": { - "type": "string", - "description": "Entity name (exact filename match)", - }, - "event": { - "type": "string", - "description": "What happened - brief description", - }, - "resolves": { - "type": "array", - "items": {"type": "string"}, - "description": "Optional: Will items to remove if this event fired triggers", - }, - }, - "required": ["entity_type", "name", "event"], - }, - }, - { - "name": "note_discovery", - "description": note_discovery.__doc__, - "input_schema": { - "type": "object", - "properties": { - "entity": { - "type": "string", - "description": "Name of what they learned about (e.g., 'Vera Blackwater')", - }, - "content": { - "type": "string", - "description": "What the player learned or observed", - }, - "content_type": { - "type": "string", - "description": "Type of content", - "enum": ["npcs", "locations", "factions", "lore"], - }, - "tags": { - "type": "array", - "items": {"type": "string"}, - "description": "Tags for categorization", - }, - }, - "required": ["entity", "content"], - }, - }, - { - "name": "tune", - "description": tune.__doc__, - "input_schema": { - "type": "object", - "properties": { - "tuning": { - "type": "string", - "description": "Complete updated style as markdown prose. Replaces the current style entirely.", - }, - }, - "required": ["tuning"], - }, - }, - { - "name": "end_session", - "description": end_session.__doc__, - "input_schema": { - "type": "object", - "properties": { - "situation": { - "type": "string", - "description": "Summary of current state for the next session", - }, - "threads": { - "type": "array", - "items": {"type": "string"}, - "description": "Open plot threads or objectives to carry forward", - }, - }, - "required": ["situation"], - }, - }, - { - "name": "run_code", - "description": ( - "Run Python code in a secure sandbox. Use for calculations, random " - "generation, data formatting, or any computation the narrative needs.\n\n" - "All your DM tools are callable as functions (see signatures below). " - "Most return a str with the result. The exception is roll(), which " - "returns a dict with keys: notation, rolls, kept, modifier, total — " - "use roll('2d6+3')['total'] for math, or index into rolls/kept for " - "individual dice. Use roll() for all randomness (no random module).\n\n" - "Language: variables, functions, loops, conditionals, comprehensions, " - "f-strings. Stdlib: re, json, datetime, math. No classes, no other " - "imports, no file/network access. Errors return as text.\n\n" - "Available functions:\n{tool_signatures}" - ), - "input_schema": { - "type": "object", - "properties": { - "description": { - "type": "string", - "description": "What this code does in game terms (e.g., 'Designing cave system', 'Splitting treasure', 'Generating NPC schedule')", - }, - "code": { - "type": "string", - "description": "Python code to execute", - }, - }, - "required": ["description", "code"], - }, - }, -] - - -def execute_tool(tool_name: str, tool_input: dict, ctx: ToolContext) -> str: - """Execute a tool by name with the given input.""" - if tool_name in ALL_INITIATIVE_TOOL_NAMES: - result = execute_initiative_tool(tool_name, tool_input, ctx.initiative) - if result is not None: - if tool_name in ("damage", "heal"): - result = _sync_player_hp(tool_input["target"], ctx, result) - return result - - if tool_name == "roll": - result = roll(tool_input["notation"]) - rolls_str = ", ".join(str(r) for r in result["rolls"]) - if result["kept"] != result["rolls"]: - kept_str = ", ".join(str(r) for r in result["kept"]) - return f"Rolled {result['notation']}: [{rolls_str}] → kept [{kept_str}] + {result['modifier']} = {result['total']}" - elif result["modifier"]: - return f"Rolled {result['notation']}: [{rolls_str}] + {result['modifier']} = {result['total']}" - else: - return f"Rolled {result['notation']}: [{rolls_str}] = {result['total']}" - - elif tool_name == "recall": - return recall( - tool_input["query"], ctx, - scope=tool_input.get("scope", "all"), - content_type=tool_input.get("content_type"), - ) - - elif tool_name == "update_character": - return update_character(tool_input["updates"], ctx) - - elif tool_name == "create_character": - return create_character( - name=tool_input["name"], - race=tool_input["race"], - char_class=tool_input["char_class"], - level=tool_input["level"], - abilities=tool_input["abilities"], - hp_max=tool_input["hp_max"], - ac=tool_input["ac"], - ctx=ctx, - background=tool_input.get("background"), - speed=tool_input.get("speed", 30), - purse=tool_input.get("purse"), - equipment=tool_input.get("equipment"), - features=tool_input.get("features"), - proficiencies=tool_input.get("proficiencies"), - backstory=tool_input.get("backstory"), - ) - - elif tool_name == "set_scene": - return set_scene( - ctx=ctx, - event=tool_input.get("event"), - duration=tool_input.get("duration"), - situation=tool_input.get("situation"), - location=tool_input.get("location"), - present=tool_input.get("present"), - threads=tool_input.get("threads"), - tags=tool_input.get("tags"), - ) - - elif tool_name == "establish": - return establish( - entity_type=tool_input["entity_type"], - name=tool_input["name"], - ctx=ctx, - description=tool_input.get("description"), - location=tool_input.get("location"), - knows=tool_input.get("knows"), - wants=tool_input.get("wants"), - will=tool_input.get("will"), - ) - - elif tool_name == "mark": - return mark( - entity_type=tool_input["entity_type"], - name=tool_input["name"], - event=tool_input["event"], - ctx=ctx, - resolves=tool_input.get("resolves"), - ) - - elif tool_name == "note_discovery": - return note_discovery( - entity=tool_input["entity"], - content=tool_input["content"], - ctx=ctx, - content_type=tool_input.get("content_type", "lore"), - tags=tool_input.get("tags"), - ) - - elif tool_name == "tune": - return tune(tuning=tool_input["tuning"], ctx=ctx) - - elif tool_name == "end_session": - return end_session( - situation=tool_input["situation"], - ctx=ctx, - threads=tool_input.get("threads"), - ) - - elif tool_name == "run_code": - from storied.sandbox import execute as sandbox_execute - return sandbox_execute(tool_input["code"], ctx=ctx) - - else: - return f"Unknown tool: {tool_name}" - - -PLANNER_TOOLS = {"recall", "establish", "mark"} - -PLANNER_TOOL_DEFINITIONS = [t for t in TOOL_DEFINITIONS if t["name"] in PLANNER_TOOLS] - - -def planner_execute_tool(tool_name: str, tool_input: dict, ctx: ToolContext) -> str: - """Execute a planner-allowed tool. Rejects anything outside the allowed set.""" - if tool_name not in PLANNER_TOOLS: - return f"Tool not available to planner: {tool_name}" - return execute_tool(tool_name, tool_input, ctx) - - -SEEDER_TOOLS = {"establish", "set_scene"} - -SEEDER_TOOL_DEFINITIONS = [t for t in TOOL_DEFINITIONS if t["name"] in SEEDER_TOOLS] - - -def seeder_execute_tool(tool_name: str, tool_input: dict, ctx: ToolContext) -> str: - """Execute a seeder-allowed tool. Rejects anything outside the allowed set.""" - if tool_name not in SEEDER_TOOLS: - return f"Tool not available to seeder: {tool_name}" - return execute_tool(tool_name, tool_input, ctx) diff --git a/src/storied/tools/__init__.py b/src/storied/tools/__init__.py new file mode 100644 --- /dev/null +++ b/src/storied/tools/__init__.py @@ -0,0 +1,197 @@ +"""DM tools for Claude to use during gameplay. + +Submodules group related tools by domain. This __init__ aggregates +definitions and re-exports public names so existing imports continue +to work unchanged. +""" + +from storied.initiative import ( + ALL_INITIATIVE_TOOL_NAMES, + execute_initiative_tool, +) +from storied.tools._context import ( + EntityIndex, + ToolContext, + _get_file_lock, + _sync_player_hp, +) +from storied.tools.character import ( + create_character, + update_character, +) +from storied.tools.entities import ( + _auto_mark_present, + _load_entity, + establish, + mark, + note_discovery, +) +from storied.tools.mechanics import ( + recall, + roll, +) +from storied.tools.scene import ( + end_session, + notify_dm, + set_scene, + tune, +) + +# Merge per-module DEFINITIONS lists into the canonical flat list +from storied.tools.character import DEFINITIONS as _CHAR_DEFS +from storied.tools.entities import DEFINITIONS as _ENTITY_DEFS +from storied.tools.mechanics import DEFINITIONS as _MECH_DEFS +from storied.tools.scene import DEFINITIONS as _SCENE_DEFS + +TOOL_DEFINITIONS: list[dict] = ( + _MECH_DEFS + _CHAR_DEFS + _SCENE_DEFS + _ENTITY_DEFS +) + + +def execute_tool(tool_name: str, tool_input: dict, ctx: ToolContext) -> str: + """Execute a tool by name with the given input.""" + if tool_name in ALL_INITIATIVE_TOOL_NAMES: + result = execute_initiative_tool(tool_name, tool_input, ctx.initiative) + if result is not None: + if tool_name in ("damage", "heal"): + result = _sync_player_hp(tool_input["target"], ctx, result) + return result + + if tool_name == "roll": + result = roll(tool_input["notation"]) + rolls_str = ", ".join(str(r) for r in result["rolls"]) + if result["kept"] != result["rolls"]: + kept_str = ", ".join(str(r) for r in result["kept"]) + return f"Rolled {result['notation']}: [{rolls_str}] → kept [{kept_str}] + {result['modifier']} = {result['total']}" + elif result["modifier"]: + return f"Rolled {result['notation']}: [{rolls_str}] + {result['modifier']} = {result['total']}" + else: + return f"Rolled {result['notation']}: [{rolls_str}] = {result['total']}" + + elif tool_name == "recall": + return recall( + tool_input["query"], ctx, + scope=tool_input.get("scope", "all"), + content_type=tool_input.get("content_type"), + ) + + elif tool_name == "update_character": + return update_character(tool_input["updates"], ctx) + + elif tool_name == "create_character": + return create_character( + name=tool_input["name"], + race=tool_input["race"], + char_class=tool_input["char_class"], + level=tool_input["level"], + abilities=tool_input["abilities"], + hp_max=tool_input["hp_max"], + ac=tool_input["ac"], + ctx=ctx, + background=tool_input.get("background"), + speed=tool_input.get("speed", 30), + purse=tool_input.get("purse"), + equipment=tool_input.get("equipment"), + features=tool_input.get("features"), + proficiencies=tool_input.get("proficiencies"), + backstory=tool_input.get("backstory"), + ) + + elif tool_name == "set_scene": + return set_scene( + ctx=ctx, + event=tool_input.get("event"), + duration=tool_input.get("duration"), + situation=tool_input.get("situation"), + location=tool_input.get("location"), + present=tool_input.get("present"), + threads=tool_input.get("threads"), + tags=tool_input.get("tags"), + ) + + elif tool_name == "establish": + return establish( + entity_type=tool_input["entity_type"], + name=tool_input["name"], + ctx=ctx, + description=tool_input.get("description"), + location=tool_input.get("location"), + knows=tool_input.get("knows"), + wants=tool_input.get("wants"), + will=tool_input.get("will"), + ) + + elif tool_name == "mark": + return mark( + entity_type=tool_input["entity_type"], + name=tool_input["name"], + event=tool_input["event"], + ctx=ctx, + resolves=tool_input.get("resolves"), + ) + + elif tool_name == "note_discovery": + return note_discovery( + entity=tool_input["entity"], + content=tool_input["content"], + ctx=ctx, + content_type=tool_input.get("content_type", "lore"), + tags=tool_input.get("tags"), + ) + + elif tool_name == "tune": + return tune(tuning=tool_input["tuning"], ctx=ctx) + + elif tool_name == "end_session": + return end_session( + situation=tool_input["situation"], + ctx=ctx, + threads=tool_input.get("threads"), + ) + + elif tool_name == "run_code": + from storied.sandbox import execute as sandbox_execute + return sandbox_execute(tool_input["code"], ctx=ctx) + + elif tool_name == "notify_dm": + return notify_dm(message=tool_input["message"], ctx=ctx) + + else: + return f"Unknown tool: {tool_name}" + + +# --- Tool sets for specialized agents --- + +PLANNER_TOOLS = {"recall", "establish", "mark", "notify_dm"} +PLANNER_TOOL_DEFINITIONS = [t for t in TOOL_DEFINITIONS if t["name"] in PLANNER_TOOLS] + + +def planner_execute_tool(tool_name: str, tool_input: dict, ctx: ToolContext) -> str: + """Execute a planner-allowed tool.""" + if tool_name not in PLANNER_TOOLS: + return f"Tool not available to planner: {tool_name}" + return execute_tool(tool_name, tool_input, ctx) + + +SEEDER_TOOLS = {"establish", "set_scene"} +SEEDER_TOOL_DEFINITIONS = [t for t in TOOL_DEFINITIONS if t["name"] in SEEDER_TOOLS] + + +def seeder_execute_tool(tool_name: str, tool_input: dict, ctx: ToolContext) -> str: + """Execute a seeder-allowed tool.""" + if tool_name not in SEEDER_TOOLS: + return f"Tool not available to seeder: {tool_name}" + return execute_tool(tool_name, tool_input, ctx) + + +ADVANCEMENT_TOOLS = {"recall", "update_character", "notify_dm"} +ADVANCEMENT_TOOL_DEFINITIONS = [ + t for t in TOOL_DEFINITIONS if t["name"] in ADVANCEMENT_TOOLS +] + + +def advancement_execute_tool(tool_name: str, tool_input: dict, ctx: ToolContext) -> str: + """Execute an advancement-evaluator-allowed tool.""" + if tool_name not in ADVANCEMENT_TOOLS: + return f"Tool not available to advancement evaluator: {tool_name}" + return execute_tool(tool_name, tool_input, ctx) diff --git a/src/storied/tools/_context.py b/src/storied/tools/_context.py new file mode 100644 --- /dev/null +++ b/src/storied/tools/_context.py @@ -0,0 +1,82 @@ +"""Shared infrastructure for all tool calls.""" + +import threading +from dataclasses import dataclass, field +from pathlib import Path + +from storied.character import update_character as char_update +from storied.initiative import InitiativeTracker +from storied.log import CampaignLog +from storied.search import VectorIndex + + +# Per-file locks for thread-safe entity writes (establish, mark) +_file_locks: dict[Path, threading.Lock] = {} +_file_locks_lock = threading.Lock() + + +def _get_file_lock(path: Path) -> threading.Lock: + """Get or create a lock for the given file path.""" + with _file_locks_lock: + if path not in _file_locks: + _file_locks[path] = threading.Lock() + return _file_locks[path] + + +class EntityIndex: + """Name->path index with write-through entity cache. + + Built once at startup by globbing the world directory. Lookups are + dict lookups instead of 6 sequential Path.exists() calls. The cache + stores parsed entity dicts so repeated loads within a turn skip disk I/O. + establish() and mark() update both the index and cache on write. + """ + + def __init__(self, world_dir: Path | None = None): + self._paths: dict[str, Path] = {} + self._cache: dict[Path, dict] = {} + if world_dir and world_dir.exists(): + for md in world_dir.rglob("*.md"): + self._paths[md.stem] = md + + def resolve(self, name: str) -> Path | None: + """Look up an entity's file path by name.""" + return self._paths.get(name) + + def register(self, name: str, path: Path) -> None: + """Register or update an entity's path in the index.""" + self._paths[name] = path + + def cache_get(self, path: Path) -> dict | None: + """Get a cached parsed entity, or None if not cached.""" + return self._cache.get(path) + + def cache_put(self, path: Path, data: dict) -> None: + """Store a parsed entity in the cache.""" + self._cache[path] = data + + +@dataclass +class ToolContext: + """Shared infrastructure for all tool calls. + + Created once per MCP server and passed to every tool invocation. + All fields are required — no defensive None checks in tool code. + """ + + world_id: str + player_id: str + base_path: Path + campaign_log: CampaignLog + entity_index: EntityIndex + vector_index: VectorIndex + initiative: InitiativeTracker = field(default_factory=InitiativeTracker) + + +def _sync_player_hp(target: str, ctx: ToolContext, result: str) -> str: + """Auto-sync player character sheet when damage/heal targets a player.""" + combatant = ctx.initiative._find(target) + if combatant and combatant.is_player: + char_update(ctx.player_id, {"hp.current": combatant.hp}, ctx.base_path) + result += f" (character sheet synced to {combatant.hp} HP)" + return result diff --git a/src/storied/tools/character.py b/src/storied/tools/character.py new file mode 100644 --- /dev/null +++ b/src/storied/tools/character.py @@ -0,0 +1,158 @@ +"""Character management tools.""" + +from storied.character import create_character as char_create +from storied.character import update_character as char_update +from storied.tools._context import ToolContext + + +def update_character(updates: dict, ctx: ToolContext) -> str: + """Update the player's character sheet to persist changes. + + Call this after HP changes, equipment gained/lost, coins spent, level ups, etc. + This ensures progress is saved and survives between sessions. + + Args: + updates: Fields to update. Use dot notation for nested fields. + Examples: + - {"hp.current": 5} - set current HP to 5 + - {"purse.gp": 25} - set gold to 25 + - {"purse.sp": 10, "purse.cp": 50} - set silver and copper + - {"level": 2, "hp.max": 20} - level up + For markdown sections, use "section.Name": + - {"section.Equipment": "- Longsword\\n- New shield"} + player_id: Player identifier (usually "default") + base_path: Base path for players directory + + Returns: + Confirmation of what was updated + """ + return char_update(ctx.player_id, updates, ctx.base_path) + + +def create_character( + name: str, + race: str, + char_class: str, + level: int, + abilities: dict[str, int], + hp_max: int, + ac: int, + ctx: ToolContext, + background: str | None = None, + speed: int = 30, + purse: dict[str, int] | None = None, + equipment: list[str] | None = None, + features: list[str] | None = None, + proficiencies: str | None = None, + backstory: str | None = None, +) -> str: + """Create a new player character and save to disk. + + Call this when character creation is complete. Include all the mechanical + details needed to play: abilities, HP, AC, equipment, and features. + + Args: + name: Character name + race: Race (e.g., "Human", "High Elf", "Hill Dwarf") + char_class: Class (e.g., "Fighter", "Wizard", "Rogue") + level: Starting level (usually 1) + abilities: All six ability scores as a dict: + {"strength": 15, "dexterity": 14, "constitution": 13, + "intelligence": 12, "wisdom": 10, "charisma": 8} + hp_max: Maximum hit points + ac: Armor class + background: Background (e.g., "Soldier", "Sage", "Criminal") + speed: Movement speed in feet (default 30) + purse: Starting coins by denomination: + {"cp": 0, "sp": 0, "ep": 0, "gp": 15, "pp": 0} + Denominations: cp (copper), sp (silver), ep (electrum), + gp (gold), pp (platinum). Omit denominations for 0. + equipment: List of equipment items + features: List of racial and class features + proficiencies: Description of proficiencies (armor, weapons, tools, saves, skills) + backstory: Character backstory and personality + + Returns: + Confirmation message + """ + return char_create( + player_id=ctx.player_id, + name=name, + race=race, + char_class=char_class, + level=level, + abilities=abilities, + hp_max=hp_max, + ac=ac, + background=background, + speed=speed, + purse=purse, + equipment=equipment, + features=features, + proficiencies=proficiencies, + backstory=backstory, + base_path=ctx.base_path, + ) + + +DEFINITIONS: list[dict] = [ + { + "name": "update_character", + "description": update_character.__doc__, + "input_schema": { + "type": "object", + "properties": { + "updates": { + "type": "object", + "description": "Fields to update. Use dot notation for nested (e.g., 'hp.current': 5). Use 'section.Name' for markdown sections.", + }, + }, + "required": ["updates"], + }, + }, + { + "name": "create_character", + "description": create_character.__doc__, + "input_schema": { + "type": "object", + "properties": { + "name": {"type": "string", "description": "Character name"}, + "race": {"type": "string", "description": "Race (e.g., 'Human', 'High Elf')"}, + "char_class": {"type": "string", "description": "Class (e.g., 'Fighter', 'Wizard')"}, + "level": {"type": "integer", "description": "Starting level (usually 1)"}, + "abilities": { + "type": "object", + "description": "All six ability scores: strength, dexterity, constitution, intelligence, wisdom, charisma", + }, + "hp_max": {"type": "integer", "description": "Maximum hit points"}, + "ac": {"type": "integer", "description": "Armor class"}, + "background": {"type": "string", "description": "Background (e.g., 'Soldier', 'Sage')"}, + "speed": {"type": "integer", "description": "Movement speed in feet"}, + "purse": { + "type": "object", + "description": "Starting coins: {cp, sp, ep, gp, pp}. Omit denominations for 0.", + "properties": { + "cp": {"type": "integer", "description": "Copper pieces"}, + "sp": {"type": "integer", "description": "Silver pieces"}, + "ep": {"type": "integer", "description": "Electrum pieces"}, + "gp": {"type": "integer", "description": "Gold pieces"}, + "pp": {"type": "integer", "description": "Platinum pieces"}, + }, + }, + "equipment": { + "type": "array", + "items": {"type": "string"}, + "description": "List of equipment items", + }, + "features": { + "type": "array", + "items": {"type": "string"}, + "description": "List of racial and class features", + }, + "proficiencies": {"type": "string", "description": "Proficiency description"}, + "backstory": {"type": "string", "description": "Character backstory and personality"}, + }, + "required": ["name", "race", "char_class", "level", "abilities", "hp_max", "ac"], + }, + }, +] diff --git a/src/storied/tools/entities.py b/src/storied/tools/entities.py new file mode 100644 --- /dev/null +++ b/src/storied/tools/entities.py @@ -0,0 +1,466 @@ +"""World entity tools — establish, mark, note_discovery.""" + +import re +from pathlib import Path + +import yaml + +from storied.session import name_to_slug +from storied.tools._context import EntityIndex, ToolContext, _get_file_lock + + +def establish( + entity_type: str, + name: str, + ctx: ToolContext, + description: str | None = None, + location: str | None = None, + knows: list[str] | None = None, + wants: list[str] | None = None, + will: list[str] | None = None, +) -> str: + """Establish or update an entity in the world. + + Use to create NPCs, locations, items, factions, or threads with their inner + state. Everything has Knows/Wants/Will: + - **Knows** = secrets, hidden truths, what isn't obvious + - **Wants** = nature, tendencies, inclinations (even non-sentient things can "want") + - **Will** = conditional triggers, what happens if... + + This isn't literal consciousness - it's narrative tendency. A bridge can "want" + to collapse. Cursed gold "wants" to be spent. Frame it this way and the world + feels alive. + + Partial updates: omit fields to preserve existing content when updating. + + Args: + entity_type: Type of entity: npcs, locations, items, factions, threads, lore + name: Display name (e.g., "Vera Blackwater", "The Rusty Anchor") + This becomes the filename directly (no slugification). + description: Prose description for the ## Is section. Include appearance, + background, current state, relationships via [[wikilinks]]. + location: Where this entity is right now. Can be a simple wikilink like + "[[The Rusty Anchor]]" or a verbal description like "In the basement + of [[The Rusty Anchor]]" or "Wandering the docks of [[Greyhaven]]". + knows: List of secrets and hidden truths. Things that aren't obvious. + wants: List of desires, tendencies, inclinations. The entity's nature. + will: List of conditional behaviors: "If X -> Y" format. + + Returns: + Confirmation with the file path + """ + world_dir = ctx.base_path / "worlds" / ctx.world_id / entity_type + world_dir.mkdir(parents=True, exist_ok=True) + file_path = world_dir / f"{name}.md" + + lock = _get_file_lock(file_path) + with lock: + existing = _load_entity(file_path, ctx.entity_index) + + if description is None: + description = existing.get("description", "") + if location is None: + location = existing.get("location", "") + if knows is None: + knows = existing.get("knows", []) + if wants is None: + wants = existing.get("wants", []) + if will is None: + will = existing.get("will", []) + was = existing.get("was", []) + + data = { + "description": description, "location": location, + "knows": knows, "wants": wants, "will": will, "was": was, + } + _write_entity(file_path, name, entity_type, data, ctx) + + action = "Updated" if existing else "Established" + return f"{action} {entity_type.rstrip('s')} '{name}'" + + +def _load_entity(file_path: Path, entity_index: EntityIndex) -> dict: + """Load an existing entity file and parse its structure.""" + cached = entity_index.cache_get(file_path) + if cached is not None: + return cached + + if not file_path.exists(): + return {} + + content = file_path.read_text() + result = {} + + # Parse ## Is section + is_match = re.search(r"## Is\n\n?(.*?)(?=\n## |\Z)", content, re.DOTALL) + if is_match: + is_content = is_match.group(1).strip() + + loc_match = re.search(r"\*\*Location:\*\*\s*(.+)", is_content) + if loc_match: + result["location"] = loc_match.group(1).strip() + + desc_match = re.match(r"(.*?)(?=\n### |\Z)", is_content, re.DOTALL) + if desc_match: + desc = desc_match.group(1).strip() + desc = re.sub(r"\*\*Location:\*\*\s*.+\n?", "", desc).strip() + result["description"] = desc + + knows_match = re.search(r"### Knows\n\n?(.*?)(?=\n### |\n## |\Z)", is_content, re.DOTALL) + if knows_match: + result["knows"] = _parse_list_items(knows_match.group(1)) + + wants_match = re.search(r"### Wants\n\n?(.*?)(?=\n### |\n## |\Z)", is_content, re.DOTALL) + if wants_match: + result["wants"] = _parse_list_items(wants_match.group(1)) + + will_match = re.search(r"### Will\n\n?(.*?)(?=\n### |\n## |\Z)", is_content, re.DOTALL) + if will_match: + result["will"] = _parse_list_items(will_match.group(1)) + + # Parse ## Was section + was_match = re.search(r"## Was\n\n?(.*?)(?=\n## |\Z)", content, re.DOTALL) + if was_match: + result["was"] = _parse_list_items(was_match.group(1)) + + entity_index.cache_put(file_path, result) + return result + + +def _parse_list_items(text: str) -> list[str]: + """Parse markdown list items into a list of strings.""" + items = [] + for line in text.strip().split("\n"): + line = line.strip() + if line.startswith("- "): + items.append(line[2:]) + return items + + +def _format_entity( + name: str, + description: str, + location: str, + knows: list[str], + wants: list[str], + will: list[str], + was: list[str], +) -> str: + """Format an entity as markdown with Is/Was structure.""" + lines = [f"# {name}", "", "## Is", ""] + + if location: + lines.append(f"**Location:** {location}") + lines.append("") + + if description: + lines.append(description) + lines.append("") + + if knows: + lines.append("### Knows") + lines.append("") + for item in knows: + lines.append(f"- {item}") + lines.append("") + + if wants: + lines.append("### Wants") + lines.append("") + for item in wants: + lines.append(f"- {item}") + lines.append("") + + if will: + lines.append("### Will") + lines.append("") + for item in will: + lines.append(f"- {item}") + lines.append("") + + lines.append("## Was") + lines.append("") + for item in was: + lines.append(f"- {item}") + lines.append("") + + return "\n".join(lines) + + +def _write_entity( + file_path: Path, + name: str, + entity_type: str, + data: dict, + ctx: ToolContext, +) -> None: + """Write an entity to disk and update all indexes.""" + file_content = _format_entity( + name, data["description"], data["location"], + data["knows"], data["wants"], data["will"], data["was"], + ) + file_path.write_text(file_content) + ctx.entity_index.register(name, file_path) + ctx.entity_index.cache_put(file_path, data) + ctx.vector_index.upsert( + f"world:{entity_type}/{name}.md:0", + file_content, + {"source": "world", "content_type": entity_type, + "path": str(file_path), "title": name}, + ) + + +def _auto_mark_present( + present: list[str], event: str, ctx: ToolContext, +) -> list[str]: + """Auto-mark present entities with the current event.""" + marked: list[str] = [] + for ref in present: + link_match = re.search(r"\[\[([^\]]+)\]\]", ref) + if not link_match: + continue + name = link_match.group(1) + + file_path = ctx.entity_index.resolve(name) + if file_path is None: + for etype in ("npcs", "locations", "items", "factions"): + candidate = ctx.base_path / "worlds" / ctx.world_id / etype / f"{name}.md" + if candidate.exists(): + file_path = candidate + break + + if file_path and file_path.exists(): + entity_type = file_path.parent.name + mark(entity_type=entity_type, name=name, event=event, ctx=ctx) + marked.append(name) + + return marked + + +def mark( + entity_type: str, + name: str, + event: str, + ctx: ToolContext, + resolves: list[str] | None = None, +) -> str: + """Record an event in an entity's history (## Was section). + + Use when something significant happens to or involving an entity. This builds + their history and helps maintain continuity across sessions. + + If the event resolves a Will trigger (e.g., "Vera introduced the player to + Harrik" resolves "If trusted -> intro to Harrik"), provide the trigger text + in `resolves` to remove it from the Will section. + + Args: + entity_type: Type of entity: npcs, locations, items, factions, threads + name: Entity name (exact filename match) + event: What happened - brief description for the Was section + resolves: Optional list of Will items to remove if this event fired triggers + + Returns: + Confirmation message + """ + file_path = ctx.entity_index.resolve(name) + if file_path is None: + file_path = ctx.base_path / "worlds" / ctx.world_id / entity_type / f"{name}.md" + + if not file_path.exists(): + return f"Error: Entity '{name}' not found in {entity_type}" + + timestamp = ctx.campaign_log.get_current_time().to_anchor() + + lock = _get_file_lock(file_path) + with lock: + existing = _load_entity(file_path, ctx.entity_index) + + was = existing.get("was", []) + was.append(f"{timestamp} | {event}") + + will = existing.get("will", []) + resolved = [] + for trigger in resolves or []: + if trigger in will: + will.remove(trigger) + resolved.append(trigger) + + data = { + "description": existing.get("description", ""), + "location": existing.get("location", ""), + "knows": existing.get("knows", []), + "wants": existing.get("wants", []), + "will": will, + "was": was, + } + _write_entity(file_path, name, entity_type, data, ctx) + + result = f"Marked: {event}" + if resolved: + if len(resolved) == 1: + result += f" (resolved: {resolved[0]})" + else: + result += f" (resolved {len(resolved)} triggers)" + return result + + +def note_discovery( + entity: str, + content: str, + ctx: ToolContext, + content_type: str = "lore", + tags: list[str] | None = None, +) -> str: + """Record what the player has learned about something. + + Use when the player discovers or learns information about an NPC, + location, faction, or other world element. This captures their + perspective, which may be incomplete or even wrong. + + The player's knowledge is separate from DM truth (use `establish` + for the full facts). This helps track what the player knows vs. + what they haven't discovered yet. + + Args: + entity: Name of what they learned about (e.g., "Vera Blackwater") + content: What the player learned or observed + content_type: Type of content - npcs, locations, factions, lore (default: lore) + tags: Optional tags for categorization + + Returns: + Confirmation message + """ + slug = name_to_slug(entity) + + knowledge_dir = ( + ctx.base_path / "players" / ctx.player_id / "worlds" + / ctx.world_id / content_type + ) + knowledge_dir.mkdir(parents=True, exist_ok=True) + file_path = knowledge_dir / f"{slug}.md" + + frontmatter: dict = { + "type": content_type.rstrip("s"), + "name": entity, + } + if tags: + frontmatter["tags"] = tags + + file_content = "---\n" + file_content += yaml.dump(frontmatter, sort_keys=False, allow_unicode=True) + file_content += "---\n\n" + file_content += content.strip() + file_content += "\n" + + file_path.write_text(file_content) + + ctx.vector_index.upsert( + f"player:{content_type}/{slug}.md:0", + file_content, + {"source": "player", "content_type": content_type, + "path": str(file_path), "title": entity}, + ) + + return f"Noted: player learned about '{entity}'" + + +DEFINITIONS: list[dict] = [ + { + "name": "establish", + "description": establish.__doc__, + "input_schema": { + "type": "object", + "properties": { + "entity_type": { + "type": "string", + "description": "Type of entity", + "enum": ["npcs", "locations", "items", "factions", "threads", "lore"], + }, + "name": { + "type": "string", + "description": "Display name (exact filename, e.g., 'Vera Blackwater')", + }, + "description": { + "type": "string", + "description": "Prose description with [[wikilinks]] for relationships", + }, + "location": { + "type": "string", + "description": "Current location (e.g., '[[The Rusty Anchor]]' or 'In the basement of [[The Rusty Anchor]]')", + }, + "knows": { + "type": "array", + "items": {"type": "string"}, + "description": "Secrets, hidden truths - what isn't obvious", + }, + "wants": { + "type": "array", + "items": {"type": "string"}, + "description": "Nature, tendencies, inclinations - even non-sentient things", + }, + "will": { + "type": "array", + "items": {"type": "string"}, + "description": "Conditional behaviors in 'If X -> Y' format", + }, + }, + "required": ["entity_type", "name"], + }, + }, + { + "name": "mark", + "description": mark.__doc__, + "input_schema": { + "type": "object", + "properties": { + "entity_type": { + "type": "string", + "description": "Type of entity", + "enum": ["npcs", "locations", "items", "factions", "threads"], + }, + "name": { + "type": "string", + "description": "Entity name (exact filename match)", + }, + "event": { + "type": "string", + "description": "What happened - brief description", + }, + "resolves": { + "type": "array", + "items": {"type": "string"}, + "description": "Optional: Will items to remove if this event fired triggers", + }, + }, + "required": ["entity_type", "name", "event"], + }, + }, + { + "name": "note_discovery", + "description": note_discovery.__doc__, + "input_schema": { + "type": "object", + "properties": { + "entity": { + "type": "string", + "description": "Name of what they learned about (e.g., 'Vera Blackwater')", + }, + "content": { + "type": "string", + "description": "What the player learned or observed", + }, + "content_type": { + "type": "string", + "description": "Type of content", + "enum": ["npcs", "locations", "factions", "lore"], + }, + "tags": { + "type": "array", + "items": {"type": "string"}, + "description": "Tags for categorization", + }, + }, + "required": ["entity", "content"], + }, + }, +] diff --git a/src/storied/tools/mechanics.py b/src/storied/tools/mechanics.py new file mode 100644 --- /dev/null +++ b/src/storied/tools/mechanics.py @@ -0,0 +1,143 @@ +"""Dice, rules lookup, and code execution tools.""" + +from pathlib import Path + +from storied.dice import roll as dice_roll +from storied.tools._context import ToolContext + + +def roll(notation: str, reason: str | None = None) -> dict: + """Roll dice using standard notation like '1d20', '2d6+3', '4d6kh3'. + + Use for attack rolls, skill checks, saving throws, and damage rolls. + Supports: XdY, XdY+Z, XdY-Z, advantage (2d20kh1), disadvantage (2d20kl1). + + Args: + notation: Dice notation string (e.g., "1d20+5", "2d6", "4d6kh3") + reason: Brief description of what the roll is for (e.g., "Athletics", + "Attack with longsword", "Wisdom save", "Fireball damage") + + Returns: + Dict with rolls, kept dice, modifier, and total + """ + result = dice_roll(notation) + return result.to_dict() + + +def recall( + query: str, + ctx: ToolContext, + scope: str = "all", + content_type: str | None = None, +) -> str: + """Look up rules, world content, or both. + + Use to recall information about: + - Rules: spells, monsters, classes, items, conditions from the SRD + - World: NPCs, locations, factions, lore you've established + - Both: search everything (default) + + Args: + query: What to look up (e.g., "fireball", "captain vex", "merchant guild") + scope: Where to search - "rules", "world", or "all" (default) + content_type: Optional type to limit search (e.g., "spells", "npcs") + + Returns: + Content of the found item, or a message if not found + """ + source_filter: str | None = None + if scope == "rules": + source_filter = "srd" + + current_day = ctx.campaign_log.get_current_time().day + hits = ctx.vector_index.search( + query, limit=5, source_filter=source_filter, + exclude_source="srd" if scope == "world" else None, + decay_ref=current_day, + ) + if hits: + if len(hits) == 1 or hits[0].score > 0.8: + hit_path = Path(hits[0].path) + if hit_path.exists(): + return hit_path.read_text() + lines = [f"Found {len(hits)} matches:"] + for h in hits: + lines.append(f" - {h.doc_id.split(':')[1]} ({h.source}): {h.snippet[:80]}") + return "\n".join(lines) + + return f"Nothing found matching '{query}'" + + +DEFINITIONS: list[dict] = [ + { + "name": "roll", + "description": roll.__doc__, + "input_schema": { + "type": "object", + "properties": { + "notation": { + "type": "string", + "description": "Dice notation (e.g., '1d20+5', '2d6', '4d6kh3')", + }, + "reason": { + "type": "string", + "description": "What the roll is for (e.g., 'Athletics', 'Longsword attack', 'Dex save')", + }, + }, + "required": ["notation", "reason"], + }, + }, + { + "name": "recall", + "description": recall.__doc__, + "input_schema": { + "type": "object", + "properties": { + "query": { + "type": "string", + "description": "What to look up (e.g., 'fireball', 'captain vex')", + }, + "scope": { + "type": "string", + "description": "Where to search: 'rules' (SRD), 'world' (established content), or 'all' (both)", + "enum": ["rules", "world", "all"], + }, + "content_type": { + "type": "string", + "description": "Type to limit search (e.g., 'spells', 'npcs', 'monsters')", + }, + }, + "required": ["query"], + }, + }, + { + "name": "run_code", + "description": ( + "Run Python code in a secure sandbox. Use for calculations, random " + "generation, data formatting, or any computation the narrative needs.\n\n" + "All your DM tools are callable as functions (see signatures below). " + "Most return a str with the result. The exception is roll(), which " + "returns a dict with keys: notation, rolls, kept, modifier, total — " + "use roll('2d6+3')['total'] for math, or index into rolls/kept for " + "individual dice. Use roll() for all randomness (no random module).\n\n" + "Language: variables, functions, loops, conditionals, comprehensions, " + "f-strings. Stdlib: re, json, datetime, math. No classes, no other " + "imports, no file/network access. Errors return as text.\n\n" + "Available functions:\n{tool_signatures}" + ), + "input_schema": { + "type": "object", + "properties": { + "description": { + "type": "string", + "description": "What this code does in game terms (e.g., 'Designing cave system', 'Splitting treasure', 'Generating NPC schedule')", + }, + "code": { + "type": "string", + "description": "Python code to execute", + }, + }, + "required": ["description", "code"], + }, + }, +] diff --git a/src/storied/tools/scene.py b/src/storied/tools/scene.py new file mode 100644 --- /dev/null +++ b/src/storied/tools/scene.py @@ -0,0 +1,215 @@ +"""Scene management, session, style tuning, and DM notification tools.""" + +import re + +from storied import notifications +from storied.session import update_session as session_update +from storied.tools._context import ToolContext +from storied.tools.entities import _auto_mark_present + + +def set_scene( + ctx: ToolContext, + event: str | None = None, + duration: str | None = None, + situation: str | None = None, + location: str | None = None, + present: list[str] | None = None, + threads: list[str] | None = None, + tags: list[str] | None = None, +) -> str: + """Call this after every response. Logs what happened, advances the clock, + and updates the scene state. + + Always include event and duration — time only advances when you log it. + + Args: + event: What happened this turn (e.g., "Spoke with the innkeeper", + "Searched the warehouse", "Fought off thugs"). + duration: How long it took (e.g., "10 min", "1 hour", "3 rounds", + "1 day"). The clock advances by this amount. + situation: Updated situation summary. Write in present tense as if + briefing another DM taking over mid-session. + location: New location when the player moves (e.g., "The Rusty Anchor") + present: Entities currently present, using [[Name]] format + (e.g., ["[[Vera Blackwater]]", "[[Henrik]] - barkeep"]) + threads: Open plot threads or objectives + tags: Optional tags: "combat", "rest:short", "rest:long", "travel" + + Returns: + Confirmation of what was updated + """ + parts = [] + + if event and duration: + anchor = ctx.campaign_log.append_entry(event, duration, tags=tags) + current = ctx.campaign_log.get_current_time() + parts.append( + f"Logged: {anchor} | {event} | {duration} → " + f"Now: {current} ({current.period_of_day()}, {current.atmosphere()})" + ) + + updates = {} + if situation is not None: + updates["situation"] = situation + if location is not None: + updates["location"] = location + if present is not None: + updates["present"] = present + if threads is not None: + updates["threads"] = threads + + if updates: + result = session_update(ctx.player_id, updates, ctx.base_path) + parts.append(result) + + if event and present: + marked = _auto_mark_present(present, event, ctx) + if marked: + parts.append(f"Auto-marked: {', '.join(marked)}") + + return "; ".join(parts) if parts else "No updates" + + +def tune(tuning: str, ctx: ToolContext) -> str: + """Update your storytelling style based on player feedback. + + Write the complete updated style as markdown prose. This replaces the + entire current style. Incorporate existing preferences where they still + apply — don't discard preferences the player hasn't contradicted. + """ + path = ctx.base_path / "worlds" / ctx.world_id / "style.md" + path.write_text(f"# Style\n\n{tuning}\n") + return "Style updated." + + +def end_session(situation: str, ctx: ToolContext, threads: list[str] | None = None) -> str: + """End the current session, saving the game state for next time. + + Call this when the player indicates they want to stop playing. This saves + the current situation so the next session can resume smoothly. + + Before calling, give a brief farewell and summary of what happened this session. + + Args: + situation: Summary of the current state of affairs for the next session. + Write as if briefing a DM who will pick up where you left off. + threads: Open plot threads or objectives to carry forward + + Returns: + Confirmation that session was saved + """ + updates: dict = {"situation": situation} + if threads is not None: + updates["threads"] = threads + + session_update(ctx.player_id, updates, ctx.base_path) + return "SESSION_ENDED" + + +def notify_dm(message: str, ctx: ToolContext) -> str: + """Send a notification that the DM will see at the start of the next turn. + + Use this to communicate important background changes to the DM, + such as advancement readiness or world state changes. + + Args: + message: The notification message for the DM + + Returns: + Confirmation that the notification was queued + """ + notifications.append(ctx.world_id, ctx.base_path, message) + return f"Notification queued: {message}" + + +DEFINITIONS: list[dict] = [ + { + "name": "set_scene", + "description": set_scene.__doc__, + "input_schema": { + "type": "object", + "properties": { + "event": { + "type": "string", + "description": "What happened this turn (logged to campaign journal)", + }, + "duration": { + "type": "string", + "description": "How long it took (e.g., '10 min', '1 hour', '3 rounds')", + }, + "situation": { + "type": "string", + "description": "Updated situation summary in present tense", + }, + "location": { + "type": "string", + "description": "New location when the player moves", + }, + "present": { + "type": "array", + "items": {"type": "string"}, + "description": "Entities present, using [[Name]] format", + }, + "threads": { + "type": "array", + "items": {"type": "string"}, + "description": "Open plot threads or objectives", + }, + "tags": { + "type": "array", + "items": {"type": "string"}, + "description": "Optional: 'combat', 'rest:short', 'rest:long', 'travel', 'level'", + }, + }, + "required": ["event", "duration"], + }, + }, + { + "name": "tune", + "description": tune.__doc__, + "input_schema": { + "type": "object", + "properties": { + "tuning": { + "type": "string", + "description": "Complete updated style as markdown prose. Replaces the current style entirely.", + }, + }, + "required": ["tuning"], + }, + }, + { + "name": "end_session", + "description": end_session.__doc__, + "input_schema": { + "type": "object", + "properties": { + "situation": { + "type": "string", + "description": "Summary of current state for the next session", + }, + "threads": { + "type": "array", + "items": {"type": "string"}, + "description": "Open plot threads or objectives to carry forward", + }, + }, + "required": ["situation"], + }, + }, + { + "name": "notify_dm", + "description": notify_dm.__doc__, + "input_schema": { + "type": "object", + "properties": { + "message": { + "type": "string", + "description": "The notification message for the DM", + }, + }, + "required": ["message"], + }, + }, +] diff --git a/tests/test_advancement.py b/tests/test_advancement.py new file mode 100644 --- /dev/null +++ b/tests/test_advancement.py @@ -0,0 +1,359 @@ +"""Tests for the character advancement evaluation system.""" + +import json +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from storied.character import load_character, save_character +from storied.log import CampaignLog +from storied.advancement import ( + AdvancementResult, + BackgroundAdvancement, + build_advancement_context, + evaluate_advancement, +) +from storied.session import save_session +from storied.tools import ToolContext +from storied.tools.scene import notify_dm + + +# --- Fixtures --- + + +@pytest.fixture +def character(ctx: ToolContext) -> dict: + """Create a level 3 rogue character.""" + data = { + "name": "Kira", + "race": "Human", + "class": "Rogue", + "level": 3, + "hp": {"current": 24, "max": 24}, + "ac": 14, + "speed": 30, + "abilities": { + "strength": 10, + "dexterity": 16, + "constitution": 14, + "intelligence": 12, + "wisdom": 10, + "charisma": 14, + }, + "body": "## Features\n- Sneak Attack 2d6\n- Cunning Action\n- Thief subclass", + } + save_character("default", data, ctx.base_path) + return data + + +@pytest.fixture +def campaign_with_events(ctx: ToolContext) -> CampaignLog: + """Campaign log with some adventure events.""" + log = ctx.campaign_log + log.append_entry("Arrived in Millford", "30 min") + log.append_entry("Investigated the warehouse", "1 hour") + log.append_entry("Fought three thugs", "3 rounds", tags=["combat"]) + log.append_entry("Discovered smuggling ring", "30 min") + log.append_entry("Traveled to Thornwall", "4 hours") + log.append_entry("Infiltrated the manor", "2 hours") + log.append_entry("Stole the merchant's ledger", "30 min") + return log + + +# --- notify_dm tool --- + + +class TestNotifyDM: + def test_appends_notification(self, ctx: ToolContext): + result = notify_dm("Test message", ctx) + + assert "queued" in result.lower() + path = ctx.base_path / "worlds" / ctx.world_id / "dm_notifications.md" + assert path.exists() + assert "Test message" in path.read_text() + + def test_multiple_notifications(self, ctx: ToolContext): + notify_dm("First", ctx) + notify_dm("Second", ctx) + + path = ctx.base_path / "worlds" / ctx.world_id / "dm_notifications.md" + content = path.read_text() + assert "First" in content + assert "Second" in content + + +# --- build_advancement_context --- + + +class TestBuildAdvancementContext: + def test_returns_none_without_character(self, ctx: ToolContext): + result = build_advancement_context( + ctx.world_id, ctx.player_id, ctx.base_path + ) + assert result is None + + def test_returns_none_when_advancement_ready( + self, ctx: ToolContext, character: dict + ): + character["advancement_ready"] = 4 + save_character("default", character, ctx.base_path) + + result = build_advancement_context( + ctx.world_id, ctx.player_id, ctx.base_path + ) + assert result is None + + def test_includes_character_info( + self, ctx: ToolContext, character: dict + ): + context = build_advancement_context( + ctx.world_id, ctx.player_id, ctx.base_path + ) + assert context is not None + assert "Kira" in context + assert "Rogue" in context + assert "Level 3" in context + + def test_includes_log_entries( + self, + ctx: ToolContext, + character: dict, + campaign_with_events: CampaignLog, + ): + context = build_advancement_context( + ctx.world_id, ctx.player_id, ctx.base_path + ) + assert context is not None + assert "warehouse" in context + assert "smuggling ring" in context + + def test_includes_entries_since_level_tag( + self, ctx: ToolContext, character: dict + ): + log = ctx.campaign_log + log.append_entry("Old event before level-up", "1 hour") + log.append_entry("Leveled up to 3", "5 min", tags=["level"]) + log.append_entry("New adventure begins", "30 min") + log.append_entry("Fought a dragon", "5 rounds", tags=["combat"]) + + context = build_advancement_context( + ctx.world_id, ctx.player_id, ctx.base_path + ) + assert context is not None + assert "Old event before level-up" not in context + assert "New adventure begins" in context + assert "Fought a dragon" in context + + def test_includes_advancement_history( + self, ctx: ToolContext, character: dict + ): + log = ctx.campaign_log + log.append_entry("Reached level 2", "5 min", tags=["level"]) + log.append_entry("Adventured more", "2 hours") + log.append_entry("Reached level 3", "5 min", tags=["level"]) + log.append_entry("Recent events", "1 hour") + + context = build_advancement_context( + ctx.world_id, ctx.player_id, ctx.base_path + ) + assert context is not None + assert "Advancement History" in context + assert "Reached level 2" in context + assert "Reached level 3" in context + + def test_includes_session_state( + self, ctx: ToolContext, character: dict + ): + save_session( + "default", + { + "location": "Town Square", + "body": "## Open Threads\n- Find the missing merchant", + }, + ctx.base_path, + ) + + context = build_advancement_context( + ctx.world_id, ctx.player_id, ctx.base_path + ) + assert context is not None + assert "missing merchant" in context + + +# --- CampaignLog tag methods --- + + +class TestLogTagMethods: + def test_get_entries_since_tag_returns_all_when_no_tag( + self, ctx: ToolContext + ): + log = ctx.campaign_log + log.append_entry("Event one", "10 min") + log.append_entry("Event two", "10 min") + + entries = log.get_entries_since_tag("level") + assert len(entries) == 2 + + def test_get_entries_since_tag_returns_after_last_tag( + self, ctx: ToolContext + ): + log = ctx.campaign_log + log.append_entry("Before level", "10 min") + log.append_entry("Level up!", "5 min", tags=["level"]) + log.append_entry("After level", "10 min") + + entries = log.get_entries_since_tag("level") + assert len(entries) == 1 + assert entries[0].event == "After level" + + def test_get_entries_since_tag_uses_last_occurrence( + self, ctx: ToolContext + ): + log = ctx.campaign_log + log.append_entry("First level", "5 min", tags=["level"]) + log.append_entry("Between levels", "1 hour") + log.append_entry("Second level", "5 min", tags=["level"]) + log.append_entry("After second", "30 min") + + entries = log.get_entries_since_tag("level") + assert len(entries) == 1 + assert entries[0].event == "After second" + + def test_find_tag_entries(self, ctx: ToolContext): + log = ctx.campaign_log + log.append_entry("Normal event", "10 min") + log.append_entry("Level 2", "5 min", tags=["level"]) + log.append_entry("More stuff", "1 hour") + log.append_entry("Level 3", "5 min", tags=["level"]) + + entries = log.find_tag_entries("level") + assert len(entries) == 2 + assert entries[0].event == "Level 2" + assert entries[1].event == "Level 3" + + def test_find_tag_entries_empty(self, ctx: ToolContext): + log = ctx.campaign_log + log.append_entry("Normal event", "10 min") + + entries = log.find_tag_entries("level") + assert entries == [] + + def test_get_all_entries(self, ctx: ToolContext): + log = ctx.campaign_log + log.append_entry("Day 1 morning", "10 min") + log.append_entry("Day 1 travel", "18 hours") + log.append_entry("Day 2 event", "10 min") + + entries = log.get_all_entries() + assert len(entries) == 3 + assert entries[0].event == "Day 1 morning" + assert entries[2].event == "Day 2 event" + + +# --- evaluate_advancement --- + + +class TestEvaluateAdvancement: + def test_skips_without_character(self, ctx: ToolContext): + result = evaluate_advancement( + world_id=ctx.world_id, + player_id=ctx.player_id, + base_path=ctx.base_path, + ) + assert result.evaluated is False + + def test_skips_when_advancement_ready( + self, ctx: ToolContext, character: dict + ): + character["advancement_ready"] = 4 + save_character("default", character, ctx.base_path) + + result = evaluate_advancement( + world_id=ctx.world_id, + player_id=ctx.player_id, + base_path=ctx.base_path, + ) + assert result.evaluated is False + + @patch("storied.claude.subprocess.Popen") + def test_calls_claude_when_character_exists( + self, + mock_popen: MagicMock, + ctx: ToolContext, + character: dict, + campaign_with_events: CampaignLog, + ): + result_line = json.dumps({ + "type": "result", + "session_id": "sess-adv", + "usage": {"input_tokens": 500, "output_tokens": 100}, + "duration_ms": 2000, + }) + mock_proc = MagicMock() + mock_proc.stdin = MagicMock() + mock_proc.stdout = iter([result_line.encode() + b"\n"]) + mock_proc.stderr = iter([]) + mock_proc.wait.return_value = 0 + mock_proc.returncode = 0 + mock_popen.return_value = mock_proc + + result = evaluate_advancement( + world_id=ctx.world_id, + player_id=ctx.player_id, + base_path=ctx.base_path, + ) + + assert result.evaluated is True + assert result.input_tokens == 500 + mock_popen.assert_called_once() + + +# --- BackgroundAdvancement --- + + +class TestBackgroundAdvancement: + def test_does_not_trigger_before_interval(self): + adv = BackgroundAdvancement( + world_id="test", + player_id="default", + base_path=Path("/tmp/fake"), + interval=5, + ) + # 4 turns should not trigger + for _ in range(4): + adv.on_turn() + assert adv._thread is None + + def test_on_combat_end_triggers_immediately( + self, ctx: ToolContext, character: dict + ): + adv = BackgroundAdvancement( + world_id=ctx.world_id, + player_id=ctx.player_id, + base_path=ctx.base_path, + interval=100, + ) + + with patch("storied.advancement.evaluate_advancement") as mock_eval: + mock_eval.return_value = AdvancementResult() + adv.on_combat_end() + # Give the thread a moment + if adv._thread: + adv._thread.join(timeout=2) + mock_eval.assert_called_once() + + def test_turn_counter_resets_after_evaluation(self): + adv = BackgroundAdvancement( + world_id="test", + player_id="default", + base_path=Path("/tmp/fake"), + interval=5, + ) + with patch("storied.advancement.evaluate_advancement") as mock_eval: + mock_eval.return_value = AdvancementResult() + for _ in range(5): + adv.on_turn() + if adv._thread: + adv._thread.join(timeout=2) + assert adv._turn_count == 0 diff --git a/tests/test_notifications.py b/tests/test_notifications.py new file mode 100644 --- /dev/null +++ b/tests/test_notifications.py @@ -0,0 +1,58 @@ +"""Tests for the DM notification channel.""" + +from pathlib import Path + +import pytest + +from storied import notifications + + +@pytest.fixture +def world(tmp_path: Path) -> tuple[str, Path]: + """Return (world_id, base_path) with world directory created.""" + world_dir = tmp_path / "worlds" / "test-world" + world_dir.mkdir(parents=True) + return "test-world", tmp_path + + +class TestAppendAndDrain: + def test_drain_empty(self, world: tuple[str, Path]): + world_id, base_path = world + assert notifications.drain(world_id, base_path) == [] + + def test_append_then_drain(self, world: tuple[str, Path]): + world_id, base_path = world + notifications.append(world_id, base_path, "Something happened") + + messages = notifications.drain(world_id, base_path) + assert messages == ["Something happened"] + + def test_drain_clears(self, world: tuple[str, Path]): + world_id, base_path = world + notifications.append(world_id, base_path, "First") + + notifications.drain(world_id, base_path) + assert notifications.drain(world_id, base_path) == [] + + def test_multiple_messages(self, world: tuple[str, Path]): + world_id, base_path = world + notifications.append(world_id, base_path, "First thing") + notifications.append(world_id, base_path, "Second thing") + notifications.append(world_id, base_path, "Third thing") + + messages = notifications.drain(world_id, base_path) + assert messages == ["First thing", "Second thing", "Third thing"] + + def test_file_removed_after_drain(self, world: tuple[str, Path]): + world_id, base_path = world + notifications.append(world_id, base_path, "Temporary") + + notifications.drain(world_id, base_path) + path = base_path / "worlds" / "test-world" / "dm_notifications.md" + assert not path.exists() + + def test_creates_world_dir_if_missing(self, tmp_path: Path): + notifications.append("new-world", tmp_path, "Hello") + + messages = notifications.drain("new-world", tmp_path) + assert messages == ["Hello"]