From 2e58e3b269bf9e91f0818dffe756aaa11d14c0f9 Mon Sep 17 00:00:00 2001 From: Felix Martin Date: Mon, 25 May 2026 15:36:15 -0400 Subject: [PATCH] Keep transition policy consistent --- src/session.rs | 359 ++++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 311 insertions(+), 48 deletions(-) diff --git a/src/session.rs b/src/session.rs index 533090b..b9445a6 100644 --- a/src/session.rs +++ b/src/session.rs @@ -198,24 +198,14 @@ impl SessionController { return_target: impl Into, expected_duration: Duration, ) -> Result<()> { - let reason = reason.into().trim().to_string(); - if reason.is_empty() { - return Err(anyhow!("transition reason is required")); - } - - let return_target = return_target.into().trim().to_string(); - if return_target.is_empty() { - return Err(anyhow!("transition return target is required")); - } - - if expected_duration.is_zero() { - return Err(anyhow!("transition expected duration must be nonzero")); - } - let active_commitment = self .active_commitment .as_ref() .ok_or_else(|| anyhow!("active commitment is required to start transition"))?; + let active_policy = self + .active_policy + .as_ref() + .ok_or_else(|| anyhow!("active policy is required to start transition"))?; let composite = state_machine::start_transition(self.runtime_state, active_commitment.state.clone()) .context("start transition state change")?; @@ -225,28 +215,57 @@ impl SessionController { let mut next_commitment = active_commitment.clone(); next_commitment.state = next_commitment_state.clone(); + let transition_policy = PolicySnapshot::try_for_runtime( + next_commitment.id.clone(), + composite.runtime_state, + active_policy.enforcement_level, + active_policy.allowed_context.clone(), + ) + .context("create transition policy snapshot")?; + next_commitment.transition_policy_id = Some(transition_policy.id.clone()); + + let transition_payload = TransitionStartedPayload { + schema_version: POLICY_SCHEMA_VERSION, + commitment_id: next_commitment.id.clone(), + reason: reason.into().trim().to_string(), + return_target: return_target.into().trim().to_string(), + expected_duration_secs: expected_duration.as_secs(), + runtime_from: self.runtime_state, + runtime_to: composite.runtime_state, + commitment_from: previous_commitment_state, + commitment_to: next_commitment_state, + }; + validate_transition_started_payload(&transition_payload)?; + self.event_log - .append( - EventType::TransitionStarted, - composite.runtime_state, - 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(), - runtime_from: self.runtime_state, - runtime_to: composite.runtime_state, - commitment_from: previous_commitment_state, - commitment_to: next_commitment_state, - }) - .context("serialize transition started payload")?, - ) + .append_batch([ + PendingEventRecord { + event_type: EventType::TransitionStarted, + runtime_state: composite.runtime_state, + commitment_id: Some(next_commitment.id.clone()), + payload_json: serde_json::to_value(transition_payload) + .context("serialize transition started payload")?, + }, + PendingEventRecord { + event_type: EventType::PolicyApplied, + runtime_state: composite.runtime_state, + commitment_id: Some(next_commitment.id.clone()), + payload_json: serde_json::to_value(PolicyAppliedPayload { + schema_version: POLICY_SCHEMA_VERSION, + policy_id: transition_policy.id.clone(), + commitment_id: transition_policy.commitment_id.clone(), + runtime_state: transition_policy.runtime_state, + enforcement_level: transition_policy.enforcement_level, + allowed_context: transition_policy.allowed_context.clone(), + }) + .context("serialize transition policy payload")?, + }, + ]) .context("log transition start")?; self.runtime_state = composite.runtime_state; self.active_commitment = Some(next_commitment); + self.active_policy = Some(transition_policy); Ok(()) } } @@ -348,21 +367,29 @@ fn hydrate_session( record.sequence )); } - if commitment.state != CommitmentState::Active { - return Err(anyhow!( - "policy requires active commitment at sequence {}", - record.sequence - )); - } - let next = transition_runtime( - runtime_state, - RuntimeAction::Activate { - policy_accepted: true, - }, - ) - .with_context(|| { - format!("replay policy application at sequence {}", record.sequence) - })?; + let next = match commitment.state { + CommitmentState::Active => transition_runtime( + runtime_state, + RuntimeAction::Activate { + policy_accepted: true, + }, + ) + .with_context(|| { + format!("replay policy application at sequence {}", record.sequence) + })?, + CommitmentState::Paused + if runtime_state == RuntimeState::Transition + && payload.runtime_state == RuntimeState::Transition => + { + RuntimeState::Transition + } + _ => { + return Err(anyhow!( + "policy requires active or transition commitment at sequence {}", + record.sequence + )); + } + }; if payload.runtime_state != next || record.runtime_state != next { return Err(anyhow!( "policy runtime target mismatch at sequence {}", @@ -386,18 +413,35 @@ fn hydrate_session( .validate() .with_context(|| format!("validate policy at sequence {}", record.sequence))?; runtime_state = next; + if runtime_state == RuntimeState::Transition { + if let Some(commitment) = active_commitment.as_mut() { + commitment.transition_policy_id = Some(policy.id.clone()); + } + } active_policy = Some(policy); } EventType::TransitionStarted => { let payload: TransitionStartedPayload = parse_payload(&record, "transition started")?; ensure_schema_version(payload.schema_version, record.sequence)?; + validate_transition_started_payload(&payload).with_context(|| { + format!( + "validate transition started payload at sequence {}", + record.sequence + ) + })?; let active = active_commitment.as_ref().ok_or_else(|| { anyhow!( "transition requires active commitment at sequence {}", record.sequence ) })?; + let current_policy = active_policy.as_ref().ok_or_else(|| { + anyhow!( + "transition requires active policy at sequence {}", + record.sequence + ) + })?; ensure_record_commitment(&record, &payload.commitment_id)?; if payload.commitment_id != active.id { return Err(anyhow!( @@ -428,8 +472,29 @@ fn hydrate_session( } let mut commitment = active.clone(); commitment.state = composite.commitment_state; + let transition_policy = PolicySnapshot { + id: commitment + .transition_policy_id + .clone() + .unwrap_or_else(|| current_policy.id.clone()), + commitment_id: commitment.id.clone(), + schema_version: POLICY_SCHEMA_VERSION, + created_at_unix_secs: record.timestamp_unix_secs, + runtime_state: composite.runtime_state, + enforcement_level: current_policy.enforcement_level, + allowed_context: current_policy.allowed_context.clone(), + required_monitors: Vec::new(), + violation_actions: Vec::new(), + expires_at_unix_secs: None, + generated_by_agent_version: env!("CARGO_PKG_VERSION").to_string(), + }; + transition_policy.validate().with_context(|| { + format!("validate transition policy at sequence {}", record.sequence) + })?; + commitment.transition_policy_id = Some(transition_policy.id.clone()); runtime_state = composite.runtime_state; active_commitment = Some(commitment); + active_policy = Some(transition_policy); } ref unsupported => { return Err(anyhow!( @@ -452,6 +517,9 @@ fn hydrate_session( "session replay ended with commitment/policy mismatch" )); } + if policy.runtime_state != runtime_state { + return Err(anyhow!("session replay ended with runtime/policy mismatch")); + } } Ok((runtime_state, active_commitment, active_policy)) @@ -491,6 +559,19 @@ fn ensure_record_commitment(record: &EventRecord, commitment_id: &str) -> Result Ok(()) } +fn validate_transition_started_payload(payload: &TransitionStartedPayload) -> Result<()> { + if payload.reason.trim().is_empty() { + return Err(anyhow!("transition reason is required")); + } + if payload.return_target.trim().is_empty() { + return Err(anyhow!("transition return target is required")); + } + if payload.expected_duration_secs == 0 { + return Err(anyhow!("transition expected duration must be nonzero")); + } + Ok(()) +} + #[cfg(test)] mod tests { use super::{CommitmentCreatedPayload, PolicyAppliedPayload, TransitionStartedPayload}; @@ -549,6 +630,18 @@ mod tests { .unwrap() } + fn transition_policy_payload(commitment_id: &str) -> serde_json::Value { + serde_json::to_value(PolicyAppliedPayload { + schema_version: POLICY_SCHEMA_VERSION, + policy_id: "policy-transition-test".to_string(), + commitment_id: commitment_id.to_string(), + runtime_state: RuntimeState::Transition, + enforcement_level: EnforcementLevel::Warn, + allowed_context: AllowedContext::default(), + }) + .unwrap() + } + fn transition_started_payload() -> serde_json::Value { serde_json::to_value(TransitionStartedPayload { schema_version: POLICY_SCHEMA_VERSION, @@ -564,6 +657,46 @@ mod tests { .unwrap() } + fn append_active_session_with_transition_payload( + path: &PathBuf, + transition_payload: serde_json::Value, + ) { + 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_id.to_string()), + payload_json: transition_payload, + }, + ]) + .unwrap(); + } + fn error_chain_contains(err: &anyhow::Error, expected: &str) -> bool { err.chain() .any(|cause| cause.to_string().contains(expected)) @@ -874,6 +1007,112 @@ mod tests { fs::remove_file(path).unwrap(); } + #[test] + fn rejects_transition_started_with_empty_reason_on_replay() { + let path = temp_log_path("transition-empty-reason"); + let mut payload = transition_started_payload(); + payload["reason"] = json!(""); + + append_active_session_with_transition_payload(&path, payload); + + let err = SessionController::new(&path) + .expect_err("transition replay with empty reason must be rejected"); + + assert!(error_chain_contains(&err, "reason")); + fs::remove_file(path).unwrap(); + } + + #[test] + fn rejects_transition_started_with_empty_return_target_on_replay() { + let path = temp_log_path("transition-empty-return-target"); + let mut payload = transition_started_payload(); + payload["return_target"] = json!(""); + + append_active_session_with_transition_payload(&path, payload); + + let err = SessionController::new(&path) + .expect_err("transition replay with empty return target must be rejected"); + + assert!(error_chain_contains(&err, "return target")); + fs::remove_file(path).unwrap(); + } + + #[test] + fn rejects_transition_started_with_zero_duration_on_replay() { + let path = temp_log_path("transition-zero-duration"); + let mut payload = transition_started_payload(); + payload["expected_duration_secs"] = json!(0); + + append_active_session_with_transition_payload(&path, payload); + + let err = SessionController::new(&path) + .expect_err("transition replay with zero duration must be rejected"); + + assert!(error_chain_contains(&err, "expected duration")); + fs::remove_file(path).unwrap(); + } + + #[test] + fn reopens_transition_with_transition_policy() { + let path = temp_log_path("hydrate-transition"); + { + 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(); + session + .start_transition( + "Need to inspect failures", + "Return to session controller", + Duration::from_secs(300), + ) + .unwrap(); + } + + let reopened = SessionController::new(&path).unwrap(); + + assert_eq!(reopened.runtime_state(), RuntimeState::Transition); + assert_eq!( + reopened.active_commitment().unwrap().state, + CommitmentState::Paused + ); + assert!(reopened.active_policy().is_some()); + assert_eq!( + reopened.active_policy().unwrap().runtime_state, + RuntimeState::Transition + ); + + fs::remove_file(path).unwrap(); + } + + #[test] + fn rejects_transition_policy_runtime_mismatch_on_replay() { + let path = temp_log_path("transition-policy-runtime-mismatch"); + let commitment_id = "commitment-test"; + append_active_session_with_transition_payload(&path, transition_started_payload()); + { + let mut log = EventLog::open(&path).unwrap(); + log.append( + EventType::PolicyApplied, + RuntimeState::Active, + Some(commitment_id.to_string()), + transition_policy_payload(commitment_id), + ) + .unwrap(); + } + + let err = SessionController::new(&path) + .expect_err("transition policy runtime mismatch must be rejected"); + + assert!(error_chain_contains(&err, "policy runtime target mismatch")); + fs::remove_file(path).unwrap(); + } + #[test] fn transition_started_payload_has_stable_schema() { let path = temp_log_path("transition-payload"); @@ -896,10 +1135,12 @@ mod tests { .unwrap(); let records = read_records(&path); - let transition = records + let transition_index = records .iter() - .find(|record| record.event_type == EventType::TransitionStarted) + .position(|record| record.event_type == EventType::TransitionStarted) .unwrap(); + let transition = &records[transition_index]; + let transition_policy = &records[transition_index + 1]; assert_eq!(transition.payload_json["schema_version"], 1); assert_eq!( @@ -922,6 +1163,24 @@ mod tests { assert!(transition.payload_json.get("runtime_state").is_none()); assert!(transition.payload_json.get("commitment_state").is_none()); + assert_eq!(transition_policy.event_type, EventType::PolicyApplied); + assert_eq!(transition_policy.sequence, transition.sequence + 1); + assert_eq!( + transition_policy.previous_hash.as_deref(), + Some(transition.hash.as_str()) + ); + assert_eq!(transition_policy.runtime_state, RuntimeState::Transition); + assert_eq!(transition_policy.commitment_id, transition.commitment_id); + assert_eq!( + transition_policy.payload_json["commitment_id"], + transition.commitment_id.as_deref().unwrap() + ); + assert_eq!( + transition_policy.payload_json["runtime_state"], + "transition" + ); + assert_eq!(transition_policy.payload_json["enforcement_level"], "warn"); + fs::remove_file(path).unwrap(); } @@ -963,6 +1222,10 @@ mod tests { session.active_commitment().unwrap().state, crate::domain::CommitmentState::Paused ); + assert_eq!( + session.active_policy().unwrap().runtime_state, + RuntimeState::Transition + ); fs::remove_file(path).unwrap(); }