brink_runtime/session.rs
1//! Story sessions: a journaling, replayable wrapper around [`Story`].
2//!
3//! [`StorySession`] *composes* a [`Story`] (which itself wraps a
4//! [`FlowInstance`](crate::FlowInstance) + [`World`](crate::World)) with a
5//! serializable [`SessionJournal`]. The VM never learns about journaling — the
6//! journal observes inputs at the session boundary (the same place the VM
7//! receives them), so instrumentation composes instead of threading an
8//! `if observer` branch through the stepping hot loop. This generalizes the
9//! in-memory [`ReplayRecorder`](crate::ReplayRecorder): where the recorder
10//! captured only external results for hot-reload, the journal captures every
11//! *input* that entered the VM (start, choices, externals, mutations) as
12//! durable, serde-serializable data.
13//!
14//! Consumers:
15//! - **`bevy-brink`** — first-class sessions/replay/save-load.
16//! - **`brink-web`** — wasm bindings expose [`StorySession`] on the web.
17//! - **`@brink/studio-store`** — `LocalSessionProvider` migrates onto this.
18//!
19//! There is no JS-side journal: the journal serializes to JSON via serde and
20//! that JSON is the durable save artifact. See `docs/story-session-spec.md`
21//! (#370, plus the snapshot half of #371).
22//!
23//! ## Turn-boundary contract
24//!
25//! `set_var` / `go_to_path` / `load_state` are **turn-boundary only**. The
26//! session rejects them mid-turn (status [`Active`](crate::StoryStatus::Active),
27//! i.e. more content is pending) with
28//! [`SessionError::MutationMidTurn`], rather than queuing them. This is the
29//! documented "one behavior" the spec permits — reject, not queue. A caller
30//! drains the current turn (to `Done`/`Choices`/`End`) before mutating, which
31//! keeps the journal event order unambiguous. The schema reserves a per-event
32//! `anchor` so exact mid-turn replay can arrive additively without a format
33//! break.
34//!
35//! ## Escape hatch
36//!
37//! [`StorySession::story`] / [`StorySession::story_mut`] expose the wrapped
38//! [`Story`]. Anything done through them **bypasses the journal** — the
39//! documented journal-bypass contract. Foreign / shared flows (#200) reached
40//! this way never journal, matching the "journaling window" gate: only the
41//! session's own `advance` / `choose` / `resolve_external` frames record.
42
43use alloc::borrow::ToOwned;
44use alloc::boxed::Box;
45use alloc::collections::{BTreeMap, VecDeque};
46use alloc::string::{String, ToString};
47use alloc::vec::Vec;
48use core::cell::RefCell;
49use core::mem;
50
51use brink_format::{SaveState, Value};
52use serde::{Deserialize, Serialize};
53
54use crate::error::RuntimeError;
55use crate::rng::{FastRng, StoryRng};
56use crate::story::Story;
57use crate::story::{
58 ExternalFnHandler, ExternalResult, FallbackHandler, Step, StepOutcome, StoryStatus,
59};
60
61/// Current [`SessionJournal`] format version.
62pub const SESSION_JOURNAL_VERSION: u32 = 1;
63
64/// Upper bound on journal events (unbounded-growth guard, mirroring
65/// [`RECORDING_CAP`](crate::RECORDING_CAP)). Beyond it, appends are dropped and
66/// [`SessionJournal::truncated`] is set — the journal degrades honestly and
67/// restore falls back to the embedded [`checkpoint`](SessionJournal::checkpoint).
68pub const SESSION_JOURNAL_CAP: usize = 65_536;
69
70// ── Journal ──────────────────────────────────────────────────────────────────
71
72/// One ordered log of every input that entered the VM during a session, plus a
73/// terminal fast-restore [`checkpoint`](Self::checkpoint).
74///
75/// Serde-serializable; the canonical durable save artifact. Values serialize
76/// **tagged** (via [`Value`]'s derived enum representation and
77/// [`SaveState`]'s `BTreeMap<String, Value>`) — no lossy `List`/`Divert` → null
78/// mapping. Deterministic: event order is insertion order; embedded maps are
79/// `BTreeMap`.
80#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
81pub struct SessionJournal {
82 /// Format version (see [`SESSION_JOURNAL_VERSION`]).
83 pub version: u32,
84 /// Checksum of the program this journal was recorded against, so replay can
85 /// detect a recompile and decide fast-restore vs full replay.
86 pub program_checksum: u32,
87 /// RNG seed applied at session start, if the host seeded one.
88 pub seed: Option<u64>,
89 /// Ordered inputs, in the order they entered the VM.
90 pub events: Vec<JournalEvent>,
91 /// Set when the cap was hit or a divergence truncated the log.
92 pub truncated: bool,
93 /// Fast-restore: terminal game-state snapshot (ruling: embedded `SaveState`
94 /// in v1). Present once the session has produced state worth checkpointing.
95 #[serde(default, skip_serializing_if = "Option::is_none")]
96 pub checkpoint: Option<SaveState>,
97}
98
99impl SessionJournal {
100 /// A fresh, empty journal bound to `program_checksum`.
101 #[must_use]
102 pub fn new(program_checksum: u32, seed: Option<u64>) -> Self {
103 Self {
104 version: SESSION_JOURNAL_VERSION,
105 program_checksum,
106 seed,
107 events: Vec::new(),
108 truncated: false,
109 checkpoint: None,
110 }
111 }
112
113 /// Append `event`, respecting [`SESSION_JOURNAL_CAP`]. Beyond the cap the
114 /// event is dropped and [`truncated`](Self::truncated) is set.
115 fn push(&mut self, event: JournalEvent) {
116 if self.events.len() >= SESSION_JOURNAL_CAP {
117 self.truncated = true;
118 return;
119 }
120 self.events.push(event);
121 }
122
123 /// Number of recorded events.
124 #[must_use]
125 pub fn len(&self) -> usize {
126 self.events.len()
127 }
128
129 /// Whether nothing has been recorded.
130 #[must_use]
131 pub fn is_empty(&self) -> bool {
132 self.events.is_empty()
133 }
134}
135
136/// One input that entered the VM.
137///
138/// The reserved `anchor` / `flow` dimensions are serialized (as `Option`, both
139/// defaulting to `None`) but **not interpreted** in v1 — they let mid-turn
140/// anchoring (`anchor`) and multi-flow journaling (`flow`) arrive additively
141/// without a format break. See the module-level turn-boundary contract.
142#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
143pub struct JournalEvent {
144 /// The input kind + payload.
145 pub kind: EventKind,
146 /// Reserved: per-event position ordinal for future mid-turn anchoring.
147 /// Serialized, never interpreted in v1.
148 #[serde(default, skip_serializing_if = "Option::is_none")]
149 pub anchor: Option<u64>,
150 /// Reserved: flow tag for future multi-flow journaling. v1 is
151 /// default-flow-only. Serialized, never interpreted in v1.
152 #[serde(default, skip_serializing_if = "Option::is_none")]
153 pub flow: Option<String>,
154}
155
156impl JournalEvent {
157 /// A v1 event with the reserved dimensions left `None`.
158 #[must_use]
159 pub fn new(kind: EventKind) -> Self {
160 Self {
161 kind,
162 anchor: None,
163 flow: None,
164 }
165 }
166}
167
168/// The kind + payload of a [`JournalEvent`].
169#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
170#[serde(tag = "type", rename_all = "snake_case")]
171pub enum EventKind {
172 /// Session start / play-from-here. `path` is `None` for the default root
173 /// entry, `Some` for a `ChoosePathString` start.
174 Start {
175 #[serde(default, skip_serializing_if = "Option::is_none")]
176 path: Option<String>,
177 #[serde(default, skip_serializing_if = "Vec::is_empty")]
178 args: Vec<Value>,
179 },
180 /// A choice selection. `label` is the choice text as seen (advisory — used
181 /// only for label-drift warnings, never as the selection key).
182 Choice {
183 index: u32,
184 #[serde(default, skip_serializing_if = "Option::is_none")]
185 label: Option<String>,
186 },
187 /// An external-function result, captured where the session's own frame
188 /// received it (the journaling-window gate).
189 External {
190 name: String,
191 #[serde(default, skip_serializing_if = "Vec::is_empty")]
192 args: Vec<Value>,
193 result: Value,
194 },
195 /// A host `set_var` (turn-boundary only).
196 SetVar { name: String, value: Value },
197 /// A host `go_to_path` / `ChoosePathString` (turn-boundary only).
198 GoToPath {
199 path: String,
200 #[serde(default, skip_serializing_if = "Vec::is_empty")]
201 args: Vec<Value>,
202 },
203 /// A host `load_state` (turn-boundary only).
204 LoadState { state: SaveState },
205 /// A journaled `call_function`. The function's *own* externals are resolved
206 /// through the isolated (non-journaling) handler path — only the top-level
207 /// call is journaled here.
208 Call {
209 name: String,
210 #[serde(default, skip_serializing_if = "Vec::is_empty")]
211 args: Vec<Value>,
212 },
213}
214
215// ── Replay ───────────────────────────────────────────────────────────────────
216
217/// How replay obtains external values.
218#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
219pub enum ExternalReplayMode {
220 /// Default. Serve `External` events from the journal (recorded results). No
221 /// effect re-fires; reads stay faithful.
222 #[default]
223 Recorded,
224 /// Re-invoke externals live against the supplied handler. A handler that
225 /// defers ([`ExternalResult::Pending`]) parks the replay as
226 /// [`ReplayOutcome::Failed`] with [`FailReason::AwaitingExternal`]; resume
227 /// via [`StorySession::continue_replay`].
228 Live,
229}
230
231/// Outcome of replaying a journal prefix against a program.
232///
233/// Typed, never silent, never panicking. `Serialize`/`Deserialize` (tagged
234/// `type`, `snake_case`) so wasm/other bindings can mirror this shape verbatim
235/// instead of hand-rolling a parallel JS type — matches the `EventKind`/
236/// `JournalEvent` convention above.
237#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
238#[serde(tag = "type", rename_all = "snake_case")]
239pub enum ReplayOutcome {
240 /// The prefix replayed successfully. `warnings` collects soft issues (e.g.
241 /// choice label drift at a matching index).
242 Replayed { warnings: Vec<ReplayWarning> },
243 /// Replay diverged at `at_event`: the recorded event could not be applied
244 /// against the current program. The journal is truncated at that point and
245 /// the session is parked at the reached position.
246 Diverged {
247 at_event: usize,
248 expected: Box<JournalEvent>,
249 found: DivergenceFound,
250 },
251 /// Replay failed at `at_event` for a non-divergence reason.
252 Failed { at_event: usize, reason: FailReason },
253}
254
255/// A non-fatal replay observation.
256#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
257#[serde(tag = "type", rename_all = "snake_case")]
258pub enum ReplayWarning {
259 /// A choice replayed by index, but its recorded label differs from the
260 /// label now presented at that index (a soft signal the story text drifted
261 /// under the same choice ordering).
262 ChoiceLabelDrift {
263 at_event: usize,
264 index: u32,
265 recorded: String,
266 found: String,
267 },
268}
269
270/// What was found at a divergence point instead of the recorded event.
271#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
272#[serde(tag = "type", rename_all = "snake_case")]
273pub enum DivergenceFound {
274 /// A choice index the current program does not present (out of range).
275 ChoiceIndexOutOfRange { index: u32, available: usize },
276 /// The session was not waiting for a choice when a `Choice` event replayed.
277 NotWaitingForChoice,
278 /// A path that no longer resolves in the current program.
279 UnknownPath { path: String },
280 /// The event kind cannot be applied from the reached state (e.g. a `Start`
281 /// after the session already started).
282 UnexpectedEvent,
283}
284
285/// Why replay stopped without diverging.
286///
287/// `RuntimeError` is a struct variant (`{ message: String }`), not a newtype
288/// (`RuntimeError(String)`) — serde's internally-tagged representation
289/// (`#[serde(tag = "type")]`) cannot serialize a newtype variant wrapping a
290/// non-map payload like a bare `String`.
291#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
292#[serde(tag = "type", rename_all = "snake_case")]
293pub enum FailReason {
294 /// A runtime error surfaced during stepping.
295 RuntimeError { message: String },
296 /// A step/line budget was exceeded (the caller can restart fresh).
297 Budget,
298 /// Live replay hit a deferred external and parked. Resolve it and call
299 /// [`StorySession::continue_replay`].
300 AwaitingExternal { name: String },
301}
302
303// ── Snapshot / diff ──────────────────────────────────────────────────────────
304
305/// A typed, name-resolved snapshot of a session's game state.
306///
307/// A NEW typed serialization path — distinct from the string-valued
308/// [`DebugSnapshot`](crate::DebugSnapshot). Globals keep their [`Value`]s (list
309/// membership included via [`SnapshotList`]); callstack is summarized to frame
310/// kinds + resolved locations. Deterministic (`BTreeMap` / sorted vectors).
311#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
312pub struct StateSnapshot {
313 /// Global variables by name, typed. `BTreeMap` for determinism.
314 pub globals: BTreeMap<String, Value>,
315 /// Resolved list memberships for any `List`-valued global, keyed by
316 /// variable name (item names, sorted). Complements `globals` for consumers
317 /// that want membership without decoding `DefinitionId`s.
318 pub lists: BTreeMap<String, SnapshotList>,
319 /// Global turn index.
320 pub turn_index: u32,
321 /// Per-knot/stitch visit counts, keyed by resolved path, sorted.
322 ///
323 /// Path-keyed projection: counts for scopes with no resolvable author
324 /// path (anonymous counted containers — gathers, choice points — keyed
325 /// only by hash id) are **omitted** here. This is the known projection
326 /// limit of the typed snapshot; the full id-keyed counts remain available
327 /// via [`StorySession::save_state`].
328 pub visit_counts: BTreeMap<String, u32>,
329 /// Per-knot/stitch turn-since counts, keyed by resolved path, sorted.
330 /// Same path-keyed projection limit as
331 /// [`visit_counts`](Self::visit_counts).
332 pub turn_counts: BTreeMap<String, u32>,
333 /// Callstack summary of the default flow, innermost frame first.
334 pub call_stack: Vec<SnapshotFrame>,
335 /// Execution status.
336 pub status: SnapshotStatus,
337}
338
339/// Resolved membership of a `List`-valued global.
340#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
341pub struct SnapshotList {
342 /// Active item names, sorted for determinism.
343 pub items: Vec<String>,
344}
345
346/// One summarized call frame.
347#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
348pub struct SnapshotFrame {
349 /// Frame kind: `root` / `function` / `tunnel` / `thread` / `external` / `eval`.
350 pub kind: String,
351 /// Nearest named container for this frame, if resolvable.
352 #[serde(default, skip_serializing_if = "Option::is_none")]
353 pub location: Option<String>,
354 /// Number of temporaries in this frame.
355 pub temps: usize,
356}
357
358/// Session execution status, serde-friendly.
359#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
360#[serde(rename_all = "snake_case")]
361pub enum SnapshotStatus {
362 Active,
363 WaitingForChoice,
364 Done,
365 Ended,
366}
367
368impl From<StoryStatus> for SnapshotStatus {
369 fn from(s: StoryStatus) -> Self {
370 match s {
371 StoryStatus::Active => Self::Active,
372 StoryStatus::WaitingForChoice => Self::WaitingForChoice,
373 StoryStatus::Done => Self::Done,
374 StoryStatus::Ended => Self::Ended,
375 }
376 }
377}
378
379/// A pure diff between two [`StateSnapshot`]s (see [`diff`]).
380#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
381pub struct StateDiff {
382 /// Globals present in `b` but not `a`.
383 pub added_globals: BTreeMap<String, Value>,
384 /// Globals present in `a` but not `b`.
385 pub removed_globals: BTreeMap<String, Value>,
386 /// Globals whose value changed, mapped to `(before, after)`.
387 pub changed_globals: BTreeMap<String, (Value, Value)>,
388 /// Per-list membership deltas for lists that changed, keyed by var name.
389 pub list_deltas: BTreeMap<String, ListDelta>,
390 /// `turn_index` delta `(before, after)` if it changed.
391 #[serde(default, skip_serializing_if = "Option::is_none")]
392 pub turn_index: Option<(u32, u32)>,
393 /// Callstack frames pushed in `b` relative to `a` (by innermost-first
394 /// comparison): frames appended beyond the common prefix.
395 pub pushed_frames: Vec<SnapshotFrame>,
396 /// Callstack frames popped in `b` relative to `a`.
397 pub popped_frames: Vec<SnapshotFrame>,
398}
399
400impl StateDiff {
401 /// Whether the two snapshots were identical in every compared dimension.
402 #[must_use]
403 pub fn is_empty(&self) -> bool {
404 self.added_globals.is_empty()
405 && self.removed_globals.is_empty()
406 && self.changed_globals.is_empty()
407 && self.list_deltas.is_empty()
408 && self.turn_index.is_none()
409 && self.pushed_frames.is_empty()
410 && self.popped_frames.is_empty()
411 }
412}
413
414/// Membership delta for one list-valued global.
415#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
416pub struct ListDelta {
417 /// Item names present in `b` but not `a`, sorted.
418 pub added: Vec<String>,
419 /// Item names present in `a` but not `b`, sorted.
420 pub removed: Vec<String>,
421}
422
423/// Pure diff of two snapshots. `a` is "before", `b` is "after".
424#[must_use]
425pub fn diff(a: &StateSnapshot, b: &StateSnapshot) -> StateDiff {
426 let mut d = StateDiff::default();
427
428 for (name, av) in &a.globals {
429 match b.globals.get(name) {
430 None => {
431 d.removed_globals.insert(name.clone(), av.clone());
432 }
433 Some(bv) if bv != av => {
434 d.changed_globals
435 .insert(name.clone(), (av.clone(), bv.clone()));
436 }
437 Some(_) => {}
438 }
439 }
440 for (name, bv) in &b.globals {
441 if !a.globals.contains_key(name) {
442 d.added_globals.insert(name.clone(), bv.clone());
443 }
444 }
445
446 // List deltas over the union of list-valued globals.
447 let mut list_names: Vec<&String> = a.lists.keys().chain(b.lists.keys()).collect();
448 list_names.sort_unstable();
449 list_names.dedup();
450 for name in list_names {
451 let empty = SnapshotList { items: Vec::new() };
452 let al = a.lists.get(name).unwrap_or(&empty);
453 let bl = b.lists.get(name).unwrap_or(&empty);
454 if al == bl {
455 continue;
456 }
457 let added: Vec<String> = bl
458 .items
459 .iter()
460 .filter(|i| !al.items.contains(i))
461 .cloned()
462 .collect();
463 let removed: Vec<String> = al
464 .items
465 .iter()
466 .filter(|i| !bl.items.contains(i))
467 .cloned()
468 .collect();
469 if !added.is_empty() || !removed.is_empty() {
470 d.list_deltas
471 .insert(name.clone(), ListDelta { added, removed });
472 }
473 }
474
475 if a.turn_index != b.turn_index {
476 d.turn_index = Some((a.turn_index, b.turn_index));
477 }
478
479 // Callstack: compare from the outermost (root) end. The frames are stored
480 // innermost-first, so reverse to find the common outer prefix.
481 let a_outer: Vec<&SnapshotFrame> = a.call_stack.iter().rev().collect();
482 let b_outer: Vec<&SnapshotFrame> = b.call_stack.iter().rev().collect();
483 let common = a_outer
484 .iter()
485 .zip(b_outer.iter())
486 .take_while(|(x, y)| x == y)
487 .count();
488 // Frames beyond the common prefix in b were pushed; in a were popped.
489 d.pushed_frames = b_outer[common..].iter().map(|f| (*f).clone()).collect();
490 d.popped_frames = a_outer[common..].iter().map(|f| (*f).clone()).collect();
491
492 d
493}
494
495// ── Errors ───────────────────────────────────────────────────────────────────
496
497/// Errors from session-level operations (distinct from VM [`RuntimeError`]s,
498/// which are wrapped).
499#[derive(Debug, thiserror::Error)]
500pub enum SessionError {
501 /// A turn-boundary-only mutation (`set_var` / `go_to_path` / `load_state`)
502 /// was attempted mid-turn (status `Active`). Drain the turn first.
503 #[error(
504 "mutation `{op}` attempted mid-turn; set_var/go_to_path/load_state are turn-boundary only"
505 )]
506 MutationMidTurn { op: &'static str },
507 /// The program checksum in a journal does not match the program being
508 /// restored/replayed against, and no fast-restore checkpoint was usable.
509 #[error("journal program checksum {journal} does not match program {program}")]
510 ChecksumMismatch { journal: u32, program: u32 },
511 /// A wrapped VM error.
512 #[error(transparent)]
513 Runtime(#[from] RuntimeError),
514}
515
516// ── Journaling handler ───────────────────────────────────────────────────────
517
518/// Composes a caller's [`ExternalFnHandler`] and journals every inline-resolved
519/// external where the session's own frame receives it. Generalizes
520/// [`RecordingHandler`](crate::RecordingHandler) from in-memory recording to the
521/// durable journal.
522///
523/// Deferred externals ([`ExternalResult::Pending`]) resolve out-of-band; the
524/// session journals those in [`StorySession::resolve_external`] where it has the
525/// name/args/result.
526struct JournalingHandler<'a, H: ExternalFnHandler + ?Sized> {
527 inner: &'a H,
528 // Interior-mutability: the trait method is `&self`, but we need to append.
529 sink: RefCell<&'a mut Vec<(String, Vec<Value>, Value)>>,
530}
531
532impl<'a, H: ExternalFnHandler + ?Sized> JournalingHandler<'a, H> {
533 fn new(inner: &'a H, sink: &'a mut Vec<(String, Vec<Value>, Value)>) -> Self {
534 Self {
535 inner,
536 sink: RefCell::new(sink),
537 }
538 }
539}
540
541impl<H: ExternalFnHandler + ?Sized> ExternalFnHandler for JournalingHandler<'_, H> {
542 fn call(&self, name: &str, args: &[Value]) -> ExternalResult {
543 let result = self.inner.call(name, args);
544 if let ExternalResult::Resolved(value) = &result {
545 self.sink
546 .borrow_mut()
547 .push((name.to_owned(), args.to_vec(), value.clone()));
548 }
549 result
550 }
551}
552
553/// Serves external values from a recorded journal prefix during replay
554/// (`ExternalReplayMode::Recorded`). Consumes `External` events in order; on
555/// mismatch it falls through to the ink fallback body (never re-invokes).
556struct RecordedReplayHandler<'a> {
557 // (name, args, result) queue, consumed front-to-back.
558 queue: RefCell<&'a mut VecDeque<(String, Vec<Value>, Value)>>,
559}
560
561impl ExternalFnHandler for RecordedReplayHandler<'_> {
562 fn call(&self, name: &str, args: &[Value]) -> ExternalResult {
563 let mut q = self.queue.borrow_mut();
564 match q.front() {
565 Some((n, a, _)) if n == name && a.as_slice() == args => {
566 let (_, _, result) = q.pop_front().unwrap_or_else(|| {
567 // Unreachable: we just matched `front`. Fall back safely.
568 (String::new(), Vec::new(), Value::Null)
569 });
570 ExternalResult::Resolved(result)
571 }
572 _ => ExternalResult::Fallback,
573 }
574 }
575}
576
577// ── Session ──────────────────────────────────────────────────────────────────
578
579/// A journaling, replayable session wrapping a [`Story`].
580///
581/// Owns a [`Story`] + a [`SessionJournal`]. Stepping mirrors
582/// [`Story::advance_with`] exactly ([`StepOutcome`]), recording inputs at the
583/// session boundary. The wrapped story is reachable via [`story`](Self::story) /
584/// [`story_mut`](Self::story_mut) for the documented journal-bypass escape
585/// hatch.
586pub struct StorySession<R: StoryRng = FastRng> {
587 story: Story<R>,
588 journal: SessionJournal,
589 started: bool,
590 /// The un-replayed tail of an in-progress replay that parked on a deferred
591 /// external ([`FailReason::AwaitingExternal`]). `Some` only between the
592 /// park and the [`continue_replay`](Self::continue_replay) that resumes
593 /// it — resuming consumes the tail from this cursor instead of dropping
594 /// the remaining recorded inputs.
595 pending_replay: Option<PendingReplay>,
596}
597
598/// Cursor state for a parked, resumable replay: the remaining source events
599/// (with their original indices, for `at_event` reporting), the
600/// recorded-externals queue, the external mode, warnings accumulated so far,
601/// and the source journal's checkpoint to carry over on completion.
602struct PendingReplay {
603 /// `(original_source_index, event)` pairs not yet applied.
604 remaining: VecDeque<(usize, JournalEvent)>,
605 /// Recorded externals still unserved (`ExternalReplayMode::Recorded`).
606 ext_queue: VecDeque<(String, Vec<Value>, Value)>,
607 mode: ExternalReplayMode,
608 warnings: Vec<ReplayWarning>,
609 /// The source journal's terminal checkpoint, applied to the rebuilt
610 /// journal when the replay completes.
611 source_checkpoint: Option<SaveState>,
612 /// Total events in the source journal (for final-step `at_event`).
613 total_events: usize,
614}
615
616/// Internal outcome of a replay stepping burst: parked on a deferred external
617/// (resumable) or failed terminally.
618enum StepPark {
619 Awaiting { name: String },
620 Fail(FailReason),
621}
622
623impl<R: StoryRng> StorySession<R> {
624 /// Wrap `story` in a fresh session. `seed` is advisory metadata recorded in
625 /// the journal (the host is responsible for actually seeding the story via
626 /// [`Story::set_rng_seed`] before/at start).
627 #[must_use]
628 pub fn new(story: Story<R>, seed: Option<u64>) -> Self {
629 let checksum = story.program().source_checksum();
630 Self {
631 journal: SessionJournal::new(checksum, seed),
632 story,
633 started: false,
634 pending_replay: None,
635 }
636 }
637
638 /// Read-only access to the journal (for export / persistence).
639 #[must_use]
640 pub fn journal(&self) -> &SessionJournal {
641 &self.journal
642 }
643
644 /// Take the journal by value, refreshing its checkpoint first so the
645 /// exported artifact can fast-restore. The session keeps a fresh empty
646 /// journal bound to the same program (rarely needed; export usually clones
647 /// via [`journal`](Self::journal)).
648 pub fn export_journal(&mut self) -> SessionJournal {
649 self.refresh_checkpoint();
650 let checksum = self.journal.program_checksum;
651 let seed = self.journal.seed;
652 mem::replace(&mut self.journal, SessionJournal::new(checksum, seed))
653 }
654
655 /// **Escape hatch**: the wrapped story. Reads through here never touch the
656 /// journal (they can't mutate anyway).
657 #[must_use]
658 pub fn story(&self) -> &Story<R> {
659 &self.story
660 }
661
662 /// **Escape hatch**: mutable access to the wrapped story. Anything done
663 /// here **bypasses the journal** — the documented journal-bypass contract.
664 /// Use for foreign / shared flows (#200) whose externals never journal.
665 pub fn story_mut(&mut self) -> &mut Story<R> {
666 &mut self.story
667 }
668
669 /// Update the journal's embedded fast-restore checkpoint from the story's
670 /// current game state.
671 fn refresh_checkpoint(&mut self) {
672 self.journal.checkpoint = Some(self.story.save_state());
673 }
674
675 /// Record the implicit default `Start` if the session hasn't started.
676 fn ensure_started(&mut self) {
677 if !self.started {
678 self.started = true;
679 self.journal.push(JournalEvent::new(EventKind::Start {
680 path: None,
681 args: Vec::new(),
682 }));
683 }
684 }
685
686 // ── Stepping ─────────────────────────────────────────────────────
687
688 /// Advance one step with the default (fallback) handler, journaling any
689 /// inline-resolved externals. Surfaces a deferred external as
690 /// [`StepOutcome::AwaitingExternal`].
691 pub fn advance(&mut self) -> Result<StepOutcome, RuntimeError> {
692 self.advance_with(&FallbackHandler)
693 }
694
695 /// Advance one step with a custom handler, journaling any inline-resolved
696 /// externals it produces (the journaling-window gate: only this frame's
697 /// externals record).
698 pub fn advance_with(
699 &mut self,
700 handler: &dyn ExternalFnHandler,
701 ) -> Result<StepOutcome, RuntimeError> {
702 self.ensure_started();
703 let mut sink: Vec<(String, Vec<Value>, Value)> = Vec::new();
704 let outcome = {
705 let jh = JournalingHandler::new(handler, &mut sink);
706 self.story.advance_with(&jh)
707 };
708 for (name, args, result) in sink {
709 self.journal.push(JournalEvent::new(EventKind::External {
710 name,
711 args,
712 result,
713 }));
714 }
715 outcome
716 }
717
718 /// Advance until one line of content or a yield point, journaling externals.
719 pub fn continue_single(&mut self) -> Result<Step, RuntimeError> {
720 self.ensure_started();
721 let mut sink: Vec<(String, Vec<Value>, Value)> = Vec::new();
722 let outcome = {
723 let jh = JournalingHandler::new(&FallbackHandler, &mut sink);
724 self.story.continue_single_with(&jh)
725 };
726 for (name, args, result) in sink {
727 self.journal.push(JournalEvent::new(EventKind::External {
728 name,
729 args,
730 result,
731 }));
732 }
733 outcome
734 }
735
736 /// Advance to the next pause, journaling externals. The last line is always
737 /// terminal (`Done` / `Choices` / `End`).
738 pub fn continue_to_pause(&mut self) -> Result<Vec<Step>, RuntimeError> {
739 self.ensure_started();
740 let mut sink: Vec<(String, Vec<Value>, Value)> = Vec::new();
741 let outcome = {
742 let jh = JournalingHandler::new(&FallbackHandler, &mut sink);
743 self.story.continue_maximally_with(&jh)
744 };
745 for (name, args, result) in sink {
746 self.journal.push(JournalEvent::new(EventKind::External {
747 name,
748 args,
749 result,
750 }));
751 }
752 outcome
753 }
754
755 /// Select a choice, journaling the `Choice` event (with its advisory label).
756 pub fn choose(&mut self, index: usize) -> Result<(), RuntimeError> {
757 self.ensure_started();
758 let label = self
759 .story
760 .pending_choices()
761 .into_iter()
762 .find(|c| c.index == index)
763 .map(|c| c.text);
764 self.story.choose(index)?;
765 #[expect(clippy::cast_possible_truncation, reason = "choice indices are small")]
766 self.journal.push(JournalEvent::new(EventKind::Choice {
767 index: index as u32,
768 label,
769 }));
770 Ok(())
771 }
772
773 /// Resolve a deferred external (out-of-band, [`ExternalResult::Pending`]),
774 /// journaling it as an `External` event. This is the journaling-window gate
775 /// in action: the session records here because it is the session's own
776 /// pause that is being resolved.
777 pub fn resolve_external(&mut self, value: Value) {
778 let name = self
779 .story
780 .pending_external_name()
781 .map(str::to_owned)
782 .unwrap_or_default();
783 let args = self.story.pending_external_args().to_vec();
784 self.journal.push(JournalEvent::new(EventKind::External {
785 name,
786 args,
787 result: value.clone(),
788 }));
789 self.story.resolve_external(value);
790 }
791
792 /// Whether the session is parked on a deferred external.
793 #[must_use]
794 pub fn has_pending_external(&self) -> bool {
795 self.story.has_pending_external()
796 }
797
798 // ── Turn-boundary mutations ──────────────────────────────────────
799
800 /// Set a global variable. **Turn-boundary only**: rejected mid-turn.
801 ///
802 /// # Errors
803 /// [`SessionError::MutationMidTurn`] if the session is mid-turn (status
804 /// `Active`).
805 pub fn set_var(&mut self, name: &str, value: Value) -> Result<bool, SessionError> {
806 self.require_turn_boundary("set_var")?;
807 let applied = self.story.set_variable(name, value.clone());
808 if applied {
809 self.journal.push(JournalEvent::new(EventKind::SetVar {
810 name: name.to_owned(),
811 value,
812 }));
813 }
814 Ok(applied)
815 }
816
817 /// Move the play head to a path (ink `ChoosePathString`). **Turn-boundary
818 /// only**: rejected mid-turn.
819 ///
820 /// # Errors
821 /// [`SessionError::MutationMidTurn`] mid-turn; wrapped [`RuntimeError`]s
822 /// from the jump.
823 pub fn go_to_path(&mut self, path: &str, args: &[Value]) -> Result<(), SessionError> {
824 self.require_turn_boundary("go_to_path")?;
825 self.ensure_started();
826 if args.is_empty() {
827 self.story.choose_path_string(path)?;
828 } else {
829 self.story.choose_path_string_with_args(path, args)?;
830 }
831 self.journal.push(JournalEvent::new(EventKind::GoToPath {
832 path: path.to_owned(),
833 args: args.to_vec(),
834 }));
835 Ok(())
836 }
837
838 /// Load a durable [`SaveState`]. **Turn-boundary only**: rejected mid-turn.
839 ///
840 /// # Errors
841 /// [`SessionError::MutationMidTurn`] mid-turn.
842 pub fn load_state(&mut self, state: &SaveState) -> Result<(), SessionError> {
843 self.require_turn_boundary("load_state")?;
844 self.story.load_state(state);
845 self.journal.push(JournalEvent::new(EventKind::LoadState {
846 state: state.clone(),
847 }));
848 Ok(())
849 }
850
851 /// Capture the current durable game state (does not journal).
852 #[must_use]
853 pub fn save_state(&self) -> SaveState {
854 self.story.save_state()
855 }
856
857 /// Evaluate an ink function from engine code, journaling a `Call` event.
858 /// The function's own externals resolve through the isolated (non-journaling)
859 /// [`Story::call_function`] handler path — only the top-level call journals.
860 ///
861 /// # Errors
862 /// Wrapped [`RuntimeError`]s from evaluation.
863 pub fn call_function(
864 &mut self,
865 name: &str,
866 args: &[Value],
867 handler: &dyn ExternalFnHandler,
868 ) -> Result<Value, RuntimeError> {
869 // Note: `story.call_function` runs isolated — its externals do NOT go
870 // through the journaling handler, so they never enter the journal.
871 let result = self.story.call_function(name, args, handler)?;
872 self.journal.push(JournalEvent::new(EventKind::Call {
873 name: name.to_owned(),
874 args: args.to_vec(),
875 }));
876 Ok(result)
877 }
878
879 fn require_turn_boundary(&self, op: &'static str) -> Result<(), SessionError> {
880 // Mid-turn == the story is Active with more content pending. A fresh,
881 // not-yet-started session is at a boundary (allowed).
882 if self.started && self.story.status_is_active() {
883 return Err(SessionError::MutationMidTurn { op });
884 }
885 Ok(())
886 }
887
888 // ── Snapshot / diff ──────────────────────────────────────────────
889
890 /// A typed snapshot of the current game state (globals with list
891 /// membership, turn counts, callstack summary). See [`StateSnapshot`].
892 #[must_use]
893 pub fn snapshot(&self) -> StateSnapshot {
894 self.story.state_snapshot()
895 }
896
897 /// Pure diff of two snapshots (convenience; see the free [`diff`] fn).
898 #[must_use]
899 pub fn diff(a: &StateSnapshot, b: &StateSnapshot) -> StateDiff {
900 diff(a, b)
901 }
902}
903
904impl<R: StoryRng> StorySession<R> {
905 // ── Replay / restore ─────────────────────────────────────────────
906
907 /// Fast-restore: apply a journal's embedded [`checkpoint`](SessionJournal::checkpoint)
908 /// and skip replay when the program checksum matches; otherwise fall back to
909 /// a full [`replay`](Self::replay).
910 ///
911 /// Returns the constructed session and the [`ReplayOutcome`]. On a checksum
912 /// match with a present checkpoint the outcome is
913 /// [`ReplayOutcome::Replayed`] with no warnings (no stepping occurred).
914 ///
915 /// # Errors
916 /// [`SessionError::ChecksumMismatch`] only if the checksum differs *and* no
917 /// checkpoint is present to restore from (nothing safe to do).
918 pub fn restore(
919 story: Story<R>,
920 journal: SessionJournal,
921 ) -> Result<(Self, ReplayOutcome), SessionError> {
922 let program_checksum = story.program().source_checksum();
923 if program_checksum == journal.program_checksum {
924 if let Some(checkpoint) = journal.checkpoint.clone() {
925 let mut session = Self {
926 story,
927 journal,
928 started: true,
929 pending_replay: None,
930 };
931 session.story.load_state(&checkpoint);
932 return Ok((
933 session,
934 ReplayOutcome::Replayed {
935 warnings: Vec::new(),
936 },
937 ));
938 }
939 } else if journal.checkpoint.is_none() {
940 return Err(SessionError::ChecksumMismatch {
941 journal: journal.program_checksum,
942 program: program_checksum,
943 });
944 }
945 // Fall back to full replay (recompiled program or no checkpoint).
946 Ok(Self::replay(
947 story,
948 &journal,
949 ExternalReplayMode::Recorded,
950 None,
951 ))
952 }
953
954 /// Replay a journal against `story` from a fresh start. Consumes the journal
955 /// prefix event-by-event; on divergence, truncates the journal at that point
956 /// and parks at the reached position.
957 ///
958 /// The session **re-records** as it replays, rebuilding its own journal. In
959 /// [`ExternalReplayMode::Recorded`], the rebuilt prefix is the **source**
960 /// prefix: source `External` events are re-pushed verbatim, including any
961 /// the re-run did not actually consume (a recorded-mode mismatch falls back
962 /// to the ink fallback body rather than diverging) — the truncated prefix
963 /// is the source's record, not a re-observed trace. In
964 /// [`ExternalReplayMode::Live`] the rebuilt journal *is* a re-observed
965 /// trace: live results are journaled as they resolve and source `External`
966 /// events are not copied.
967 ///
968 /// `mode` selects recorded (journal-served) vs live externals. Live replay
969 /// hitting a deferred external parks with [`FailReason::AwaitingExternal`],
970 /// **retaining the un-replayed tail**: resolve the external
971 /// ([`resolve_external`](Self::resolve_external)) and resume with
972 /// [`continue_replay`](Self::continue_replay), which picks up the remaining
973 /// recorded inputs from the park point.
974 #[must_use]
975 pub fn replay(
976 story: Story<R>,
977 journal: &SessionJournal,
978 mode: ExternalReplayMode,
979 live_handler: Option<&dyn ExternalFnHandler>,
980 ) -> (Self, ReplayOutcome) {
981 let mut session = Self {
982 story,
983 journal: SessionJournal::new(0, None),
984 started: false,
985 pending_replay: None,
986 };
987 // Rebuild an empty journal to re-record faithfully as we replay.
988 session.journal =
989 SessionJournal::new(session.story.program().source_checksum(), journal.seed);
990 // Queue of recorded externals for the recorded-mode handler.
991 let ext_queue = journal
992 .events
993 .iter()
994 .filter_map(|ev| match &ev.kind {
995 EventKind::External { name, args, result } => {
996 Some((name.clone(), args.clone(), result.clone()))
997 }
998 _ => None,
999 })
1000 .collect();
1001 let state = PendingReplay {
1002 remaining: journal.events.iter().cloned().enumerate().collect(),
1003 ext_queue,
1004 mode,
1005 warnings: Vec::new(),
1006 source_checkpoint: journal.checkpoint.clone(),
1007 total_events: journal.events.len(),
1008 };
1009 let outcome = session.drive_replay(state, live_handler);
1010 (session, outcome)
1011 }
1012
1013 /// Whether a parked replay tail is pending (a live replay hit a deferred
1014 /// external). Resolve it and call [`continue_replay`](Self::continue_replay).
1015 #[must_use]
1016 pub fn has_pending_replay(&self) -> bool {
1017 self.pending_replay.is_some()
1018 }
1019
1020 /// Resume a replay parked on a deferred external
1021 /// ([`FailReason::AwaitingExternal`]). Resolve the pending external first
1022 /// (via [`resolve_external`](Self::resolve_external)), then call this: the
1023 /// session resumes consuming the retained journal tail from the park
1024 /// point — it can complete ([`ReplayOutcome::Replayed`] with all warnings
1025 /// accumulated across parks), park again, diverge later, or fail.
1026 ///
1027 /// With **no pending replay tail** this keeps the advance-only behavior:
1028 /// it steps the live story to its next pause with `live_handler`
1029 /// (journaling externals as in normal play) and returns `Replayed` on
1030 /// reaching the pause, or `Failed` if it parks or errors (`at_event` is
1031 /// then the rebuilt journal's current length, where the next event would
1032 /// land).
1033 pub fn continue_replay(
1034 &mut self,
1035 live_handler: Option<&dyn ExternalFnHandler>,
1036 ) -> ReplayOutcome {
1037 if let Some(state) = self.pending_replay.take() {
1038 return self.drive_replay(state, live_handler);
1039 }
1040 // Advance-only: no recorded inputs left to consume.
1041 let handler = live_handler.unwrap_or(&FallbackHandler);
1042 let mut steps = 0usize;
1043 loop {
1044 if steps >= Self::REPLAY_STEP_BUDGET {
1045 self.journal.truncated = true;
1046 return ReplayOutcome::Failed {
1047 at_event: self.journal.len(),
1048 reason: FailReason::Budget,
1049 };
1050 }
1051 steps += 1;
1052 match self.advance_with(handler) {
1053 Ok(StepOutcome::Step(step)) if step.is_terminal() => {
1054 return ReplayOutcome::Replayed {
1055 warnings: Vec::new(),
1056 };
1057 }
1058 Ok(StepOutcome::Step(_)) => {}
1059 Ok(StepOutcome::AwaitingExternal) => {
1060 let name = self
1061 .story
1062 .pending_external_name()
1063 .map(str::to_owned)
1064 .unwrap_or_default();
1065 return ReplayOutcome::Failed {
1066 at_event: self.journal.len(),
1067 reason: FailReason::AwaitingExternal { name },
1068 };
1069 }
1070 Err(e) => {
1071 self.journal.truncated = true;
1072 return ReplayOutcome::Failed {
1073 at_event: self.journal.len(),
1074 reason: runtime_fail(e),
1075 };
1076 }
1077 }
1078 }
1079 }
1080
1081 /// Maximum `advance` calls per replay stepping burst (unbounded-growth /
1082 /// no-hang guard).
1083 const REPLAY_STEP_BUDGET: usize = 100_000;
1084
1085 /// Drive (or resume) a replay from its cursor `state`. On a resumable park
1086 /// ([`FailReason::AwaitingExternal`]) the state is retained in
1087 /// [`pending_replay`](Self::pending_replay); on divergence or terminal
1088 /// failure it is dropped (the journal is truncated at that point).
1089 fn drive_replay(
1090 &mut self,
1091 mut state: PendingReplay,
1092 live_handler: Option<&dyn ExternalFnHandler>,
1093 ) -> ReplayOutcome {
1094 loop {
1095 // All recorded inputs consumed: final trailing step (a story with
1096 // no choices, or content after the last input, only advances
1097 // here), then complete.
1098 let Some((at, _)) = state.remaining.front().cloned() else {
1099 if self.started {
1100 let at = state.total_events.saturating_sub(1);
1101 if let Err(park) =
1102 self.replay_step_to_pause(state.mode, live_handler, &mut state.ext_queue)
1103 {
1104 return self.park_or_fail(state, at, park);
1105 }
1106 }
1107 // Carry over the terminal checkpoint if the source had one.
1108 self.journal.checkpoint.clone_from(&state.source_checkpoint);
1109 return ReplayOutcome::Replayed {
1110 warnings: state.warnings,
1111 };
1112 };
1113
1114 // A Choice consumes a pause: step to it first (this is also where
1115 // a resumed drive picks up after its external was resolved).
1116 let next_is_choice = matches!(
1117 state.remaining.front().map(|(_, ev)| &ev.kind),
1118 Some(EventKind::Choice { .. })
1119 );
1120 if next_is_choice
1121 && let Err(park) =
1122 self.replay_step_to_pause(state.mode, live_handler, &mut state.ext_queue)
1123 {
1124 return self.park_or_fail(state, at, park);
1125 }
1126
1127 let Some((i, ev)) = state.remaining.pop_front() else {
1128 // Unreachable: `front` was `Some` above.
1129 continue;
1130 };
1131 match self.replay_apply(i, &ev, state.mode, &mut state.warnings) {
1132 Ok(()) => {}
1133 Err(outcome) => return outcome,
1134 }
1135 }
1136 }
1137
1138 /// Park (retaining `state` for [`continue_replay`](Self::continue_replay))
1139 /// or fail terminally (truncating the rebuilt journal).
1140 fn park_or_fail(&mut self, state: PendingReplay, at: usize, park: StepPark) -> ReplayOutcome {
1141 match park {
1142 StepPark::Awaiting { name } => {
1143 self.pending_replay = Some(state);
1144 ReplayOutcome::Failed {
1145 at_event: at,
1146 reason: FailReason::AwaitingExternal { name },
1147 }
1148 }
1149 StepPark::Fail(reason) => {
1150 self.journal.truncated = true;
1151 ReplayOutcome::Failed {
1152 at_event: at,
1153 reason,
1154 }
1155 }
1156 }
1157 }
1158
1159 /// Apply one recorded input during replay. `Err` is the terminal
1160 /// [`ReplayOutcome`] (divergence or failure).
1161 fn replay_apply(
1162 &mut self,
1163 i: usize,
1164 ev: &JournalEvent,
1165 mode: ExternalReplayMode,
1166 warnings: &mut Vec<ReplayWarning>,
1167 ) -> Result<(), ReplayOutcome> {
1168 match &ev.kind {
1169 EventKind::Start { path, args } => {
1170 if self.started {
1171 return Err(self.diverge_at(i, ev, DivergenceFound::UnexpectedEvent));
1172 }
1173 self.started = true;
1174 self.journal.push(JournalEvent::new(EventKind::Start {
1175 path: path.clone(),
1176 args: args.clone(),
1177 }));
1178 if let Some(p) = path {
1179 let jump = if args.is_empty() {
1180 self.story.choose_path_string(p)
1181 } else {
1182 self.story.choose_path_string_with_args(p, args)
1183 };
1184 if jump.is_err() {
1185 return Err(self.diverge_at(
1186 i,
1187 ev,
1188 DivergenceFound::UnknownPath { path: p.clone() },
1189 ));
1190 }
1191 }
1192 }
1193 EventKind::External { .. } => {
1194 // Recorded mode: re-push the source event verbatim — the
1195 // rebuilt prefix is the SOURCE prefix, not a re-observed trace
1196 // (a mismatched entry falls back rather than diverging, so the
1197 // re-run may not have consumed it). Live mode journals actual
1198 // results during stepping instead, so nothing to copy here.
1199 if mode == ExternalReplayMode::Recorded {
1200 self.journal.push(ev.clone());
1201 }
1202 }
1203 EventKind::Choice { index, label } => {
1204 self.replay_choice(i, *index, label.as_ref(), ev, warnings)?;
1205 }
1206 EventKind::SetVar { name, value } => {
1207 self.story.set_variable(name, value.clone());
1208 self.journal.push(ev.clone());
1209 }
1210 EventKind::GoToPath { path, args } => {
1211 let jump = if args.is_empty() {
1212 self.story.choose_path_string(path)
1213 } else {
1214 self.story.choose_path_string_with_args(path, args)
1215 };
1216 if jump.is_err() {
1217 return Err(self.diverge_at(
1218 i,
1219 ev,
1220 DivergenceFound::UnknownPath { path: path.clone() },
1221 ));
1222 }
1223 self.journal.push(ev.clone());
1224 }
1225 EventKind::LoadState { state } => {
1226 self.story.load_state(state);
1227 self.journal.push(ev.clone());
1228 }
1229 EventKind::Call { name, args } => {
1230 // Journaled but isolated: re-invoke through the fallback
1231 // handler (recorded externals for a Call aren't separately
1232 // journaled; a live handler would be host-supplied). We use
1233 // the fallback so replay never blocks.
1234 let _ = self.story.call_function(name, args, &FallbackHandler);
1235 self.journal.push(ev.clone());
1236 }
1237 }
1238 Ok(())
1239 }
1240
1241 /// Replay one `Choice` event (the driver has already stepped to the choice
1242 /// pause): range-check the recorded index against what the current program
1243 /// presents, emit a soft label-drift warning, then select. Returns
1244 /// `Err(outcome)` on divergence/failure.
1245 fn replay_choice(
1246 &mut self,
1247 at: usize,
1248 index: u32,
1249 label: Option<&String>,
1250 ev: &JournalEvent,
1251 warnings: &mut Vec<ReplayWarning>,
1252 ) -> Result<(), ReplayOutcome> {
1253 if !self.story.status_is_waiting_for_choice() {
1254 return Err(self.diverge_at(at, ev, DivergenceFound::NotWaitingForChoice));
1255 }
1256 let presented = self.story.pending_choices();
1257 let available = presented.len();
1258 let Some(current) = presented.iter().find(|c| c.index == index as usize) else {
1259 return Err(self.diverge_at(
1260 at,
1261 ev,
1262 DivergenceFound::ChoiceIndexOutOfRange { index, available },
1263 ));
1264 };
1265 // Label-drift soft warning (matching index, different text).
1266 if let Some(recorded) = label
1267 && recorded != ¤t.text
1268 {
1269 warnings.push(ReplayWarning::ChoiceLabelDrift {
1270 at_event: at,
1271 index,
1272 recorded: recorded.clone(),
1273 found: current.text.clone(),
1274 });
1275 }
1276 if let Err(e) = self.story.choose(index as usize) {
1277 self.journal.truncated = true;
1278 return Err(ReplayOutcome::Failed {
1279 at_event: at,
1280 reason: runtime_fail(e),
1281 });
1282 }
1283 self.journal.push(ev.clone());
1284 Ok(())
1285 }
1286
1287 /// Step to the next pause during replay, serving externals per `mode`. In
1288 /// Live mode, inline-resolved results are journaled as they happen (the
1289 /// rebuilt journal is a re-observed trace). Returns `Err(park)` when a
1290 /// deferred external pauses the flow (resumable) or on a terminal failure.
1291 fn replay_step_to_pause(
1292 &mut self,
1293 mode: ExternalReplayMode,
1294 live_handler: Option<&dyn ExternalFnHandler>,
1295 ext_queue: &mut VecDeque<(String, Vec<Value>, Value)>,
1296 ) -> Result<(), StepPark> {
1297 let mut steps = 0usize;
1298 loop {
1299 if steps >= Self::REPLAY_STEP_BUDGET {
1300 return Err(StepPark::Fail(FailReason::Budget));
1301 }
1302 steps += 1;
1303 let outcome = match mode {
1304 ExternalReplayMode::Recorded => {
1305 let h = RecordedReplayHandler {
1306 queue: RefCell::new(ext_queue),
1307 };
1308 self.story.advance_with(&h)
1309 }
1310 ExternalReplayMode::Live => {
1311 // Journal live results where the VM receives them, so the
1312 // rebuilt journal reflects what actually fed this run.
1313 let mut sink: Vec<(String, Vec<Value>, Value)> = Vec::new();
1314 let outcome = {
1315 let h = live_handler.unwrap_or(&FallbackHandler);
1316 let jh = JournalingHandler::new(h, &mut sink);
1317 self.story.advance_with(&jh)
1318 };
1319 for (name, args, result) in sink {
1320 self.journal.push(JournalEvent::new(EventKind::External {
1321 name,
1322 args,
1323 result,
1324 }));
1325 }
1326 outcome
1327 }
1328 };
1329 match outcome {
1330 Ok(StepOutcome::Step(step)) => {
1331 if step.is_terminal() {
1332 return Ok(());
1333 }
1334 }
1335 Ok(StepOutcome::AwaitingExternal) => {
1336 let name = self
1337 .story
1338 .pending_external_name()
1339 .map(str::to_owned)
1340 .unwrap_or_default();
1341 return Err(StepPark::Awaiting { name });
1342 }
1343 Err(e) => return Err(StepPark::Fail(runtime_fail(e))),
1344 }
1345 }
1346 }
1347
1348 /// Truncate the rebuilt journal at the divergence point and build a
1349 /// `Diverged` outcome. The rebuilt journal keeps the **source prefix** as
1350 /// re-pushed so far (in recorded mode this includes source `External`
1351 /// events verbatim, even ones the re-run fell back on instead of
1352 /// consuming — see [`replay`](Self::replay)); the checkpoint is cleared
1353 /// because it described the source's terminal state, not this park point.
1354 fn diverge_at(
1355 &mut self,
1356 at: usize,
1357 expected: &JournalEvent,
1358 found: DivergenceFound,
1359 ) -> ReplayOutcome {
1360 self.journal.truncated = true;
1361 self.journal.checkpoint = None;
1362 ReplayOutcome::Diverged {
1363 at_event: at,
1364 expected: Box::new(expected.clone()),
1365 found,
1366 }
1367 }
1368}
1369
1370fn runtime_fail(e: RuntimeError) -> FailReason {
1371 match e {
1372 RuntimeError::StepLimitExceeded(_) | RuntimeError::LineLimitExceeded(_) => {
1373 FailReason::Budget
1374 }
1375 other => FailReason::RuntimeError {
1376 message: other.to_string(),
1377 },
1378 }
1379}