feat(conductor): persist scene state across restarts

.gitignore

@@ -1,3 +1,7 @@
 /target
 __pycache__/
 *.pyc
+pchi_governance_key.hex
+pchi_governance_log.jsonl
+pchi_scene_state.json
+pchi_scene_state.json.tmp

primeswarm-pchi/src/conductor.rs

@@ -23,6 +23,12 @@ pub struct PCHIConductor {
 
     /// Equilibrium threshold
     equilibrium_threshold: f64,
+
+    /// Where to periodically snapshot scene state, if persistence is
+    /// enabled (`None` by default — see `with_state_file`). A crash
+    /// restart with this unset starts from a blank `SceneState`, same as
+    /// every version of this conductor before this field existed.
+    state_file: Option<String>,
 }
 
 impl PCHIConductor {
@@ -53,6 +59,7 @@ impl PCHIConductor {
             governance: Arc::new(GovernanceLog::open(signing_key_path, governance_log_path)?),
             transport: TransportLayer::new(),
             equilibrium_threshold,
+            state_file: None,
         })
     }
 
@@ -70,9 +77,38 @@ impl PCHIConductor {
             governance: Arc::new(GovernanceLog::open(signing_key_path, governance_log_path)?),
             transport: TransportLayer::new(),
             equilibrium_threshold,
+            state_file: None,
         })
     }
 
+    /// Enable periodic scene-state persistence to `path`, loading an
+    /// existing snapshot from it right now if one's there — so a restart
+    /// resumes from the last snapshot instead of starting blank. Chain
+    /// this onto `new`/`with_governance_paths`/`with_rules_file` before
+    /// wrapping the conductor in `Arc` and spawning its background loops;
+    /// `run_state_snapshot_loop` is what actually keeps writing to `path`
+    /// afterward.
+    ///
+    /// This is a resilience convenience, not the safety-critical
+    /// durability guarantee the governance log already has (that log
+    /// fsyncs every signed receipt as it's recorded) — losing scene
+    /// state on an ungraceful crash means tools and the conductor
+    /// disagree about "where were we," not that an unsafe state went
+    /// unlogged.
+    pub fn with_state_file(mut self, path: &str) -> Self {
+        match SceneState::load_snapshot(path) {
+            Some(loaded) => {
+                info!("Resumed scene state from {} (last updated {})", path, loaded.timestamp);
+                self.state = Arc::new(RwLock::new(loaded));
+            }
+            None => {
+                info!("No existing scene state snapshot at {} — starting fresh", path);
+            }
+        }
+        self.state_file = Some(path.to_string());
+        self
+    }
+
     /// Process an incoming PCHI message
     pub async fn process_message(&self, message: PCHIMessage) -> Result<(), ConductorError> {
         // Validate the message
@@ -285,6 +321,50 @@ impl PCHIConductor {
     pub async fn stop_transport(&mut self) -> Result<(), ConductorError> {
         self.transport.stop().await
     }
+
+    /// Periodically snapshot scene state to disk, if `with_state_file`
+    /// enabled it — a no-op loop that returns immediately otherwise, so
+    /// it's always safe to spawn alongside the other background loops
+    /// regardless of whether persistence is configured. Skips the write
+    /// when nothing has changed since the last snapshot, compared via
+    /// `SceneState::timestamp` (already updated by every mutation), so
+    /// an idle conductor isn't rewriting an unchanged file every
+    /// interval.
+    pub async fn run_state_snapshot_loop(self: Arc<Self>, interval: std::time::Duration) {
+        let Some(path) = self.state_file.clone() else {
+            return;
+        };
+        let mut last_snapshotted: Option<chrono::DateTime<chrono::Utc>> = None;
+        loop {
+            tokio::time::sleep(interval).await;
+            let (timestamp, snapshot) = {
+                let state = self.state.read().await;
+                (state.timestamp, state.clone())
+            };
+            if last_snapshotted == Some(timestamp) {
+                continue;
+            }
+            match snapshot.save_snapshot(&path) {
+                Ok(()) => last_snapshotted = Some(timestamp),
+                Err(e) => error!("Failed to write scene state snapshot to {}: {}", path, e),
+            }
+        }
+    }
+
+    /// Snapshot scene state to disk right now, if persistence is
+    /// enabled. Called on graceful shutdown so a planned restart never
+    /// loses anything, regardless of where in `run_state_snapshot_loop`'s
+    /// interval the shutdown happened.
+    pub async fn snapshot_state_now(&self) {
+        let Some(path) = &self.state_file else {
+            return;
+        };
+        let state = self.state.read().await;
+        match state.save_snapshot(path) {
+            Ok(()) => info!("Final scene state snapshot written to {}", path),
+            Err(e) => error!("Failed to write final scene state snapshot to {}: {}", path, e),
+        }
+    }
 }
 
 impl Default for PCHIConductor {
@@ -488,6 +568,53 @@ mod tests {
         let _ = std::fs::remove_dir_all(&dir);
     }
 
+    #[tokio::test]
+    async fn restart_resumes_scene_state_from_a_snapshot() {
+        let dir = std::env::temp_dir().join(format!("pchi-restart-test-{}", uuid::Uuid::new_v4()));
+        std::fs::create_dir_all(&dir).unwrap();
+        let state_path = dir.join("state.json");
+
+        // "Before restart": set a value, then snapshot the way graceful
+        // shutdown does.
+        let before = PCHIConductor::with_governance_paths(
+            1e-12,
+            dir.join("key.hex").to_str().unwrap(),
+            dir.join("log.jsonl").to_str().unwrap(),
+        )
+        .unwrap()
+        .with_state_file(state_path.to_str().unwrap());
+        let msg = PCHIMessage::new(
+            MessageType::ControlParameter,
+            "test".to_string(),
+            Payload::ControlParameter(ControlParameterData {
+                target_id: "kraken_main".to_string(),
+                parameter: "wobbliness".to_string(),
+                value: serde_json::json!(0.42),
+            }),
+        );
+        before.process_message(msg).await.unwrap();
+        before.snapshot_state_now().await;
+
+        // "After restart": a brand new conductor, same state file.
+        let after = PCHIConductor::with_governance_paths(
+            1e-12,
+            dir.join("key2.hex").to_str().unwrap(),
+            dir.join("log2.jsonl").to_str().unwrap(),
+        )
+        .unwrap()
+        .with_state_file(state_path.to_str().unwrap());
+
+        let resumed = after.get_state().await;
+        assert_eq!(
+            resumed.get_control_parameter("kraken_main.wobbliness"),
+            Some(0.42),
+            "a conductor pointed at the same state file must resume what the previous one set, \
+             not start blank"
+        );
+
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
     #[tokio::test]
     async fn with_rules_file_rejects_a_missing_path_instead_of_silently_falling_back() {
         let dir = std::env::temp_dir().join(format!("pchi-rulesfile-missing-{}", uuid::Uuid::new_v4()));

primeswarm-pchi/src/main.rs

@@ -17,6 +17,12 @@ struct Cli {
     /// same rules every fresh checkout has always run, unchanged.
     #[arg(long, env = "PCHI_RULES_FILE")]
     rules: Option<String>,
+
+    /// Where to periodically snapshot scene state, so a restart resumes
+    /// instead of starting blank. Always enabled, same as the governance
+    /// log's own default-local-file behavior.
+    #[arg(long, env = "PCHI_STATE_FILE", default_value = "pchi_scene_state.json")]
+    state_file: String,
 }
 
 #[tokio::main]
@@ -46,7 +52,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             })
         }
         None => PCHIConductor::new(1e-12),
-    };
+    }
+    .with_state_file(&cli.state_file);
 
     println!("PCHI Conductor started");
 
@@ -122,17 +129,31 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let ws_handle = tokio::spawn(conductor.clone().run_ws_accept_loop());
     println!("Accepting governance-event WebSocket clients on ws://127.0.0.1:8889");
 
+    // Periodically snapshot scene state so a restart resumes instead of
+    // starting blank. A no-op loop if persistence somehow ended up
+    // disabled, so always safe to spawn.
+    let snapshot_handle = tokio::spawn(
+        conductor.clone().run_state_snapshot_loop(std::time::Duration::from_secs(2)),
+    );
+    println!("Persisting scene state to {} every 2s", cli.state_file);
+
     println!("PCHI Conductor running. Press Ctrl+C to stop.");
 
     // Wait for shutdown signal
     tokio::signal::ctrl_c().await?;
     println!("Shutting down...");
 
+    // One last snapshot before tearing down, so a planned restart loses
+    // nothing regardless of where in the snapshot loop's interval this
+    // landed.
+    conductor.snapshot_state_now().await;
+
     // Stop background tasks (dropping the process closes the sockets;
     // stop_transport needs &mut self, which the Arc above gave up)
     web_handle.abort();
     udp_handle.abort();
     ws_handle.abort();
+    snapshot_handle.abort();
 
     println!("PCHI Conductor stopped");
 

primeswarm-pchi/src/state.rs

@@ -143,10 +143,101 @@ impl SceneState {
             EquilibriumStatus::Disequilibrium { residual }
         };
     }
+
+    /// Atomically write this state to `path` as JSON: serialize, write to
+    /// a sibling temp file, then rename into place. The rename is what
+    /// makes it atomic — a crash mid-write leaves the temp file
+    /// incomplete but never touches `path` itself, so the file there is
+    /// always either the previous complete snapshot or the new one,
+    /// never a half-written one.
+    pub fn save_snapshot(&self, path: &str) -> std::io::Result<()> {
+        let json = serde_json::to_vec_pretty(self)
+            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
+        let tmp_path = format!("{path}.tmp");
+        std::fs::write(&tmp_path, json)?;
+        std::fs::rename(&tmp_path, path)?;
+        Ok(())
+    }
+
+    /// Load a previously saved snapshot, if one exists and parses.
+    /// Returns `None` rather than an error on a missing or corrupt file:
+    /// scene state is a resilience convenience, not the safety-critical
+    /// record (that's the governance log, which does fail hard on
+    /// corruption — see `governance.rs`) — a bad snapshot should log a
+    /// warning and start fresh, not keep the conductor from coming up.
+    pub fn load_snapshot(path: &str) -> Option<Self> {
+        let bytes = std::fs::read(path).ok()?;
+        match serde_json::from_slice(&bytes) {
+            Ok(state) => Some(state),
+            Err(e) => {
+                tracing::warn!(
+                    "Scene state snapshot at {} is unreadable ({}), starting fresh",
+                    path,
+                    e
+                );
+                None
+            }
+        }
+    }
 }
 
 impl Default for SceneState {
     fn default() -> Self {
         Self::new()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn temp_path(name: &str) -> std::path::PathBuf {
+        std::env::temp_dir().join(format!("pchi-state-test-{}-{}", name, uuid::Uuid::new_v4()))
+    }
+
+    #[test]
+    fn save_then_load_round_trips_real_values() {
+        let path = temp_path("roundtrip");
+        let mut state = SceneState::new();
+        state.set_control_parameter("tentacle_system.total_curl".to_string(), 2.5);
+        state.update_musical_context(MusicalContext {
+            bpm: 128.0,
+            beat: 3,
+            kick: true,
+            snare: false,
+            section: "drop".to_string(),
+        });
+
+        state.save_snapshot(path.to_str().unwrap()).unwrap();
+        let loaded = SceneState::load_snapshot(path.to_str().unwrap()).unwrap();
+
+        assert_eq!(loaded.get_control_parameter("tentacle_system.total_curl"), Some(2.5));
+        assert_eq!(loaded.musical_context.unwrap().beat, 3);
+        assert_eq!(loaded.scene_id, state.scene_id, "identity, not just values, must survive a round trip");
+
+        let _ = std::fs::remove_file(&path);
+    }
+
+    #[test]
+    fn save_leaves_no_temp_file_behind() {
+        let path = temp_path("no-temp-litter");
+        SceneState::new().save_snapshot(path.to_str().unwrap()).unwrap();
+        assert!(path.exists());
+        assert!(!std::path::Path::new(&format!("{}.tmp", path.to_str().unwrap())).exists());
+        let _ = std::fs::remove_file(&path);
+    }
+
+    #[test]
+    fn load_missing_file_returns_none_not_an_error() {
+        let path = temp_path("does-not-exist");
+        assert!(SceneState::load_snapshot(path.to_str().unwrap()).is_none());
+    }
+
+    #[test]
+    fn load_corrupt_file_returns_none_instead_of_panicking() {
+        let path = temp_path("corrupt");
+        std::fs::write(&path, b"not valid json at all").unwrap();
+        assert!(SceneState::load_snapshot(path.to_str().unwrap()).is_none());
+        let _ = std::fs::remove_file(&path);
+    }
+}