feat(story): let a story span scenes and end when its arc resolves

docs/3_loops_and_systems/story_loop.md

@@ -12,6 +12,26 @@ between them, and finalize when the arc completes.
 **Source:** `packages/agents/src/monitor_agents/loops/story_loop.py`
 (`build_story_graph`). Delegates each scene to the [Scene Loop](./scene_loop.md).
 
+## Arc contract
+
+A story spans several scenes. Finishing a scene advances the story to the next
+one; the story itself ends when its arc resolves.
+
+The arc runs through four phases, and the story's `scene_budget` (5 scenes by
+default) decides how many scenes each phase takes:
+
+| Phase | Scenes | What happens |
+|---|---|---|
+| `rising_action` | until `budget - 3` | Tension climbs; the story opens its plot threads. |
+| `climax` | `budget - 2` | The conflict is confronted and the plot threads are paid off. |
+| `falling_action` | `budget - 1` | The aftermath of the climax. |
+| `resolution` | `budget` | The arc resolves and the story is complete. |
+
+Because the phases are ordered, the arc always reaches its resolution — a
+story cannot stall. A caller that wants a longer or shorter campaign sets
+`scene_budget` on the `StoryState`; a caller that wants the story to end early
+passes `story_complete` in the scene outcome.
+
 ## Nodes
 
 - `init_story` — establish arc parameters + initial world state.
@@ -22,5 +42,12 @@ between them, and finalize when the arc completes.
 - `transition` — update continuity + plot threads before the next scene.
 - `finalize` — wrap the arc; ensure final world state is canonized.
 
+## State
+
+The story state is checkpointed after every scene, so a campaign that is
+interrupted — or resumed in another process — keeps its scenes, its arc
+position and its in-game time. `StoryLoop.start_or_resume()` returns the
+checkpointed story when there is one instead of starting over.
+
 ## See Also
 - [Scene Loop](./scene_loop.md) · [Loops Index](./_index.md)

packages/agents/src/monitor_agents/indexer/agent.py

@@ -766,8 +766,13 @@ async def _chunk_pdf_with_structure(
         # worker has no shared state race with the chunking loop below.
         blocks_with_tags: list[tuple[Any, str]] = []
         for block in section_blocks:
-            heading_text = " > ".join(block.heading_path)
-            tag_category = TagPool.classify_text(heading_text) or "general"
+            # A subsection gives a more precise cue than its parent chapter.
+            # Fall back through ancestors when a heading has no known tag.
+            tag_category = next(
+                (category for heading in reversed(block.heading_path)
+                 if (category := TagPool.classify_text(heading)) and category != "general"),
+                "general",
+            )
             blocks_with_tags.append((block, tag_category))
 
         sem = asyncio.Semaphore(_SECTION_CATEGORIZATION_CONCURRENCY)
@@ -848,6 +853,9 @@ async def _classify(block: Any, tag_category: str) -> str:
                     "heading_path": block.heading_path,
                     "section_path": section_path,
                     "section_depth": block.depth,
+                    # These are section bounds, not exact chunk citations.
+                    "section_page_start": page_num,
+                    "section_page_end": block.page_end + 1,
                 },
             )
             for chunk in block_chunks:

packages/agents/src/monitor_agents/ingestion/agent.py

@@ -69,7 +69,7 @@
 import traceback
 from datetime import UTC, datetime
 from pathlib import Path
-from typing import Any
+from typing import Any, Callable
 from uuid import UUID
 
 from monitor_data.db.minio import get_minio_client
@@ -241,6 +241,7 @@ async def ingest_file(
         content_type: str | None = None,
         reuse_source_id: UUID | None = None,
         reuse_doc_id: UUID | None = None,
+        on_job_created: Callable[[UUID], None] | None = None,
     ) -> IngestionJobResponse:
         """
         Ingest a raw file through the full pipeline.
@@ -264,6 +265,7 @@ async def ingest_file(
             multiverse_id: Optional multiverse for auto-applying the KnowledgePack.
             auto_apply:    If True, immediately creates ProposedChanges from the pack.
             content_type:  MIME type override (auto-detected from filename if None).
+            on_job_created: Called with the persisted job ID as soon as it is created.
 
         Returns:
             IngestionJobResponse tracking pipeline progress.
@@ -342,6 +344,8 @@ async def ingest_file(
                     processing_checklist=selected_layers,
                 ),
             )
+            if on_job_created is not None:
+                on_job_created(job.job_id)
             return await asyncio.to_thread(
                 mongodb_update_ingestion_job,
                 job.job_id,
@@ -433,6 +437,8 @@ async def ingest_file(
                 ),
             )
             job_id = job.job_id
+            if on_job_created is not None:
+                on_job_created(job_id)
             await asyncio.to_thread(
                 mongodb_update_ingestion_job,
                 job_id,

packages/agents/src/monitor_agents/loops/conversation_loop.py

@@ -369,12 +369,15 @@ async def generate_npc_responses(state: ConversationState) -> dict[str, Any]:
 
     for npc_id in state.npc_ids:
         try:
+            # Each NPC hears the replies the others already gave this round,
+            # so a group answers as a group instead of in isolation.
+            conversation_history = [*prior_turns, *new_turns[len(state.turns) :]]
             if state.mode == ConversationMode.DIRECT:
                 result = await agent.respond_direct(
                     conversation_id=state.conversation_id,
                     npc_id=npc_id,
                     player_said=state.current_player_input,
-                    conversation_history=prior_turns,
+                    conversation_history=conversation_history,
                     player_entity_id=state.player_entity_id,
                     scene_id=state.scene_id,
                     story_id=state.story_id,
@@ -445,7 +448,7 @@ async def generate_npc_responses(state: ConversationState) -> dict[str, Any]:
                     conversation_id=state.conversation_id,
                     npc_id=npc_id,
                     gm_question=state.current_player_input,
-                    conversation_history=prior_turns,
+                    conversation_history=conversation_history,
                     source_profile=state.source_profile,
                     npc_data=state.npc_contexts.get(str(npc_id)),
                 )

packages/agents/src/monitor_agents/loops/scene_loop.py

@@ -221,6 +221,9 @@ class SceneState(BaseModel):
 
     # Flow control
     scene_complete: bool = False
+    # Set by ``SceneLoop.finalize`` so the graph closes the scene without
+    # resolving or narrating an empty turn first.
+    finalize_requested: bool = False
     turns_count: int = 0
     max_turns: int = 50  # safety ceiling per scene
 
@@ -1475,7 +1478,13 @@ def build_scene_graph() -> StateGraph:
     graph.set_entry_point("load_context")
 
     # Edges
-    graph.add_edge("load_context", "resolve")
+    # A finalization request closes the scene straight away — no resolve, no
+    # narration of an empty turn.
+    graph.add_conditional_edges(
+        "load_context",
+        lambda state: "complete_current_scene" if state.finalize_requested else "resolve",
+        {"complete_current_scene": "complete_current_scene", "resolve": "resolve"},
+    )
     graph.add_conditional_edges(
         "resolve",
         route_after_resolve,
@@ -1771,6 +1780,7 @@ async def finalize(self) -> None:
                     "scene_id": self.scene_id,
                     "story_id": self.story_id,
                     "scene_complete": True,
+                    "finalize_requested": True,
                     "user_input": None,
                     "memories_to_persist": [],
                     "max_turns": self.max_turns,

packages/agents/src/monitor_agents/loops/story_loop.py

@@ -71,6 +71,13 @@ class StoryState(BaseModel):
     scenes: list[UUID] = Field(default_factory=list)
     current_scene_id: UUID | None = None
     scenes_completed: int = 0
+    # How many scenes the arc runs before it resolves. The arc spends the
+    # last three on the climax, the falling action and the resolution.
+    scene_budget: int = Field(
+        default=5,
+        ge=3,
+        description="Number of scenes in the story arc before it resolves.",
+    )
 
     # Flow control
     story_complete: bool = False
@@ -356,6 +363,12 @@ async def start_or_resume(self) -> dict[str, Any]:
             thread_id = f"story-{self.story_id}"
             config = {"configurable": {"thread_id": thread_id}}
 
+            # A story already under way keeps the state it checkpointed —
+            # re-running the graph here would reset it.
+            snapshot = await compiled.aget_state(config)  # type: ignore[arg-type]
+            if snapshot.values:
+                return dict(snapshot.values)
+
             initial = StoryState(
                 story_id=self.story_id,
                 universe_id=self.universe_id,
@@ -373,6 +386,8 @@ async def advance(self, story_complete: bool = False) -> dict[str, Any]:
             story_id=self.story_id,
             universe_id=self.universe_id,
             story_complete=story_complete,
+            # The scene this call follows has been played.
+            scenes_completed=1,
         )
         if story_complete:
             await finalize_story(state)
@@ -416,45 +431,54 @@ async def complete_current_scene(
         Returns:
             Updated StoryState dict.
         """
-        force_complete = bool((outcome or {}).get("scene_complete", False))
+        # A finished scene advances the story; the story itself ends when its
+        # arc resolves, or when the caller says so outright.
+        force_complete = bool((outcome or {}).get("story_complete", False))
         duration = int((outcome or {}).get("total_minutes_passed") or 0)
 
         with MongoDBSaver.from_conn_string(settings.mongodb_uri) as checkpointer:
             compiled = self._graph.compile(checkpointer=checkpointer)
             thread_id = f"story-{self.story_id}"
             config = {"configurable": {"thread_id": thread_id}}
 
-            # Load current checkpoint state
-            try:
-                current = await compiled.aget_state(config)  # type: ignore[arg-type]
-                state_values = dict(current.values) if current.values else {}
-            except Exception:  # noqa: BLE001  # reason: falls back to `{}` on failure (in complete_current_scene)
-                state_values = {}
+            # Load current checkpoint state. A read failure must not pass for
+            # "no story yet" — that would silently restart the campaign and
+            # overwrite its checkpoint.
+            current = await compiled.aget_state(config)  # type: ignore[arg-type]
+            state_values = dict(current.values) if current.values else {}
 
-            # Build state with scene completion context
+            # Build state with scene completion context, keeping every other
+            # checkpointed field (in-game time, agreements, player pitch).
             previous_scenes = list(state_values.get("scenes", []))
             previous_history = list(state_values.get("scenes_history", []))
             previous_threads = list(state_values.get("active_threads", []))
             previous_completed = list(state_values.get("completed_threads", []))
+            # A story that already resolved stays resolved.
+            story_complete = force_complete or bool(state_values.get("story_complete", False))
 
-            run_state = StoryState(
-                story_id=self.story_id,
-                universe_id=self.universe_id,
-                current_scene_id=scene_id,
-                story_complete=force_complete,
-                last_scene_duration_minutes=duration,
-                scenes=previous_scenes,
-                scenes_completed=state_values.get("scenes_completed", 0) + 1,
-                scenes_history=previous_history,
-                world_tone=state_values.get("world_tone", "dramatic"),
-                arc_label=state_values.get("arc_label", "rising_action"),
-                tension_score=float(state_values.get("tension_score", 0.3)),
-                active_threads=previous_threads,
-                completed_threads=previous_completed,
+            run_state = StoryState.model_validate(
+                {
+                    **state_values,
+                    "story_id": self.story_id,
+                    "universe_id": self.universe_id,
+                    "current_scene_id": scene_id,
+                    "story_complete": story_complete,
+                    "last_scene_duration_minutes": duration,
+                    "scenes": previous_scenes,
+                    "scenes_completed": state_values.get("scenes_completed", 0) + 1,
+                    "scenes_history": previous_history,
+                    "active_threads": previous_threads,
+                    "completed_threads": previous_completed,
+                }
             )
 
-            if force_complete:
+            if story_complete:
                 await finalize_story(run_state)
+                await compiled.aupdate_state(
+                    config,  # type: ignore[arg-type]
+                    run_state.model_dump(),
+                    as_node="init_story",
+                )
                 return {"story_complete": True, "completed_scene_id": scene_id}
 
             # Step 1: Run world simulation
@@ -510,6 +534,11 @@ async def complete_current_scene(
 
             if arc_update.get("story_complete"):
                 await finalize_story(run_state)
+                await compiled.aupdate_state(
+                    config,  # type: ignore[arg-type]
+                    run_state.model_dump(),
+                    as_node="init_story",
+                )
                 return {
                     **run_state.model_dump(),
                     **progression_update,
@@ -518,9 +547,14 @@ async def complete_current_scene(
 
             # Step 3: Create next scene via transition
             transition_update = await transition_scene(run_state)
+            checkpoint_state = {**run_state.model_dump(), **transition_update}
+            await compiled.aupdate_state(
+                config,  # type: ignore[arg-type]
+                checkpoint_state,
+                as_node="init_story",
+            )
             return {
-                **run_state.model_dump(),
-                **transition_update,
+                **checkpoint_state,
                 **progression_update,
                 "agenda_moves": agenda_moves,
                 "completed_scene_id": scene_id,