Keep transition policy consistent

src/session.rs

@@ -198,24 +198,14 @@ impl SessionController {
         return_target: impl Into<String>,
         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();
 
-        self.event_log
+        let transition_policy = PolicySnapshot::try_for_runtime(
-            .append(
+            next_commitment.id.clone(),
-                EventType::TransitionStarted,
             composite.runtime_state,
-                Some(next_commitment.id.clone()),
+            active_policy.enforcement_level,
-                serde_json::to_value(TransitionStartedPayload {
+            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: reason.into().trim().to_string(),
-                    return_target,
+            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_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,13 +367,8 @@ fn hydrate_session(
                         record.sequence
                     ));
                 }
-                if commitment.state != CommitmentState::Active {
+                let next = match commitment.state {
-                    return Err(anyhow!(
+                    CommitmentState::Active => transition_runtime(
-                        "policy requires active commitment at sequence {}",
-                        record.sequence
-                    ));
-                }
-                let next = transition_runtime(
                         runtime_state,
                         RuntimeAction::Activate {
                             policy_accepted: true,
@@ -362,7 +376,20 @@ fn hydrate_session(
                     )
                     .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();
     }