feat(runtime): persist atomic turn plans

crates/nana-runtime/Cargo.toml

@@ -7,7 +7,10 @@ license.workspace = true
 
 [dependencies]
 nana-domain.workspace = true
+nana-engine.workspace = true
+nana-store.workspace = true
 openlapp.workspace = true
+serde_json.workspace = true
 thiserror.workspace = true
 
 [lints]

crates/nana-runtime/src/lib.rs

@@ -1,8 +1,11 @@
 use std::collections::{BTreeSet, VecDeque};
 
 use nana_domain::{
+    ActionSuggestion, PlayerView, PresentationBeat, RuntimeState, StateDelta, StoryNode,
     TurnFailure, TurnFailureCode, TurnIntent, TurnRequest, TurnResult, WorldBookEntry,
 };
+use nana_engine::{ReduceError, apply_delta};
+use nana_store::{StoreError, StoryStore};
 use thiserror::Error;
 
 pub const LAPP_BASELINE_COMMIT: &str = "5ba3c659e1536ec4bee16340faca603940a5cb17";
@@ -20,6 +23,134 @@ pub trait TurnProvider {
     fn complete_turn(&mut self, request: &TurnRequest) -> Result<TurnResult, ProviderError>;
 }
 
+/// Non-view model output used by the persistent turn engine.
+///
+/// The provider can propose narrative beats and state changes, but it cannot
+/// construct the final [`PlayerView`]. That view is derived from committed state
+/// by a separate trusted projection boundary.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct TurnPlan {
+    pub committed_node_id: String,
+    pub beats: Vec<PresentationBeat>,
+    pub delta: StateDelta,
+    pub suggestions: Vec<ActionSuggestion>,
+}
+
+/// Produces the uncommitted model plan for a turn.
+pub trait TurnPlanProvider {
+    fn plan_turn(
+        &mut self,
+        request: &TurnRequest,
+        state: &RuntimeState,
+    ) -> Result<TurnPlan, ProviderError>;
+}
+
+/// Projects only already-committed state into the player-safe read model.
+///
+/// Projection is deliberately infallible: it is a pure, defensive operation
+/// that omits or degrades data it cannot safely represent. This prevents a
+/// post-commit error window where the branch advances but the turn reports a
+/// failure.
+pub trait TurnProjector {
+    fn project_committed_turn(
+        &mut self,
+        state: &RuntimeState,
+        node: &StoryNode,
+        suggestions: &[ActionSuggestion],
+    ) -> PlayerView;
+}
+
+/// Store-backed single-turn coordinator.
+///
+/// All fallible work is completed before the atomic append. Projection runs
+/// only after the store confirms the new branch head.
+pub struct TurnEngine<'store, Store, Provider, Projector> {
+    store: &'store Store,
+    provider: Provider,
+    projector: Projector,
+}
+
+impl<'store, Store, Provider, Projector> TurnEngine<'store, Store, Provider, Projector>
+where
+    Store: StoryStore,
+    Provider: TurnPlanProvider,
+    Projector: TurnProjector,
+{
+    #[must_use]
+    pub fn new(store: &'store Store, provider: Provider, projector: Projector) -> Self {
+        Self {
+            store,
+            provider,
+            projector,
+        }
+    }
+
+    /// Validate, plan, reduce, commit, then project one player turn.
+    pub fn submit_turn(&mut self, request: &TurnRequest) -> Result<TurnResult, TurnFailure> {
+        validate_turn_request(request)?;
+
+        let current = self
+            .store
+            .load_state(&request.story_id, &request.branch_id)
+            .map_err(map_store_error)?;
+        if current.current_node != request.expected_node_id {
+            return Err(stale_node());
+        }
+
+        let plan = self
+            .provider
+            .plan_turn(request, &current)
+            .map_err(|_| provider_unavailable())?;
+        validate_turn_plan(request, &plan)?;
+
+        let mut committed = apply_delta(&current, &plan.delta).map_err(map_reduce_error)?;
+        committed.current_node.clone_from(&plan.committed_node_id);
+        committed.current_branch.clone_from(&request.branch_id);
+
+        let state_hash = hash_runtime_state(&committed)?;
+        let node = StoryNode {
+            id: plan.committed_node_id.clone(),
+            story_id: request.story_id.clone(),
+            branch_id: request.branch_id.clone(),
+            parent_id: Some(current.current_node),
+            action_id: request.action_id.clone(),
+            user_input: request.input.clone(),
+            beats: plan.beats,
+            delta: plan.delta,
+            state_hash,
+        };
+
+        self.store
+            .append_node(&node, &committed)
+            .map_err(map_store_error)?;
+
+        let mut player_view =
+            self.projector
+                .project_committed_turn(&committed, &node, &plan.suggestions);
+        // Identity comes from the committed state, never from projection input.
+        // Normalizing these fields keeps even a defensive fallback projector
+        // aligned with the commit it represents.
+        player_view.story_id.clone_from(&committed.story_id);
+        player_view.node_id.clone_from(&committed.current_node);
+        player_view.branch_id.clone_from(&committed.current_branch);
+
+        Ok(TurnResult {
+            committed_node_id: node.id,
+            player_view,
+        })
+    }
+
+    #[must_use]
+    pub fn provider(&self) -> &Provider {
+        &self.provider
+    }
+
+    #[must_use]
+    pub fn projector(&self) -> &Projector {
+        &self.projector
+    }
+}
+
 #[derive(Debug)]
 pub struct FakeProvider {
     responses: VecDeque<TurnResult>,
@@ -175,6 +306,43 @@ fn validate_turn_result(request: &TurnRequest, result: &TurnResult) -> Result<()
     Ok(())
 }
 
+fn validate_turn_plan(request: &TurnRequest, plan: &TurnPlan) -> Result<(), TurnFailure> {
+    if plan.committed_node_id.trim().is_empty() {
+        return Err(invalid_model_output("committed node id is empty"));
+    }
+    if plan.committed_node_id == request.expected_node_id {
+        return Err(invalid_model_output(
+            "provider returned the uncommitted expected node",
+        ));
+    }
+    Ok(())
+}
+
+fn hash_runtime_state(state: &RuntimeState) -> Result<String, TurnFailure> {
+    serde_json::to_vec(state)
+        .map(|bytes| nana_domain::stable_json_hash(&bytes))
+        .map_err(|_| internal_failure("turn state could not be prepared"))
+}
+
+fn map_reduce_error(_error: ReduceError) -> TurnFailure {
+    invalid_model_output("turn plan could not be applied")
+}
+
+fn map_store_error(error: StoreError) -> TurnFailure {
+    match error {
+        StoreError::StaleBranchHead { .. } => stale_node(),
+        StoreError::StoryNotFound(_) | StoreError::BranchNotFound { .. } => TurnFailure {
+            code: TurnFailureCode::InvalidInput,
+            message: "story or branch is unavailable".into(),
+            retryable: false,
+        },
+        StoreError::NodeAlreadyExists(_)
+        | StoreError::ParentNotFound(_)
+        | StoreError::StateMismatch(_)
+        | StoreError::Poisoned => internal_failure("turn could not be committed"),
+    }
+}
+
 fn validate_required_text(field: &str, value: &str) -> Result<(), TurnFailure> {
     if value.trim().is_empty() {
         Err(invalid_input(format!("{field} must not be empty")))
@@ -203,6 +371,22 @@ fn provider_unavailable() -> TurnFailure {
     provider_failure("turn provider is unavailable")
 }
 
+fn stale_node() -> TurnFailure {
+    TurnFailure {
+        code: TurnFailureCode::StaleNode,
+        message: "story branch changed; refresh and retry".into(),
+        retryable: true,
+    }
+}
+
+fn internal_failure(message: impl Into<String>) -> TurnFailure {
+    TurnFailure {
+        code: TurnFailureCode::Internal,
+        message: message.into(),
+        retryable: true,
+    }
+}
+
 fn normalize_trigger(value: &str) -> String {
     value.trim().to_lowercase()
 }
@@ -514,3 +698,344 @@ mod tests {
         assert_eq!(selected, repeated);
     }
 }
+
+#[cfg(test)]
+mod persistent_turn_tests {
+    use std::collections::{BTreeMap, VecDeque};
+
+    use nana_domain::{
+        ActionSuggestion, BeatKind, PlayerView, PresentationBeat, RelationshipAdjustment,
+        RelationshipBand, RelationshipDimension, RelationshipView, RuntimeState, StateDelta,
+        StateOp, StoryNode, TurnFailureCode, TurnIntent, TurnRequest,
+    };
+    use nana_store::{InMemoryStoryStore, StoryStore};
+
+    use super::{
+        ProviderError, TurnEngine, TurnPlan, TurnPlanProvider, TurnProjector, hash_runtime_state,
+    };
+
+    struct RecordingPlanProvider {
+        responses: VecDeque<Result<TurnPlan, ProviderError>>,
+        calls: usize,
+    }
+
+    impl RecordingPlanProvider {
+        fn new(response: Result<TurnPlan, ProviderError>) -> Self {
+            Self {
+                responses: VecDeque::from([response]),
+                calls: 0,
+            }
+        }
+    }
+
+    impl TurnPlanProvider for RecordingPlanProvider {
+        fn plan_turn(
+            &mut self,
+            _request: &TurnRequest,
+            _state: &RuntimeState,
+        ) -> Result<TurnPlan, ProviderError> {
+            self.calls += 1;
+            self.responses
+                .pop_front()
+                .unwrap_or(Err(ProviderError::FixtureExhausted))
+        }
+    }
+
+    struct RecordingProjector<'store> {
+        store: &'store InMemoryStoryStore,
+        calls: usize,
+    }
+
+    impl TurnProjector for RecordingProjector<'_> {
+        fn project_committed_turn(
+            &mut self,
+            state: &RuntimeState,
+            node: &StoryNode,
+            suggestions: &[ActionSuggestion],
+        ) -> PlayerView {
+            self.calls += 1;
+
+            let stored = self
+                .store
+                .load_state(&state.story_id, &state.current_branch)
+                .expect("projection must run after append");
+            assert_eq!(stored, *state);
+            assert_eq!(node.id, state.current_node);
+
+            PlayerView {
+                // Deliberately wrong: the engine normalizes commit identity.
+                story_id: "untrusted_story".into(),
+                node_id: "untrusted_node".into(),
+                branch_id: "untrusted_branch".into(),
+                scene_id: "station".into(),
+                scene_title: "Station".into(),
+                character_name: "Nana".into(),
+                beats: node.beats.clone(),
+                suggestions: suggestions.to_vec(),
+                inventory: Vec::new(),
+                knowledge: Vec::new(),
+                promises: Vec::new(),
+                relationship: RelationshipView {
+                    affinity: RelationshipBand::Warming,
+                    trust: RelationshipBand::Guarded,
+                    hope: RelationshipBand::Guarded,
+                    respect: RelationshipBand::Warming,
+                    intimacy: RelationshipBand::Guarded,
+                    attachment: RelationshipBand::Warming,
+                    updated_at_node: Some(node.id.clone()),
+                },
+                history: Vec::new(),
+                can_continue: true,
+            }
+        }
+    }
+
+    fn state(node: &str, branch: &str) -> RuntimeState {
+        RuntimeState {
+            story_id: "story_1".into(),
+            current_node: node.into(),
+            current_branch: branch.into(),
+            world_flags: BTreeMap::new(),
+            relationships: BTreeMap::new(),
+            relationship_states: Vec::new(),
+            promises: Vec::new(),
+            knowledge: Vec::new(),
+            items: Vec::new(),
+            clocks: Vec::new(),
+            checks: Vec::new(),
+        }
+    }
+
+    fn node(id: &str, parent_id: Option<&str>, branch: &str) -> StoryNode {
+        StoryNode {
+            id: id.into(),
+            story_id: "story_1".into(),
+            branch_id: branch.into(),
+            parent_id: parent_id.map(Into::into),
+            action_id: format!("action_{id}"),
+            user_input: String::new(),
+            beats: Vec::new(),
+            delta: StateDelta { ops: Vec::new() },
+            state_hash: format!("hash_{id}"),
+        }
+    }
+
+    fn seeded_store() -> InMemoryStoryStore {
+        let store = InMemoryStoryStore::new();
+        store
+            .append_node(
+                &node("node_1", None, "branch_main"),
+                &state("node_1", "branch_main"),
+            )
+            .expect("seed root");
+        store
+    }
+
+    fn request(expected_node_id: &str) -> TurnRequest {
+        TurnRequest {
+            story_id: "story_1".into(),
+            branch_id: "branch_main".into(),
+            expected_node_id: expected_node_id.into(),
+            action_id: "action_2".into(),
+            intent: TurnIntent::SpeakOrAct,
+            input: "I will return before dawn.".into(),
+        }
+    }
+
+    fn plan(node_id: &str, delta: StateDelta) -> TurnPlan {
+        TurnPlan {
+            committed_node_id: node_id.into(),
+            beats: vec![PresentationBeat {
+                id: "beat_1".into(),
+                kind: BeatKind::Dialogue,
+                speaker: Some("Nana".into()),
+                text: "Then I will wait.".into(),
+                visual: None,
+            }],
+            delta,
+            suggestions: vec![ActionSuggestion {
+                id: "suggestion_1".into(),
+                label: "Promise".into(),
+                draft: "I promise.".into(),
+            }],
+        }
+    }
+
+    fn projector(store: &InMemoryStoryStore) -> RecordingProjector<'_> {
+        RecordingProjector { store, calls: 0 }
+    }
+
+    #[test]
+    fn successful_turn_commits_before_projecting() {
+        let store = seeded_store();
+        let delta = StateDelta {
+            ops: vec![StateOp::SetWorldFlag {
+                key: "promise_spoken".into(),
+                value: true,
+            }],
+        };
+        let mut engine = TurnEngine::new(
+            &store,
+            RecordingPlanProvider::new(Ok(plan("node_2", delta))),
+            projector(&store),
+        );
+
+        let result = engine.submit_turn(&request("node_1")).expect("turn");
+
+        assert_eq!(result.committed_node_id, "node_2");
+        assert_eq!(result.player_view.story_id, "story_1");
+        assert_eq!(result.player_view.node_id, "node_2");
+        assert_eq!(result.player_view.branch_id, "branch_main");
+        assert_eq!(result.player_view.suggestions.len(), 1);
+        assert_eq!(engine.provider().calls, 1);
+        assert_eq!(engine.projector().calls, 1);
+
+        let committed = store
+            .load_state("story_1", "branch_main")
+            .expect("committed state");
+        assert_eq!(committed.current_node, "node_2");
+        assert_eq!(committed.world_flags.get("promise_spoken"), Some(&true));
+    }
+
+    #[test]
+    fn stale_request_does_not_call_provider_or_move_head() {
+        let store = seeded_store();
+        let mut engine = TurnEngine::new(
+            &store,
+            RecordingPlanProvider::new(Ok(plan(
+                "node_2",
+                StateDelta { ops: Vec::new() },
+            ))),
+            projector(&store),
+        );
+
+        let failure = engine
+            .submit_turn(&request("node_stale"))
+            .expect_err("stale request");
+
+        assert_eq!(failure.code, TurnFailureCode::StaleNode);
+        assert!(failure.retryable);
+        assert_eq!(engine.provider().calls, 0);
+        assert_eq!(engine.projector().calls, 0);
+        assert_eq!(
+            store.branch_head("story_1", "branch_main").expect("head"),
+            Some("node_1".into())
+        );
+    }
+
+    #[test]
+    fn reducer_failure_is_redacted_and_does_not_save() {
+        let store = seeded_store();
+        let delta = StateDelta {
+            ops: vec![StateOp::AdjustRelationship {
+                from: "nana".into(),
+                to: "player".into(),
+                adjustment: RelationshipAdjustment {
+                    dimension: RelationshipDimension::Trust,
+                    delta: 9,
+                    cause: "secret model reasoning".into(),
+                    judgment_rule: Some("secret.rule".into()),
+                },
+            }],
+        };
+        let mut engine = TurnEngine::new(
+            &store,
+            RecordingPlanProvider::new(Ok(plan("node_2", delta))),
+            projector(&store),
+        );
+
+        let failure = engine
+            .submit_turn(&request("node_1"))
+            .expect_err("invalid delta");
+
+        assert_eq!(failure.code, TurnFailureCode::InvalidModelOutput);
+        assert_eq!(failure.message, "turn plan could not be applied");
+        assert!(!failure.message.contains("secret"));
+        assert_eq!(engine.provider().calls, 1);
+        assert_eq!(engine.projector().calls, 0);
+        assert_eq!(
+            store.branch_head("story_1", "branch_main").expect("head"),
+            Some("node_1".into())
+        );
+    }
+
+    #[test]
+    fn provider_failure_is_redacted_and_does_not_save() {
+        let store = seeded_store();
+        let mut engine = TurnEngine::new(
+            &store,
+            RecordingPlanProvider::new(Err(ProviderError::Profile(
+                "secret upstream endpoint and token".into(),
+            ))),
+            projector(&store),
+        );
+
+        let failure = engine
+            .submit_turn(&request("node_1"))
+            .expect_err("provider failure");
+
+        assert_eq!(failure.code, TurnFailureCode::ProviderUnavailable);
+        assert_eq!(failure.message, "turn provider is unavailable");
+        assert!(!failure.message.contains("secret"));
+        assert_eq!(engine.projector().calls, 0);
+        assert_eq!(
+            store.branch_head("story_1", "branch_main").expect("head"),
+            Some("node_1".into())
+        );
+    }
+
+    #[test]
+    fn append_failure_does_not_project_or_move_requested_branch() {
+        let store = seeded_store();
+        store
+            .append_node(
+                &node("node_duplicate", Some("node_1"), "branch_other"),
+                &state("node_duplicate", "branch_other"),
+            )
+            .expect("seed duplicate id on another branch");
+        let mut engine = TurnEngine::new(
+            &store,
+            RecordingPlanProvider::new(Ok(plan(
+                "node_duplicate",
+                StateDelta { ops: Vec::new() },
+            ))),
+            projector(&store),
+        );
+
+        let failure = engine
+            .submit_turn(&request("node_1"))
+            .expect_err("duplicate node append");
+
+        assert_eq!(failure.code, TurnFailureCode::Internal);
+        assert_eq!(failure.message, "turn could not be committed");
+        assert_eq!(engine.projector().calls, 0);
+        assert_eq!(
+            store.branch_head("story_1", "branch_main").expect("head"),
+            Some("node_1".into())
+        );
+    }
+
+    #[test]
+    fn turn_plan_is_non_view_provider_output_and_state_hash_is_stable() {
+        fn provider_output(
+            provider: &mut impl TurnPlanProvider,
+            request: &TurnRequest,
+            state: &RuntimeState,
+        ) -> TurnPlan {
+            provider.plan_turn(request, state).expect("turn plan")
+        }
+
+        let current = state("node_1", "branch_main");
+        let expected = hash_runtime_state(&current).expect("hash");
+        assert_eq!(hash_runtime_state(&current).expect("repeat hash"), expected);
+
+        let mut provider = RecordingPlanProvider::new(Ok(plan(
+            "node_2",
+            StateDelta { ops: Vec::new() },
+        )));
+        let output = provider_output(&mut provider, &request("node_1"), &current);
+        assert_eq!(output.committed_node_id, "node_2");
+        assert_eq!(output.beats.len(), 1);
+        assert_eq!(output.suggestions.len(), 1);
+    }
+}