diff --git a/src/event_log.rs b/src/event_log.rs index a79d961..12497fe 100644 --- a/src/event_log.rs +++ b/src/event_log.rs @@ -35,6 +35,14 @@ pub struct EventRecord { pub hash: String, } +#[derive(Clone, Debug, PartialEq)] +pub struct PendingEventRecord { + pub event_type: EventType, + pub runtime_state: RuntimeState, + pub commitment_id: Option, + pub payload_json: Value, +} + #[derive(Debug)] pub struct EventLog { path: PathBuf, @@ -68,6 +76,26 @@ impl EventLog { commitment_id: Option, payload_json: Value, ) -> Result { + let mut records = self.append_batch([PendingEventRecord { + event_type, + runtime_state, + commitment_id, + payload_json, + }])?; + records + .pop() + .ok_or_else(|| anyhow!("single event append produced no record")) + } + + pub fn append_batch( + &mut self, + events: impl IntoIterator, + ) -> Result> { + let events: Vec = events.into_iter().collect(); + if events.is_empty() { + return Ok(Vec::new()); + } + create_parent_dirs(&self.path)?; let existed = self @@ -85,24 +113,42 @@ impl EventLog { let (next_sequence, previous_hash) = load_tail_locked(&mut file, &self.path)?; - let mut record = EventRecord { - schema_version: POLICY_SCHEMA_VERSION, - sequence: next_sequence, - timestamp_unix_secs: unix_secs_now(), - event_type, - commitment_id, - runtime_state, - payload_json, - previous_hash: previous_hash.clone(), - hash: String::new(), - }; - record.hash = record_hash(&record)?; + let mut records = Vec::with_capacity(events.len()); + let mut sequence = next_sequence; + let mut record_previous_hash = previous_hash; + for event in events { + let mut record = EventRecord { + schema_version: POLICY_SCHEMA_VERSION, + sequence, + timestamp_unix_secs: unix_secs_now(), + event_type: event.event_type, + commitment_id: event.commitment_id, + runtime_state: event.runtime_state, + payload_json: event.payload_json, + previous_hash: record_previous_hash, + hash: String::new(), + }; + record.hash = record_hash(&record)?; + record_previous_hash = Some(record.hash.clone()); + sequence += 1; + records.push(record); + } - let mut bytes = serde_json::to_vec(&record) - .with_context(|| format!("serialize event log record to {}", self.path.display()))?; - bytes.push(b'\n'); - file.write_all(&bytes) - .with_context(|| format!("write event log record to {}", self.path.display()))?; + let mut bytes = Vec::new(); + for record in &records { + serde_json::to_writer(&mut bytes, record).with_context(|| { + format!("serialize event log record to {}", self.path.display()) + })?; + bytes.push(b'\n'); + } + + file.write_all(&bytes).with_context(|| { + format!( + "write {} event log records to {}", + records.len(), + self.path.display() + ) + })?; file.flush() .with_context(|| format!("flush event log {}", self.path.display()))?; file.sync_data() @@ -112,10 +158,20 @@ impl EventLog { self.needs_parent_sync_after_append = false; } - self.next_sequence = next_sequence + 1; - self.previous_hash = Some(record.hash.clone()); + self.next_sequence = sequence; + self.previous_hash = record_previous_hash; - Ok(record) + Ok(records) + } + + pub fn records(&self) -> Result> { + let mut file = OpenOptions::new() + .read(true) + .open(&self.path) + .with_context(|| format!("open event log for read {}", self.path.display()))?; + file.lock_shared() + .with_context(|| format!("lock event log for read {}", self.path.display()))?; + load_records_locked(&mut file, &self.path) } } @@ -142,8 +198,16 @@ fn open_log_file(path: &Path) -> Result<(File, bool)> { } fn load_tail_locked(file: &mut File, path: &Path) -> Result<(u64, Option)> { + let records = load_records_locked(file, path)?; + let next_sequence = records.last().map_or(1, |record| record.sequence + 1); + let previous_hash = records.last().map(|record| record.hash.clone()); + Ok((next_sequence, previous_hash)) +} + +fn load_records_locked(file: &mut File, path: &Path) -> Result> { let mut next_sequence = 1; let mut previous_hash = None; + let mut records = Vec::new(); file.seek(SeekFrom::Start(0)) .with_context(|| format!("seek event log {}", path.display()))?; @@ -163,10 +227,11 @@ fn load_tail_locked(file: &mut File, path: &Path) -> Result<(u64, Option .map_err(|err| anyhow!("validate event log line {line_number}: {err}"))?; next_sequence += 1; - previous_hash = Some(record.hash); + previous_hash = Some(record.hash.clone()); + records.push(record); } - Ok((next_sequence, previous_hash)) + Ok(records) } #[cfg(unix)] @@ -287,6 +352,43 @@ mod tests { fs::remove_file(path).unwrap(); } + #[test] + fn batch_appends_hash_chained_adjacent_records() { + let path = temp_log_path("batch"); + let mut log = EventLog::open(&path).unwrap(); + + let records = log + .append_batch([ + PendingEventRecord { + event_type: EventType::CommitmentCreated, + runtime_state: RuntimeState::Active, + commitment_id: Some("commitment-1".to_string()), + payload_json: json!({"commitment_id": "commitment-1"}), + }, + PendingEventRecord { + event_type: EventType::PolicyApplied, + runtime_state: RuntimeState::Active, + commitment_id: Some("commitment-1".to_string()), + payload_json: json!({"policy_id": "policy-1"}), + }, + ]) + .unwrap(); + + assert_eq!(records.len(), 2); + assert_eq!(records[0].sequence, 1); + assert_eq!(records[1].sequence, 2); + assert_eq!( + records[1].previous_hash.as_deref(), + Some(records[0].hash.as_str()) + ); + + let contents = fs::read_to_string(&path).unwrap(); + assert_eq!(contents.lines().count(), 2); + EventLog::open(&path).unwrap(); + + fs::remove_file(path).unwrap(); + } + #[test] fn opening_existing_log_validates_chain_and_continues_sequence() { let path = temp_log_path("reopen"); diff --git a/src/session.rs b/src/session.rs index 2afaf7a..8ed8df1 100644 --- a/src/session.rs +++ b/src/session.rs @@ -1,13 +1,61 @@ -use crate::domain::{AllowedContext, Commitment, EnforcementLevel, PolicySnapshot, RuntimeState}; -use crate::event_log::{EventLog, EventType}; +use crate::domain::{ + AllowedContext, Commitment, CommitmentSource, CommitmentState, EnforcementLevel, + PolicySnapshot, RuntimeState, POLICY_SCHEMA_VERSION, +}; +use crate::event_log::{EventLog, EventRecord, EventType, PendingEventRecord}; use crate::state_machine::{ self, transition_commitment, transition_runtime, CommitmentAction, RuntimeAction, }; use anyhow::{anyhow, Context, Result}; -use serde_json::json; +use serde::{Deserialize, Serialize}; use std::path::Path; use std::time::Duration; +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct RuntimeTransitionPayload { + schema_version: u16, + action: String, + from: RuntimeState, + to: RuntimeState, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct CommitmentCreatedPayload { + schema_version: u16, + commitment_id: String, + source: CommitmentSource, + next_action: String, + success_condition: String, + timebox_secs: u64, + state: CommitmentState, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct PolicyAppliedPayload { + schema_version: u16, + policy_id: String, + commitment_id: String, + runtime_state: RuntimeState, + enforcement_level: EnforcementLevel, + allowed_context: AllowedContext, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct TransitionStartedPayload { + schema_version: u16, + reason: String, + return_target: String, + expected_duration_secs: u64, + runtime_from: RuntimeState, + runtime_to: RuntimeState, + commitment_from: CommitmentState, + commitment_to: CommitmentState, +} + #[derive(Debug)] pub struct SessionController { runtime_state: RuntimeState, @@ -24,11 +72,13 @@ impl SessionController { event_log_path.as_ref().display() ) })?; + let (runtime_state, active_commitment, active_policy) = + hydrate_session(&event_log).context("hydrate session from event log")?; Ok(Self { - runtime_state: RuntimeState::Locked, - active_commitment: None, - active_policy: None, + runtime_state, + active_commitment, + active_policy, event_log, }) } @@ -58,11 +108,13 @@ impl SessionController { self.active_commitment .as_ref() .map(|commitment| commitment.id.clone()), - json!({ - "from": previous_runtime_state, - "to": next_runtime_state, - "action": "enter_planning", - }), + serde_json::to_value(RuntimeTransitionPayload { + schema_version: POLICY_SCHEMA_VERSION, + action: "enter_planning".to_string(), + from: previous_runtime_state, + to: next_runtime_state, + }) + .context("serialize runtime transition payload")?, ) .context("log runtime transition into planning")?; @@ -100,21 +152,38 @@ impl SessionController { .context("create active policy snapshot")?; self.event_log - .append( - EventType::CommitmentCreated, - next_runtime_state, - Some(commitment.id.clone()), - serde_json::to_value(&commitment).context("serialize commitment payload")?, - ) - .context("log commitment creation")?; - self.event_log - .append( - EventType::PolicyApplied, - next_runtime_state, - Some(commitment.id.clone()), - serde_json::to_value(&policy).context("serialize policy payload")?, - ) - .context("log policy application")?; + .append_batch([ + PendingEventRecord { + event_type: EventType::CommitmentCreated, + runtime_state: next_runtime_state, + commitment_id: Some(commitment.id.clone()), + payload_json: serde_json::to_value(CommitmentCreatedPayload { + schema_version: POLICY_SCHEMA_VERSION, + commitment_id: commitment.id.clone(), + source: commitment.source.clone(), + next_action: commitment.next_action.clone(), + success_condition: commitment.success_condition.clone(), + timebox_secs: commitment.timebox_secs, + state: commitment.state.clone(), + }) + .context("serialize commitment payload")?, + }, + PendingEventRecord { + event_type: EventType::PolicyApplied, + runtime_state: next_runtime_state, + commitment_id: Some(commitment.id.clone()), + payload_json: serde_json::to_value(PolicyAppliedPayload { + schema_version: POLICY_SCHEMA_VERSION, + policy_id: policy.id.clone(), + commitment_id: policy.commitment_id.clone(), + runtime_state: policy.runtime_state, + enforcement_level: policy.enforcement_level, + allowed_context: policy.allowed_context.clone(), + }) + .context("serialize policy payload")?, + }, + ]) + .context("log commitment activation")?; self.runtime_state = next_runtime_state; self.active_commitment = Some(commitment); @@ -160,19 +229,17 @@ impl SessionController { EventType::TransitionStarted, composite.runtime_state, Some(next_commitment.id.clone()), - json!({ - "reason": reason, - "return_target": return_target, - "expected_duration_secs": expected_duration.as_secs(), - "runtime_state": { - "from": self.runtime_state, - "to": composite.runtime_state, - }, - "commitment_state": { - "from": previous_commitment_state, - "to": next_commitment_state, - }, - }), + serde_json::to_value(TransitionStartedPayload { + schema_version: POLICY_SCHEMA_VERSION, + 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")?, ) .context("log transition start")?; @@ -182,6 +249,202 @@ impl SessionController { } } +fn hydrate_session( + event_log: &EventLog, +) -> Result<(RuntimeState, Option, Option)> { + let mut runtime_state = RuntimeState::Locked; + let mut active_commitment: Option = None; + let mut active_policy: Option = None; + + for record in event_log.records()? { + match record.event_type { + EventType::RuntimeTransition => { + let payload: RuntimeTransitionPayload = + parse_payload(&record, "runtime transition")?; + ensure_schema_version(payload.schema_version, record.sequence)?; + if payload.action != "enter_planning" { + return Err(anyhow!( + "unsupported runtime transition action '{}' at sequence {}", + payload.action, + record.sequence + )); + } + if payload.from != runtime_state { + return Err(anyhow!( + "runtime transition source mismatch at sequence {}: expected {:?}, found {:?}", + record.sequence, + runtime_state, + payload.from + )); + } + let next = transition_runtime(runtime_state, RuntimeAction::EnterPlanning) + .with_context(|| { + format!("replay runtime transition at sequence {}", record.sequence) + })?; + if payload.to != next || record.runtime_state != next { + return Err(anyhow!( + "runtime transition target mismatch at sequence {}", + record.sequence + )); + } + runtime_state = next; + } + EventType::CommitmentCreated => { + let payload: CommitmentCreatedPayload = + parse_payload(&record, "commitment created")?; + ensure_schema_version(payload.schema_version, record.sequence)?; + ensure_record_commitment(&record, &payload.commitment_id)?; + let commitment = Commitment { + id: payload.commitment_id, + created_at_unix_secs: record.timestamp_unix_secs, + source: payload.source, + project_id: None, + template_id: None, + next_action: payload.next_action, + success_condition: payload.success_condition, + timebox_secs: payload.timebox_secs, + transition_policy_id: None, + state: payload.state, + }; + commitment.validate().with_context(|| { + format!("validate commitment at sequence {}", record.sequence) + })?; + active_commitment = Some(commitment); + } + EventType::PolicyApplied => { + let payload: PolicyAppliedPayload = parse_payload(&record, "policy applied")?; + ensure_schema_version(payload.schema_version, record.sequence)?; + ensure_record_commitment(&record, &payload.commitment_id)?; + let commitment = active_commitment.as_ref().ok_or_else(|| { + anyhow!( + "policy references unknown commitment '{}' at sequence {}", + payload.commitment_id, + record.sequence + ) + })?; + if commitment.id != payload.commitment_id { + return Err(anyhow!( + "policy commitment mismatch 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) + })?; + if payload.runtime_state != next || record.runtime_state != next { + return Err(anyhow!( + "policy runtime target mismatch at sequence {}", + record.sequence + )); + } + let policy = PolicySnapshot { + id: payload.policy_id, + commitment_id: payload.commitment_id, + schema_version: POLICY_SCHEMA_VERSION, + created_at_unix_secs: record.timestamp_unix_secs, + runtime_state: payload.runtime_state, + enforcement_level: payload.enforcement_level, + allowed_context: payload.allowed_context, + required_monitors: Vec::new(), + violation_actions: Vec::new(), + expires_at_unix_secs: None, + generated_by_agent_version: env!("CARGO_PKG_VERSION").to_string(), + }; + policy + .validate() + .with_context(|| format!("validate policy at sequence {}", record.sequence))?; + runtime_state = next; + active_policy = Some(policy); + } + EventType::TransitionStarted => { + let payload: TransitionStartedPayload = + parse_payload(&record, "transition started")?; + ensure_schema_version(payload.schema_version, record.sequence)?; + let active = active_commitment.as_ref().ok_or_else(|| { + anyhow!( + "transition requires active commitment at sequence {}", + record.sequence + ) + })?; + if payload.runtime_from != runtime_state || payload.commitment_from != active.state + { + return Err(anyhow!( + "transition source mismatch at sequence {}", + record.sequence + )); + } + let composite = + state_machine::start_transition(runtime_state, active.state.clone()) + .with_context(|| { + format!("replay transition start at sequence {}", record.sequence) + })?; + if payload.runtime_to != composite.runtime_state + || payload.commitment_to != composite.commitment_state + || record.runtime_state != composite.runtime_state + { + return Err(anyhow!( + "transition target mismatch at sequence {}", + record.sequence + )); + } + let mut commitment = active.clone(); + commitment.state = composite.commitment_state; + runtime_state = composite.runtime_state; + active_commitment = Some(commitment); + } + ref unsupported => { + return Err(anyhow!( + "unsupported session event {:?} at sequence {}", + unsupported, + record.sequence + )); + } + } + } + + Ok((runtime_state, active_commitment, active_policy)) +} + +fn parse_payload(record: &EventRecord, description: &str) -> Result +where + T: for<'de> Deserialize<'de>, +{ + serde_json::from_value(record.payload_json.clone()).with_context(|| { + format!( + "parse {description} payload at sequence {}", + record.sequence + ) + }) +} + +fn ensure_schema_version(schema_version: u16, sequence: u64) -> Result<()> { + if schema_version != POLICY_SCHEMA_VERSION { + return Err(anyhow!( + "unsupported payload schema version at sequence {}: expected {}, found {}", + sequence, + POLICY_SCHEMA_VERSION, + schema_version + )); + } + Ok(()) +} + +fn ensure_record_commitment(record: &EventRecord, commitment_id: &str) -> Result<()> { + if record.commitment_id.as_deref() != Some(commitment_id) { + return Err(anyhow!( + "commitment id mismatch at sequence {}", + record.sequence + )); + } + Ok(()) +} + #[cfg(test)] mod tests { use crate::domain::{EnforcementLevel, RuntimeState}; @@ -242,6 +505,138 @@ mod tests { fs::remove_file(path).unwrap(); } + #[test] + fn activation_batch_emits_adjacent_stable_payloads() { + let path = temp_log_path("activation-batch"); + 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 records = read_records(&path); + assert_eq!(records.len(), 3); + assert_eq!(records[1].event_type, EventType::CommitmentCreated); + assert_eq!(records[2].event_type, EventType::PolicyApplied); + assert_eq!(records[2].sequence, records[1].sequence + 1); + assert_eq!( + records[2].previous_hash.as_deref(), + Some(records[1].hash.as_str()) + ); + + let commitment_payload = &records[1].payload_json; + assert_eq!(commitment_payload["schema_version"], 1); + let commitment_id = records[1].commitment_id.as_deref().unwrap(); + assert_eq!(commitment_payload["commitment_id"], commitment_id); + assert_eq!(commitment_payload["source"], "manual"); + assert_eq!( + commitment_payload["next_action"], + "Implement session controller" + ); + assert_eq!( + commitment_payload["success_condition"], + "session tests pass" + ); + assert_eq!(commitment_payload["timebox_secs"], 1500); + assert_eq!(commitment_payload["state"], "active"); + assert!(commitment_payload.get("created_at_unix_secs").is_none()); + + let policy_payload = &records[2].payload_json; + assert_eq!(policy_payload["schema_version"], 1); + assert_eq!(policy_payload["commitment_id"], commitment_id); + assert_eq!(policy_payload["runtime_state"], "active"); + assert_eq!(policy_payload["enforcement_level"], "warn"); + assert_eq!( + policy_payload["allowed_context"], + serde_json::json!({ + "window_classes": [], + "window_title_substrings": [], + "domains": [], + "repos": [], + "commands": [], + }) + ); + assert!(policy_payload.get("created_at_unix_secs").is_none()); + + fs::remove_file(path).unwrap(); + } + + #[test] + fn hydrates_active_session_from_existing_log() { + let path = temp_log_path("hydrate-active"); + { + 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 reopened = SessionController::new(&path).unwrap(); + + assert_eq!(reopened.runtime_state(), RuntimeState::Active); + assert!(reopened.active_commitment().is_some()); + assert!(reopened.active_policy().is_some()); + + fs::remove_file(path).unwrap(); + } + + #[test] + fn transition_started_payload_has_stable_schema() { + let path = temp_log_path("transition-payload"); + 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 records = read_records(&path); + let transition = records + .iter() + .find(|record| record.event_type == EventType::TransitionStarted) + .unwrap(); + + assert_eq!(transition.payload_json["schema_version"], 1); + assert_eq!( + transition.payload_json["reason"], + "Need to inspect failures" + ); + assert_eq!( + transition.payload_json["return_target"], + "Return to session controller" + ); + assert_eq!(transition.payload_json["expected_duration_secs"], 300); + assert_eq!(transition.payload_json["runtime_from"], "active"); + assert_eq!(transition.payload_json["runtime_to"], "transition"); + assert_eq!(transition.payload_json["commitment_from"], "active"); + assert_eq!(transition.payload_json["commitment_to"], "paused"); + assert!(transition.payload_json.get("runtime_state").is_none()); + assert!(transition.payload_json.get("commitment_state").is_none()); + + fs::remove_file(path).unwrap(); + } + #[test] fn transition_requires_reason_and_return_target() { let path = temp_log_path("transition");