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