Enforce session replay invariants

src/session.rs

@@ -47,6 +47,7 @@ struct PolicyAppliedPayload {
 #[serde(deny_unknown_fields)]
 struct TransitionStartedPayload {
     schema_version: u16,
+    commitment_id: String,
     reason: String,
     return_target: String,
     expected_duration_secs: u64,
@@ -231,6 +232,7 @@ impl SessionController {
                 Some(next_commitment.id.clone()),
                 serde_json::to_value(TransitionStartedPayload {
                     schema_version: POLICY_SCHEMA_VERSION,
+                    commitment_id: next_commitment.id.clone(),
                     reason,
                     return_target,
                     expected_duration_secs: expected_duration.as_secs(),
@@ -294,6 +296,12 @@ fn hydrate_session(
                     parse_payload(&record, "commitment created")?;
                 ensure_schema_version(payload.schema_version, record.sequence)?;
                 ensure_record_commitment(&record, &payload.commitment_id)?;
+                if active_commitment.is_some() || active_policy.is_some() {
+                    return Err(anyhow!(
+                        "commitment created while active commitment already exists at sequence {}",
+                        record.sequence
+                    ));
+                }
                 if record.runtime_state != RuntimeState::Active {
                     return Err(anyhow!(
                         "commitment activation runtime mismatch at sequence {}",
@@ -390,6 +398,13 @@ fn hydrate_session(
                         record.sequence
                     )
                 })?;
+                ensure_record_commitment(&record, &payload.commitment_id)?;
+                if payload.commitment_id != active.id {
+                    return Err(anyhow!(
+                        "transition commitment mismatch at sequence {}",
+                        record.sequence
+                    ));
+                }
                 if payload.runtime_from != runtime_state || payload.commitment_from != active.state
                 {
                     return Err(anyhow!(
@@ -431,6 +446,13 @@ fn hydrate_session(
             "session replay ended with active commitment but no active policy"
         ));
     }
+    if let (Some(commitment), Some(policy)) = (&active_commitment, &active_policy) {
+        if policy.commitment_id != commitment.id {
+            return Err(anyhow!(
+                "session replay ended with commitment/policy mismatch"
+            ));
+        }
+    }
 
     Ok((runtime_state, active_commitment, active_policy))
 }
@@ -471,7 +493,7 @@ fn ensure_record_commitment(record: &EventRecord, commitment_id: &str) -> Result
 
 #[cfg(test)]
 mod tests {
-    use super::{CommitmentCreatedPayload, PolicyAppliedPayload};
+    use super::{CommitmentCreatedPayload, PolicyAppliedPayload, TransitionStartedPayload};
     use crate::domain::{
         AllowedContext, CommitmentSource, CommitmentState, EnforcementLevel, RuntimeState,
         POLICY_SCHEMA_VERSION,
@@ -527,6 +549,21 @@ mod tests {
         .unwrap()
     }
 
+    fn transition_started_payload() -> serde_json::Value {
+        serde_json::to_value(TransitionStartedPayload {
+            schema_version: POLICY_SCHEMA_VERSION,
+            commitment_id: "commitment-test".to_string(),
+            reason: "Need to inspect failures".to_string(),
+            return_target: "Return to session controller".to_string(),
+            expected_duration_secs: 300,
+            runtime_from: RuntimeState::Active,
+            runtime_to: RuntimeState::Transition,
+            commitment_from: CommitmentState::Active,
+            commitment_to: CommitmentState::Paused,
+        })
+        .unwrap()
+    }
+
     fn error_chain_contains(err: &anyhow::Error, expected: &str) -> bool {
         err.chain()
             .any(|cause| cause.to_string().contains(expected))
@@ -712,6 +749,131 @@ mod tests {
         fs::remove_file(path).unwrap();
     }
 
+    #[test]
+    fn rejects_stray_commitment_created_after_active_session_on_replay() {
+        let path = temp_log_path("stray-commitment");
+        {
+            let mut session = SessionController::new(&path).unwrap();
+            session.enter_planning().unwrap();
+            session
+                .start_manual_commitment(
+                    "Implement session controller",
+                    "session tests pass",
+                    Duration::from_secs(1500),
+                )
+                .unwrap();
+
+            let mut log = EventLog::open(&path).unwrap();
+            log.append(
+                EventType::CommitmentCreated,
+                RuntimeState::Active,
+                Some("commitment-stray".to_string()),
+                commitment_payload("commitment-stray", CommitmentState::Active),
+            )
+            .unwrap();
+        }
+
+        let err = SessionController::new(&path)
+            .expect_err("stray commitment must not replace active commitment");
+
+        assert!(error_chain_contains(&err, "active commitment"));
+        fs::remove_file(path).unwrap();
+    }
+
+    #[test]
+    fn rejects_transition_started_with_wrong_envelope_commitment_on_replay() {
+        let path = temp_log_path("transition-wrong-commitment");
+        let commitment_id = "commitment-test";
+        {
+            let mut log = EventLog::open(&path).unwrap();
+            log.append_batch([
+                PendingEventRecord {
+                    event_type: EventType::RuntimeTransition,
+                    runtime_state: RuntimeState::Planning,
+                    commitment_id: None,
+                    payload_json: json!({
+                        "schema_version": POLICY_SCHEMA_VERSION,
+                        "action": "enter_planning",
+                        "from": "locked",
+                        "to": "planning",
+                    }),
+                },
+                PendingEventRecord {
+                    event_type: EventType::CommitmentCreated,
+                    runtime_state: RuntimeState::Active,
+                    commitment_id: Some(commitment_id.to_string()),
+                    payload_json: commitment_payload(commitment_id, CommitmentState::Active),
+                },
+                PendingEventRecord {
+                    event_type: EventType::PolicyApplied,
+                    runtime_state: RuntimeState::Active,
+                    commitment_id: Some(commitment_id.to_string()),
+                    payload_json: policy_payload(commitment_id),
+                },
+                PendingEventRecord {
+                    event_type: EventType::TransitionStarted,
+                    runtime_state: RuntimeState::Transition,
+                    commitment_id: Some("commitment-other".to_string()),
+                    payload_json: transition_started_payload(),
+                },
+            ])
+            .unwrap();
+        }
+
+        let err = SessionController::new(&path)
+            .expect_err("transition envelope commitment mismatch must be rejected");
+
+        assert!(error_chain_contains(&err, "commitment id mismatch"));
+        fs::remove_file(path).unwrap();
+    }
+
+    #[test]
+    fn rejects_transition_started_without_envelope_commitment_on_replay() {
+        let path = temp_log_path("transition-missing-commitment");
+        let commitment_id = "commitment-test";
+        {
+            let mut log = EventLog::open(&path).unwrap();
+            log.append_batch([
+                PendingEventRecord {
+                    event_type: EventType::RuntimeTransition,
+                    runtime_state: RuntimeState::Planning,
+                    commitment_id: None,
+                    payload_json: json!({
+                        "schema_version": POLICY_SCHEMA_VERSION,
+                        "action": "enter_planning",
+                        "from": "locked",
+                        "to": "planning",
+                    }),
+                },
+                PendingEventRecord {
+                    event_type: EventType::CommitmentCreated,
+                    runtime_state: RuntimeState::Active,
+                    commitment_id: Some(commitment_id.to_string()),
+                    payload_json: commitment_payload(commitment_id, CommitmentState::Active),
+                },
+                PendingEventRecord {
+                    event_type: EventType::PolicyApplied,
+                    runtime_state: RuntimeState::Active,
+                    commitment_id: Some(commitment_id.to_string()),
+                    payload_json: policy_payload(commitment_id),
+                },
+                PendingEventRecord {
+                    event_type: EventType::TransitionStarted,
+                    runtime_state: RuntimeState::Transition,
+                    commitment_id: None,
+                    payload_json: transition_started_payload(),
+                },
+            ])
+            .unwrap();
+        }
+
+        let err = SessionController::new(&path)
+            .expect_err("transition without envelope commitment must be rejected");
+
+        assert!(error_chain_contains(&err, "commitment id mismatch"));
+        fs::remove_file(path).unwrap();
+    }
+
     #[test]
     fn transition_started_payload_has_stable_schema() {
         let path = temp_log_path("transition-payload");
@@ -740,6 +902,10 @@ mod tests {
             .unwrap();
 
         assert_eq!(transition.payload_json["schema_version"], 1);
+        assert_eq!(
+            transition.payload_json["commitment_id"],
+            transition.commitment_id.as_deref().unwrap()
+        );
         assert_eq!(
             transition.payload_json["reason"],
             "Need to inspect failures"