A 5e storytelling engine with an LLM DM
1# pyright: reportArgumentType=false
2# Tests pass session dicts with `object` values where helpers expect `str`;
3# the test harness shapes the dict correctly.
4"""Tests for the world planner — entity richness scoring, discovery, and context."""
5
6import json
7from pathlib import Path
8from unittest.mock import MagicMock, patch
9
10import pytest
11
12from storied.planner import (
13 BackgroundTicker,
14 PlanResult,
15 _find_entities_with_will,
16 build_planning_context,
17 build_tick_context,
18 entity_richness,
19 find_nearby_entities,
20 plan_world,
21)
22from storied.session import save_session
23from storied.testing import call_tool
24from storied.tools import ToolContext
25from storied.tools.entities import establish, mark
26
27
28@pytest.fixture
29def populated_world(ctx: ToolContext) -> ToolContext:
30 """Create a world with some entities for testing."""
31 call_tool(
32 establish,
33 entity_type="locations",
34 name="Town Square",
35 description=(
36 "The center of [[Millford]]. A fountain stands here, "
37 "surrounded by market stalls."
38 ),
39 location="Central [[Millford]]",
40 knows=["The fountain was built by [[Old Gregor]]"],
41 wants=["To be a gathering place"],
42 will=["If market day → attract crowds"],
43 )
44 call_tool(
45 establish,
46 entity_type="npcs",
47 name="Old Gregor",
48 description="An elderly stonemason.",
49 location="[[Town Square]]",
50 )
51 call_tool(
52 establish,
53 entity_type="locations",
54 name="Millford",
55 description="A small riverside town.",
56 )
57 call_tool(
58 establish,
59 entity_type="npcs",
60 name="Thin NPC",
61 description="Someone with no inner life.",
62 )
63 return ctx
64
65
66class TestEntityRichness:
67 """Tests for richness scoring."""
68
69 def test_empty_entity_scores_zero(self, ctx: ToolContext, tmp_path: Path):
70 call_tool(establish, entity_type="npcs", name="Empty")
71 path = tmp_path / "worlds/test-world/npcs/Empty.md"
72 assert entity_richness(path) == 0.0
73
74 def test_description_only(self, ctx: ToolContext, tmp_path: Path):
75 call_tool(
76 establish,
77 entity_type="npcs",
78 name="Described",
79 description="A tall warrior.",
80 )
81 path = tmp_path / "worlds/test-world/npcs/Described.md"
82 assert entity_richness(path) == pytest.approx(0.2)
83
84 def test_fully_rich_entity(self, ctx: ToolContext, tmp_path: Path):
85 call_tool(
86 establish,
87 entity_type="npcs",
88 name="Rich NPC",
89 description="A fully fleshed out character in [[Town Square]].",
90 knows=["Secret one", "Secret two"],
91 wants=["Goal one"],
92 will=["If X → do Y"],
93 )
94 call_tool(
95 mark,
96 entity_type="npcs",
97 name="Rich NPC",
98 event="Something happened",
99 )
100 path = tmp_path / "worlds/test-world/npcs/Rich NPC.md"
101 score = entity_richness(path)
102 assert score == pytest.approx(1.0)
103
104 def test_partial_richness(self, ctx: ToolContext, tmp_path: Path):
105 call_tool(
106 establish,
107 entity_type="npcs",
108 name="Partial",
109 description="Has description and knows.",
110 knows=["A secret"],
111 )
112 path = tmp_path / "worlds/test-world/npcs/Partial.md"
113 score = entity_richness(path)
114 # description (0.2) + knows (0.2) = 0.4
115 assert score == pytest.approx(0.4)
116
117 def test_wikilinks_contribute(self, ctx: ToolContext, tmp_path: Path):
118 call_tool(
119 establish,
120 entity_type="npcs",
121 name="Linked",
122 description="Hangs out at [[The Tavern]] with [[Bob]].",
123 )
124 path = tmp_path / "worlds/test-world/npcs/Linked.md"
125 score = entity_richness(path)
126 # description (0.2) + wikilinks (0.1) = 0.3
127 assert score == pytest.approx(0.3)
128
129
130class TestFindNearbyEntities:
131 """Tests for discovering entities near the player."""
132
133 def test_finds_entities_from_location_links(self, populated_world: ToolContext):
134 session = {
135 "location": "Town Square",
136 "body": "",
137 }
138 nearby = find_nearby_entities(session, "test-world")
139 names = {name for name, _ in nearby}
140 assert "Old Gregor" in names
141 assert "Millford" in names
142
143 def test_finds_entities_from_session_body(self, populated_world: ToolContext):
144 session = {
145 "location": "Town Square",
146 "body": "## Present\n- [[Thin NPC]]",
147 }
148 nearby = find_nearby_entities(session, "test-world")
149 names = {name for name, _ in nearby}
150 assert "Thin NPC" in names
151
152 def test_includes_current_location(self, populated_world: ToolContext):
153 session = {
154 "location": "Town Square",
155 "body": "",
156 }
157 nearby = find_nearby_entities(session, "test-world")
158 names = {name for name, _ in nearby}
159 assert "Town Square" in names
160
161 def test_no_duplicates(self, populated_world: ToolContext):
162 session = {
163 "location": "Town Square",
164 "body": "## Present\n- [[Old Gregor]]",
165 }
166 nearby = find_nearby_entities(session, "test-world")
167 names = [name for name, _ in nearby]
168 assert names.count("Old Gregor") == 1
169
170 def test_empty_session(self, ctx: ToolContext):
171 session = {"body": ""}
172 nearby = find_nearby_entities(session, "test-world")
173 assert nearby == []
174
175
176class TestGetRecentEntries:
177 """Tests for CampaignLog.get_recent_entries()."""
178
179 def test_returns_current_day_entries(self, ctx: ToolContext):
180 ctx.campaign_log.append_entry("Arrived at the tavern", "10 min")
181 ctx.campaign_log.append_entry("Spoke with bartender", "15 min")
182
183 entries = ctx.campaign_log.get_recent_entries(days=1)
184 events = [e.event for e in entries]
185 assert "Arrived at the tavern" in events
186 assert "Spoke with bartender" in events
187
188 def test_returns_previous_day_entries(self, ctx: ToolContext):
189 ctx.campaign_log.append_entry("Morning event", "10 min")
190 ctx.campaign_log.append_entry("Traveled all day", "18 hours")
191 ctx.campaign_log.append_entry("Day 2 event", "10 min")
192
193 entries = ctx.campaign_log.get_recent_entries(days=2)
194 events = [e.event for e in entries]
195 assert "Morning event" in events
196 assert "Day 2 event" in events
197
198 def test_respects_days_limit(self, ctx: ToolContext):
199 ctx.campaign_log.append_entry("Day 1 event", "18 hours")
200 ctx.campaign_log.append_entry("Day 2 event", "18 hours")
201 ctx.campaign_log.append_entry("Day 3 event", "1 hour")
202
203 entries = ctx.campaign_log.get_recent_entries(days=1)
204 events = [e.event for e in entries]
205 assert "Day 3 event" in events
206 assert "Day 1 event" not in events
207
208 def test_empty_log(self, ctx: ToolContext):
209 entries = ctx.campaign_log.get_recent_entries(days=2)
210 assert entries == []
211
212
213class TestBuildPlanningContext:
214 """Tests for assembling the planner's context."""
215
216 def test_includes_session_info(self, populated_world: ToolContext):
217 save_session(
218 "default",
219 {
220 "location": "Town Square",
221 "body": (
222 "## Situation\nThe player just arrived.\n\n"
223 "## Open Threads\n- Find the missing cat"
224 ),
225 },
226 )
227 context = build_planning_context(
228 world_id=populated_world.world_id,
229 player_id=populated_world.player_id,
230 candidates=[],
231 )
232 assert "Town Square" in context
233 assert "missing cat" in context
234
235 def test_includes_candidate_content(
236 self,
237 populated_world: ToolContext,
238 tmp_path: Path,
239 ):
240 save_session(
241 "default",
242 {"location": "Town Square", "body": ""},
243 )
244 path = tmp_path / "worlds/test-world/npcs/Thin NPC.md"
245 context = build_planning_context(
246 world_id=populated_world.world_id,
247 player_id=populated_world.player_id,
248 candidates=[("Thin NPC", path)],
249 )
250 assert "Thin NPC" in context
251 assert "Someone with no inner life" in context
252
253 def test_empty_candidates(self, populated_world: ToolContext):
254 save_session(
255 "default",
256 {"location": "Town Square", "body": ""},
257 )
258 context = build_planning_context(
259 world_id=populated_world.world_id,
260 player_id=populated_world.player_id,
261 candidates=[],
262 )
263 assert "No entities" in context
264
265 def test_includes_recent_log_entries(self, populated_world: ToolContext):
266 save_session(
267 "default",
268 {"location": "Town Square", "body": ""},
269 )
270 populated_world.campaign_log.append_entry(
271 "Fought three goblins in the clearing", "5 rounds"
272 )
273 populated_world.campaign_log.append_entry(
274 "Spotted two lookouts near the old mill", "30 min"
275 )
276
277 context = build_planning_context(
278 world_id=populated_world.world_id,
279 player_id=populated_world.player_id,
280 candidates=[],
281 )
282 assert "Recent Events" in context
283 assert "three goblins" in context
284 assert "two lookouts" in context
285
286
287class TestPlanWorld:
288 """Tests for the plan_world orchestrator."""
289
290 def test_dry_run_returns_candidates(self, populated_world: ToolContext):
291 save_session(
292 "default",
293 {
294 "location": "Town Square",
295 "body": "## Present\n- [[Thin NPC]]",
296 },
297 )
298 result = plan_world(
299 world_id=populated_world.world_id,
300 player_id=populated_world.player_id,
301 dry_run=True,
302 )
303 assert isinstance(result, PlanResult)
304 assert result.dry_run is True
305 assert len(result.candidates) > 0
306 names = [name for name, _, _ in result.candidates]
307 assert "Thin NPC" in names
308
309 def test_dry_run_respects_threshold(self, populated_world: ToolContext):
310 save_session(
311 "default",
312 {"location": "Town Square", "body": ""},
313 )
314 result = plan_world(
315 world_id=populated_world.world_id,
316 player_id=populated_world.player_id,
317 dry_run=True,
318 threshold=0.0,
319 )
320 assert len(result.candidates) == 0
321
322 def test_dry_run_respects_max_entities(self, populated_world: ToolContext):
323 save_session(
324 "default",
325 {"location": "Town Square", "body": ""},
326 )
327 result = plan_world(
328 world_id=populated_world.world_id,
329 player_id=populated_world.player_id,
330 dry_run=True,
331 max_entities=1,
332 )
333 assert len(result.candidates) <= 1
334
335 def test_no_session_returns_empty(self, ctx: ToolContext):
336 result = plan_world(
337 world_id=ctx.world_id,
338 player_id=ctx.player_id,
339 dry_run=True,
340 )
341 assert len(result.candidates) == 0
342
343 @patch("storied.claude.subprocess.Popen")
344 def test_plan_world_calls_claude(
345 self, mock_popen: MagicMock, populated_world: ToolContext
346 ):
347 save_session(
348 "default",
349 {
350 "location": "Town Square",
351 "body": "## Present\n- [[Thin NPC]]",
352 },
353 )
354
355 # Mock subprocess returning a result event
356 result_line = json.dumps(
357 {
358 "type": "result",
359 "session_id": "sess-1",
360 "usage": {"input_tokens": 1000, "output_tokens": 200, "tool_calls": 0},
361 "duration_ms": 5000,
362 }
363 )
364 mock_proc = MagicMock()
365 mock_proc.stdin = MagicMock()
366 mock_proc.stdout = iter([result_line.encode() + b"\n"])
367 mock_proc.stderr = iter([])
368 mock_proc.wait.return_value = 0
369 mock_proc.returncode = 0
370 mock_popen.return_value = mock_proc
371
372 result = plan_world(
373 world_id=populated_world.world_id,
374 player_id=populated_world.player_id,
375 model="claude-opus-4-7",
376 )
377
378 assert result.dry_run is False
379 assert result.input_tokens == 1000
380 assert result.output_tokens == 200
381 mock_popen.assert_called_once()
382
383 @patch("storied.claude.subprocess.Popen")
384 def test_plan_world_counts_tool_calls(
385 self, mock_popen: MagicMock, populated_world: ToolContext
386 ):
387 save_session(
388 "default",
389 {
390 "location": "Town Square",
391 "body": "## Present\n- [[Thin NPC]]",
392 },
393 )
394
395 # Stream with tool_use events followed by result
396 lines = [
397 json.dumps(
398 {
399 "type": "stream_event",
400 "event": {
401 "type": "content_block_start",
402 "index": 1,
403 "content_block": {
404 "type": "tool_use",
405 "id": "t1",
406 "name": "mcp__storied__establish",
407 },
408 },
409 }
410 ),
411 json.dumps(
412 {
413 "type": "result",
414 "session_id": "sess-2",
415 "usage": {"input_tokens": 800, "output_tokens": 100},
416 "duration_ms": 3000,
417 }
418 ),
419 ]
420
421 mock_proc = MagicMock()
422 mock_proc.stdin = MagicMock()
423 mock_proc.stdout = iter([line.encode() + b"\n" for line in lines])
424 mock_proc.stderr = iter([])
425 mock_proc.wait.return_value = 0
426 mock_proc.returncode = 0
427 mock_popen.return_value = mock_proc
428
429 result = plan_world(
430 world_id=populated_world.world_id,
431 player_id=populated_world.player_id,
432 model="claude-opus-4-7",
433 )
434
435 assert result.tool_calls == 1
436
437
438# --- World tick helpers (subprocess-bound tick_world is pragma'd out) ------
439
440
441class TestFindEntitiesWithWill:
442 def test_returns_empty_when_world_missing(self, ctx: ToolContext):
443 results = _find_entities_with_will("nonexistent")
444 assert results == []
445
446 def test_finds_entities_with_will_triggers(self, populated_world: ToolContext):
447 # Establish one with will, one without
448 call_tool(
449 establish,
450 entity_type="npcs",
451 name="Triggered",
452 description="Has triggers.",
453 will=["If approached → flee"],
454 )
455 call_tool(
456 establish,
457 entity_type="npcs",
458 name="Idle",
459 description="No triggers.",
460 )
461 results = _find_entities_with_will(
462 populated_world.world_id,
463 )
464 names = {name for name, _ in results}
465 assert "Triggered" in names
466 assert "Idle" not in names
467
468 def test_skips_missing_type_directories(
469 self,
470 populated_world: ToolContext,
471 ):
472 # The fixture creates npcs and locations but not items/factions/threads.
473 # The function should iterate without crashing.
474 results = _find_entities_with_will(
475 populated_world.world_id,
476 )
477 assert isinstance(results, list)
478
479
480class TestBuildTickContext:
481 def test_includes_current_time(self, populated_world: ToolContext):
482 ctx_str = build_tick_context(
483 populated_world.world_id,
484 populated_world.player_id,
485 entities=[],
486 )
487 assert "Current Game Time" in ctx_str
488
489 def test_includes_session_location(self, populated_world: ToolContext):
490 save_session(
491 "default",
492 {
493 "location": "Town Square",
494 "body": "## Present\n- [[Old Gregor]]",
495 },
496 )
497
498 ctx_str = build_tick_context(
499 populated_world.world_id,
500 populated_world.player_id,
501 entities=[],
502 )
503 assert "Town Square" in ctx_str
504 assert "Old Gregor" in ctx_str
505
506 def test_includes_recent_events(self, populated_world: ToolContext):
507 populated_world.campaign_log.append_entry(
508 "Met the merchant",
509 "10 min",
510 )
511 populated_world.campaign_log.append_entry(
512 "Found the secret door",
513 "5 min",
514 )
515 ctx_str = build_tick_context(
516 populated_world.world_id,
517 populated_world.player_id,
518 entities=[],
519 )
520 assert "Recent Events" in ctx_str
521 assert "Met the merchant" in ctx_str
522
523 def test_includes_entity_with_triggers(self, populated_world: ToolContext):
524 call_tool(
525 establish,
526 entity_type="npcs",
527 name="Lurker",
528 description="Hides in shadows.",
529 will=["If alone → emerge"],
530 )
531 triggers = _find_entities_with_will(
532 populated_world.world_id,
533 )
534 ctx_str = build_tick_context(
535 populated_world.world_id,
536 populated_world.player_id,
537 entities=triggers,
538 )
539 assert "Active Triggers" in ctx_str
540 assert "Lurker" in ctx_str
541 assert "If alone → emerge" in ctx_str
542
543
544class TestBackgroundTicker:
545 """Cover the non-subprocess paths of BackgroundTicker.
546
547 The threaded `_run` method is pragma'd because it spawns a real claude
548 subprocess. Everything around it (init, day-tracking, no-op fast paths)
549 is testable here.
550 """
551
552 def test_init_stores_state(self, tmp_path: Path):
553 ticker = BackgroundTicker(
554 world_id="test",
555 player_id="default",
556 )
557 assert ticker._world_id == "test"
558 assert ticker._last_tick_day == 0
559 assert ticker._thread is None
560
561 def test_maybe_tick_skips_when_day_unchanged(
562 self,
563 populated_world: ToolContext,
564 ):
565 ticker = BackgroundTicker(
566 world_id=populated_world.world_id,
567 player_id=populated_world.player_id,
568 )
569 ticker._last_tick_day = populated_world.campaign_log.current_day
570 ticker.maybe_tick(populated_world.campaign_log)
571 # No thread should have been spawned
572 assert ticker._thread is None
573
574 def test_maybe_tick_skips_when_no_triggers(self, ctx: ToolContext):
575 # ctx has a fresh empty world with no entities → no triggers,
576 # so maybe_tick should mark the day done without spawning anything.
577 ticker = BackgroundTicker(
578 world_id=ctx.world_id,
579 player_id=ctx.player_id,
580 )
581 # Advance the day so the day-changed guard passes
582 ctx.campaign_log.append_entry("Travel", "18 hours")
583 ticker.maybe_tick(ctx.campaign_log)
584 assert ticker._thread is None
585 assert ticker._last_tick_day == ctx.campaign_log.current_day
586
587 def test_pop_result_returns_none_when_no_thread(self, tmp_path: Path):
588 ticker = BackgroundTicker(
589 world_id="test",
590 player_id="default",
591 )
592 assert ticker.pop_result() is None