From 999d99749a0d2c5f0deb825ab1b4ee79d6a4f17e Mon Sep 17 00:00:00 2001 From: Felix Martin Date: Mon, 25 May 2026 15:29:02 -0400 Subject: [PATCH] Enforce session replay invariants --- src/session.rs | 168 ++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 167 insertions(+), 1 deletion(-) diff --git a/src/session.rs b/src/session.rs index 70678e3..533090b 100644 --- a/src/session.rs +++ b/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"