| 1 | #![allow(dead_code)] |
| 2 | // Legacy per-host replay fixtures only; live production consumers use the companion. |
| 3 | //! Sandboxed, read-only execution of the generated Whalesong world. This reuses |
| 4 | //! the workspace's existing QuickJS dependency; no JS filesystem/network APIs, |
| 5 | //! second Engine, async runtime, Node installation or external process is needed. |
| 6 | use std::sync::{Arc, Mutex, mpsc}; |
| 7 | use std::time::{Duration, Instant}; |
| 8 | |
| 9 | use rquickjs::{Context, Runtime}; |
| 10 | use serde::Deserialize; |
| 11 | |
| 12 | use super::audio::{self, Target}; |
| 13 | use super::persistence::{self, Store}; |
| 14 | |
| 15 | #[derive(Debug, Clone, PartialEq, Deserialize)] |
| 16 | pub struct Raster { |
| 17 | pub width: u16, |
| 18 | pub height: u16, |
| 19 | pub cells: Vec<u8>, |
| 20 | #[serde(rename = "timeMs")] |
| 21 | pub time_ms: f64, |
| 22 | /// Host-relative receipt clock, separate from the restored creature clock. |
| 23 | #[serde(skip)] |
| 24 | pub host_time_ms: f64, |
| 25 | pub channel: String, |
| 26 | pub arch: String, |
| 27 | pub hollow: bool, |
| 28 | pub dozing: bool, |
| 29 | pub lit: f64, |
| 30 | } |
| 31 | |
| 32 | impl Raster { |
| 33 | /// A hidden clock advance is not a new picture in reduced motion. |
| 34 | pub fn same_picture(&self, other: &Self) -> bool { |
| 35 | self.width == other.width |
| 36 | && self.height == other.height |
| 37 | && self.cells == other.cells |
| 38 | && self.channel == other.channel |
| 39 | && self.arch == other.arch |
| 40 | && self.hollow == other.hollow |
| 41 | && self.dozing == other.dozing |
| 42 | && self.lit == other.lit |
| 43 | } |
| 44 | } |
| 45 | |
| 46 | pub enum Command { |
| 47 | Export, |
| 48 | Observe { |
| 49 | json: String, |
| 50 | time_ms: f64, |
| 51 | }, |
| 52 | Advance { |
| 53 | time_ms: f64, |
| 54 | motion: bool, |
| 55 | waiting: bool, |
| 56 | width: u16, |
| 57 | height: u16, |
| 58 | audio: Option<Target>, |
| 59 | }, |
| 60 | } |
| 61 | |
| 62 | pub enum Notice { |
| 63 | Restored, |
| 64 | StorageUnavailable, |
| 65 | Exported(std::path::PathBuf), |
| 66 | ExportFailed, |
| 67 | } |
| 68 | |
| 69 | pub struct Worker { |
| 70 | pub tx: mpsc::SyncSender<Command>, |
| 71 | pub latest: Arc<Mutex<Option<Result<Raster, ()>>>>, |
| 72 | pub notices: mpsc::Receiver<Notice>, |
| 73 | } |
| 74 | |
| 75 | use super::persistence::export_recording; |
| 76 | |
| 77 | use super::audio_cursor::AudioCursor; |
| 78 | |
| 79 | impl Worker { |
| 80 | pub fn start(session: Option<String>) -> std::io::Result<Self> { |
| 81 | let (tx, rx) = mpsc::sync_channel(128); |
| 82 | let (notices_tx, notices) = mpsc::sync_channel(16); |
| 83 | let latest = Arc::new(Mutex::new(None)); |
| 84 | let output = Arc::clone(&latest); |
| 85 | std::thread::Builder::new() |
| 86 | .name("pet-world".into()) |
| 87 | .spawn(move || { |
| 88 | if run(rx, &output, ¬ices_tx, session.as_deref()).is_err() |
| 89 | && let Ok(mut slot) = output.lock() |
| 90 | { |
| 91 | *slot = Some(Err(())); |
| 92 | } |
| 93 | })?; |
| 94 | Ok(Self { |
| 95 | tx, |
| 96 | latest, |
| 97 | notices, |
| 98 | }) |
| 99 | } |
| 100 | } |
| 101 | |
| 102 | fn run( |
| 103 | rx: mpsc::Receiver<Command>, |
| 104 | output: &Mutex<Option<Result<Raster, ()>>>, |
| 105 | notices: &mpsc::SyncSender<Notice>, |
| 106 | session: Option<&str>, |
| 107 | ) -> Result<(), ()> { |
| 108 | let runtime = Runtime::new().map_err(|_| ())?; |
| 109 | runtime.set_memory_limit(64 * 1024 * 1024); |
| 110 | runtime.set_max_stack_size(2 * 1024 * 1024); |
| 111 | let deadline = Arc::new(Mutex::new(Instant::now() + Duration::from_secs(2))); |
| 112 | let check = Arc::clone(&deadline); |
| 113 | runtime.set_interrupt_handler(Some(Box::new(move || { |
| 114 | check.lock().map_or(true, |d| Instant::now() > *d) |
| 115 | }))); |
| 116 | let context = Context::full(&runtime).map_err(|_| ())?; |
| 117 | context |
| 118 | .with(|ctx| -> rquickjs::Result<()> { |
| 119 | let points: Vec<Vec<f64>> = include_str!("../ambient_life/whale-points.tsv") |
| 120 | .lines() |
| 121 | .map(|line| { |
| 122 | line.split_whitespace() |
| 123 | .filter_map(|s| s.parse().ok()) |
| 124 | .collect() |
| 125 | }) |
| 126 | .collect(); |
| 127 | ctx.globals().set( |
| 128 | "points", |
| 129 | serde_json::to_string(&points).expect("finite points"), |
| 130 | )?; |
| 131 | ctx.eval::<(), _>(include_bytes!("pet-native.js").as_slice())?; |
| 132 | ctx.eval::<(), _>("globalThis.pet = new PetNative(points, '', '[]', true)")?; |
| 133 | Ok(()) |
| 134 | }) |
| 135 | .map_err(|_| ())?; |
| 136 | let mut store = None; |
| 137 | let mut offset_ms = 0.0; |
| 138 | if let Some(session) = session { |
| 139 | // Hydrating a bounded recording validates its whole accepted history. |
| 140 | // It runs off the UI thread and can take longer than a frame command. |
| 141 | *deadline.lock().map_err(|_| ())? = Instant::now() + Duration::from_secs(10); |
| 142 | let loaded = (|| -> Result<Store, ()> { |
| 143 | let mut files = Store::open(session).map_err(|_| ())?; |
| 144 | if let Some(saved) = files.load().map_err(|_| ())? { |
| 145 | offset_ms = context |
| 146 | .with(|ctx| -> rquickjs::Result<f64> { |
| 147 | ctx.globals().set("savedHabitat", saved)?; |
| 148 | ctx.eval("pet.restoreRecording(savedHabitat); delete globalThis.savedHabitat; pet.resumeEngine()") |
| 149 | }) |
| 150 | .map_err(|_| ())?; |
| 151 | let _ = notices.try_send(Notice::Restored); |
| 152 | } |
| 153 | Ok(files) |
| 154 | })(); |
| 155 | match loaded { |
| 156 | Ok(files) => store = Some(files), |
| 157 | Err(()) => { |
| 158 | // A damaged/concurrent file stays untouched; this run can still |
| 159 | // show unknown/live telemetry and export its own accepted tape. |
| 160 | context |
| 161 | .with(|ctx| { |
| 162 | ctx.eval::<(), _>("delete globalThis.savedHabitat; globalThis.pet = new PetNative(points, '', '[]', true)") |
| 163 | }) |
| 164 | .map_err(|_| ())?; |
| 165 | offset_ms = 0.0; |
| 166 | let _ = notices.try_send(Notice::StorageUnavailable); |
| 167 | } |
| 168 | } |
| 169 | } |
| 170 | let mut saved_at = None; |
| 171 | let mut audio_cursor: Option<AudioCursor> = None; |
| 172 | for command in rx { |
| 173 | // A delayed host can ask for up to 300 fixed ticks (ten seconds). |
| 174 | // The worker remains interruptible without treating legitimate catch-up |
| 175 | // as a telemetry failure on a busy machine. The UI never waits here. |
| 176 | *deadline.lock().map_err(|_| ())? = Instant::now() + Duration::from_secs(5); |
| 177 | let save_at = match &command { |
| 178 | Command::Advance { time_ms, .. } |
| 179 | if saved_at.is_none_or(|last| time_ms - last >= 5_000.0) => |
| 180 | { |
| 181 | Some(*time_ms) |
| 182 | } |
| 183 | _ => None, |
| 184 | }; |
| 185 | context |
| 186 | .with(|ctx| -> rquickjs::Result<()> { |
| 187 | match command { |
| 188 | Command::Export => { |
| 189 | let result = session |
| 190 | .ok_or_else(|| std::io::Error::other("No saved session")) |
| 191 | .and_then(|id| { |
| 192 | let bytes = export_recording(&ctx, false).map_err(|_| { |
| 193 | let _ = ctx.catch(); |
| 194 | std::io::Error::other("Pet recording could not be exported") |
| 195 | })?; |
| 196 | persistence::export(id, &bytes) |
| 197 | }); |
| 198 | let notice = match result { |
| 199 | Ok(path) => Notice::Exported(path), |
| 200 | Err(_) => Notice::ExportFailed, |
| 201 | }; |
| 202 | let _ = notices.try_send(notice); |
| 203 | } |
| 204 | Command::Observe { json, time_ms } => { |
| 205 | ctx.globals().set("metadata", json)?; |
| 206 | ctx.globals().set("timeMs", time_ms + offset_ms)?; |
| 207 | ctx.eval::<(), _>("pet.observeEngine(metadata, timeMs)")?; |
| 208 | } |
| 209 | Command::Advance { |
| 210 | time_ms, |
| 211 | motion, |
| 212 | waiting, |
| 213 | width, |
| 214 | height, |
| 215 | audio, |
| 216 | } => { |
| 217 | ctx.globals().set("timeMs", time_ms + offset_ms)?; |
| 218 | ctx.globals().set("motion", motion)?; |
| 219 | ctx.globals().set("waiting", waiting)?; |
| 220 | ctx.globals().set("width", width)?; |
| 221 | ctx.globals().set("height", height)?; |
| 222 | let json: String = ctx.eval( |
| 223 | "pet.advanceEngine(timeMs,motion,waiting); pet.terminal(width,height)", |
| 224 | )?; |
| 225 | let mut frame: Raster = |
| 226 | serde_json::from_str(&json).map_err(|_| rquickjs::Error::Unknown)?; |
| 227 | if !frame.time_ms.is_finite() || frame.time_ms < offset_ms { |
| 228 | return Err(rquickjs::Error::Unknown); |
| 229 | } |
| 230 | frame.host_time_ms = time_ms; |
| 231 | if let Some(target) = audio.filter(Target::active) { |
| 232 | *deadline.lock().map_err(|_| rquickjs::Error::Unknown)? = |
| 233 | Instant::now() + Duration::from_millis(500); |
| 234 | if audio_cursor |
| 235 | .as_ref() |
| 236 | .is_none_or(|c| !c.target.same_stream(&target)) |
| 237 | { |
| 238 | audio_cursor = Some(AudioCursor { |
| 239 | target: target.clone(), |
| 240 | sample: (frame.time_ms * audio::SAMPLE_RATE as f64 / 1000.0) |
| 241 | .floor() |
| 242 | as usize, |
| 243 | voices: Vec::new(), |
| 244 | }); |
| 245 | } |
| 246 | if audio_cursor |
| 247 | .as_mut() |
| 248 | .expect("initialized audio cursor") |
| 249 | .present(&ctx, &target, frame.time_ms) |
| 250 | .is_err() |
| 251 | { |
| 252 | // Sound failure cannot stop telemetry or its recording. |
| 253 | let _ = ctx.catch(); |
| 254 | target.fail(); |
| 255 | audio_cursor = None; |
| 256 | } |
| 257 | *deadline.lock().map_err(|_| rquickjs::Error::Unknown)? = |
| 258 | Instant::now() + Duration::from_secs(5); |
| 259 | } else { |
| 260 | audio_cursor = None; |
| 261 | } |
| 262 | if let Ok(mut slot) = output.lock() { |
| 263 | *slot = Some(Ok(frame)); |
| 264 | } |
| 265 | } |
| 266 | } |
| 267 | Ok(()) |
| 268 | }) |
| 269 | .map_err(|_| ())?; |
| 270 | if let Some(at) = save_at { |
| 271 | save(&context, &mut store, notices)?; |
| 272 | saved_at = Some(at); |
| 273 | } |
| 274 | } |
| 275 | *deadline.lock().map_err(|_| ())? = Instant::now() + Duration::from_secs(5); |
| 276 | save(&context, &mut store, notices)?; |
| 277 | Ok(()) |
| 278 | } |
| 279 | |
| 280 | fn save( |
| 281 | context: &Context, |
| 282 | store: &mut Option<Store>, |
| 283 | notices: &mpsc::SyncSender<Notice>, |
| 284 | ) -> Result<(), ()> { |
| 285 | if let Some(files) = store { |
| 286 | let saved = context.with(|ctx| -> rquickjs::Result<()> { |
| 287 | let segment: Option<String> = |
| 288 | ctx.eval("pet.needsSegment() ? pet.prepareSegment() : null")?; |
| 289 | if let Some(text) = segment { |
| 290 | let archive = export_recording(&ctx, true)?; |
| 291 | let tick: u64 = |
| 292 | ctx.eval("Math.round(JSON.parse(pet.snapshot()).timeMs * 30 / 1000)")?; |
| 293 | files |
| 294 | .save_archived(&text, Some((&archive, tick))) |
| 295 | .map_err(|_| rquickjs::Error::Unknown)?; |
| 296 | ctx.eval::<(), _>("pet.commitSegment()")?; |
| 297 | } else { |
| 298 | let text: String = ctx.eval("pet.recording(true)")?; |
| 299 | files.save(&text).map_err(|_| rquickjs::Error::Unknown)?; |
| 300 | } |
| 301 | Ok(()) |
| 302 | }); |
| 303 | if saved.is_err() { |
| 304 | context.with(|ctx| { |
| 305 | let _ = ctx.catch(); |
| 306 | }); |
| 307 | *store = None; |
| 308 | let _ = notices.try_send(Notice::StorageUnavailable); |
| 309 | } |
| 310 | } |
| 311 | Ok(()) |
| 312 | } |
| 313 | |
| 314 | #[cfg(test)] |
| 315 | mod tests { |
| 316 | use super::*; |
| 317 | |
| 318 | struct ArtifactRoot(Option<std::path::PathBuf>); |
| 319 | impl Drop for ArtifactRoot { |
| 320 | fn drop(&mut self) { |
| 321 | crate::artifacts::set_test_artifact_sessions_root(self.0.take()); |
| 322 | } |
| 323 | } |
| 324 | |
| 325 | fn frame(worker: &Worker, at: f64) -> Raster { |
| 326 | frame_with_audio(worker, at, None) |
| 327 | } |
| 328 | |
| 329 | fn frame_with_audio(worker: &Worker, at: f64, audio: Option<Target>) -> Raster { |
| 330 | worker |
| 331 | .tx |
| 332 | .send(Command::Advance { |
| 333 | time_ms: at, |
| 334 | motion: false, |
| 335 | waiting: true, |
| 336 | width: 78, |
| 337 | height: 22, |
| 338 | audio, |
| 339 | }) |
| 340 | .unwrap(); |
| 341 | let until = Instant::now() + Duration::from_secs(10); |
| 342 | loop { |
| 343 | if let Some(value) = worker.latest.lock().unwrap().take() { |
| 344 | return value.expect("world worker failed"); |
| 345 | } |
| 346 | assert!(Instant::now() < until, "worker timed out"); |
| 347 | std::thread::sleep(Duration::from_millis(5)); |
| 348 | } |
| 349 | } |
| 350 | |
| 351 | fn finish(worker: Worker) { |
| 352 | let output = Arc::clone(&worker.latest); |
| 353 | drop(worker); |
| 354 | let until = Instant::now() + Duration::from_secs(10); |
| 355 | while Arc::strong_count(&output) > 1 { |
| 356 | assert!(Instant::now() < until, "worker did not finish saving"); |
| 357 | std::thread::sleep(Duration::from_millis(5)); |
| 358 | } |
| 359 | assert!(!matches!(*output.lock().unwrap(), Some(Err(())))); |
| 360 | } |
| 361 | |
| 362 | #[test] |
| 363 | fn audio_cursor_preserves_the_shared_score_across_buffer_boundaries() { |
| 364 | let runtime = Runtime::new().unwrap(); |
| 365 | let context = Context::full(&runtime).unwrap(); |
| 366 | let (sink, _packets) = audio::Output::capture(); |
| 367 | context.with(|ctx| { |
| 368 | let points: Vec<Vec<f64>> = include_str!("../ambient_life/whale-points.tsv") |
| 369 | .lines() |
| 370 | .map(|line| line.split_whitespace().map(|s| s.parse().unwrap()).collect()) |
| 371 | .collect(); |
| 372 | ctx.globals().set("points", serde_json::to_string(&points).unwrap()).unwrap(); |
| 373 | ctx.eval::<(), _>(include_bytes!("pet-native.js").as_slice()).unwrap(); |
| 374 | ctx.eval::<(), _>(r#"globalThis.pet = new PetNative(points, '', '[]', true); |
| 375 | pet.observeEngine('{"event":"approval_required","id":"audio-test"}', 0); |
| 376 | globalThis.allVoices = [];"#).unwrap(); |
| 377 | let mut cursor = AudioCursor { target: sink.target(), sample: 0, voices: Vec::new() }; |
| 378 | let mut received = Vec::new(); |
| 379 | // Compare the actual cursor's samples without a wall-clock delivery |
| 380 | // deadline. The sink tests cover expiry, queue bounds and interleaving. |
| 381 | for tick in 1..=48 { |
| 382 | let at = f64::from(tick) * 1000.0 / 30.0; |
| 383 | ctx.globals().set("timeMs", at).unwrap(); |
| 384 | ctx.eval::<(), _>("pet.advanceEngine(timeMs,false,true); allVoices.push(...JSON.parse(pet.snapshot()).voices)").unwrap(); |
| 385 | let time: f64 = ctx.eval("JSON.parse(pet.snapshot()).timeMs").unwrap(); |
| 386 | let [left, right] = cursor.render_samples(&ctx, time).unwrap().unwrap(); |
| 387 | received.extend(left.iter().zip(&right) |
| 388 | .flat_map(|(l, r)| [*l, *r]).flat_map(f32::to_le_bytes)); |
| 389 | } |
| 390 | let expected: String = ctx.eval(format!("pet.pcm(JSON.stringify(allVoices),0,{},48000)", cursor.sample)).unwrap(); |
| 391 | let [left, right]: [Vec<f32>; 2] = serde_json::from_str(&expected).unwrap(); |
| 392 | let interleaved: Vec<_> = left.iter().zip(&right) |
| 393 | .flat_map(|(l, r)| [*l, *r]).flat_map(f32::to_le_bytes).collect(); |
| 394 | assert!(left.iter().any(|s| s.abs() > 0.001), "fixture must exercise actual voices"); |
| 395 | assert_eq!(received, interleaved, "host chunking changed the shared PCM"); |
| 396 | // Reopening a stream at the current clock cannot replay its past. |
| 397 | let (reopened, packets) = audio::Output::capture(); |
| 398 | assert!(!sink.target().same_stream(&reopened.target())); |
| 399 | cursor.voices.clear(); |
| 400 | cursor.present(&ctx, &reopened.target(), cursor.sample as f64 / 48.0).unwrap(); |
| 401 | assert!(packets.try_recv().is_err()); |
| 402 | }); |
| 403 | } |
| 404 | |
| 405 | #[test] |
| 406 | fn output_failures_do_not_stop_the_world() { |
| 407 | let worker = Worker::start(None).unwrap(); |
| 408 | let (sink, packets) = audio::Output::capture(); |
| 409 | drop(packets); |
| 410 | frame_with_audio(&worker, 0.0, Some(sink.target())); |
| 411 | let before = frame_with_audio(&worker, 400.0, Some(sink.target())); |
| 412 | // Provoke the disconnect directly rather than relying on the worker to |
| 413 | // consume a target inside `Target::current()`'s 500 ms freshness |
| 414 | // window. `send` returns `Ok(())` early for a stale target and never |
| 415 | // reaches `try_send`, so on a host slow enough to miss that window the |
| 416 | // failure is never recorded — which is why this was red on Windows and |
| 417 | // green everywhere else. This is the same path the worker takes, minus |
| 418 | // the wall clock: a fresh target, a live `active()`, and a `try_send` |
| 419 | // that observes `Disconnected` because `packets` was dropped above. |
| 420 | assert!( |
| 421 | sink.target().send([vec![0.0], vec![0.0]]).is_err(), |
| 422 | "a dropped receiver must fail a fresh send" |
| 423 | ); |
| 424 | assert!(sink.failed()); |
| 425 | worker.tx.send(Command::Export).unwrap(); |
| 426 | assert!(matches!( |
| 427 | worker |
| 428 | .notices |
| 429 | .recv_timeout(Duration::from_secs(10)) |
| 430 | .unwrap(), |
| 431 | Notice::ExportFailed |
| 432 | )); |
| 433 | let after = frame_with_audio(&worker, 800.0, Some(sink.target())); |
| 434 | assert!(after.time_ms > before.time_ms); |
| 435 | assert!(after.hollow, "sound failure is not observed telemetry"); |
| 436 | finish(worker); |
| 437 | } |
| 438 | |
| 439 | #[test] |
| 440 | fn session_checkpoint_reopens_in_the_actual_worker_without_reviving_an_approval() { |
| 441 | let _guard = crate::artifacts::TEST_ARTIFACT_SESSIONS_GUARD |
| 442 | .lock() |
| 443 | .unwrap(); |
| 444 | let root = tempfile::tempdir().unwrap(); |
| 445 | let _restore = ArtifactRoot(crate::artifacts::set_test_artifact_sessions_root(Some( |
| 446 | root.path().to_owned(), |
| 447 | ))); |
| 448 | let worker = Worker::start(Some("pet-worker-test".into())).unwrap(); |
| 449 | worker |
| 450 | .tx |
| 451 | .send(Command::Observe { |
| 452 | json: r#"{"event":"approval_required","id":"old-approval"}"#.into(), |
| 453 | time_ms: 0.0, |
| 454 | }) |
| 455 | .unwrap(); |
| 456 | let before = frame(&worker, 5_000.0); |
| 457 | assert_eq!(before.channel, "human"); |
| 458 | worker.tx.send(Command::Export).unwrap(); |
| 459 | let export = match worker |
| 460 | .notices |
| 461 | .recv_timeout(Duration::from_secs(10)) |
| 462 | .unwrap() |
| 463 | { |
| 464 | Notice::Exported(path) => path, |
| 465 | _ => panic!("The live world did not export"), |
| 466 | }; |
| 467 | let exported: serde_json::Value = |
| 468 | serde_json::from_slice(&std::fs::read(export).unwrap()).unwrap(); |
| 469 | assert_eq!(exported["checkpoint"]["frame"]["timeMs"], before.time_ms); |
| 470 | assert!(exported["checkpoint"]["sim"]["particles"].is_array()); |
| 471 | finish(worker); |
| 472 | let path = root |
| 473 | .path() |
| 474 | .join("pet-worker-test/artifacts/pet/habitat.json"); |
| 475 | let saved: serde_json::Value = |
| 476 | serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap(); |
| 477 | assert_eq!( |
| 478 | exported, saved, |
| 479 | "Export omitted the current pose, score or history" |
| 480 | ); |
| 481 | let worker = Worker::start(Some("pet-worker-test".into())).unwrap(); |
| 482 | let after = frame(&worker, 800.0); |
| 483 | assert!(after.hollow); |
| 484 | assert!(after.time_ms > before.time_ms); |
| 485 | assert_eq!(after.host_time_ms, 800.0); |
| 486 | assert!( |
| 487 | worker |
| 488 | .notices |
| 489 | .try_iter() |
| 490 | .any(|n| matches!(n, Notice::Restored)) |
| 491 | ); |
| 492 | finish(worker); |
| 493 | let resumed: serde_json::Value = |
| 494 | serde_json::from_slice(&std::fs::read(path).unwrap()).unwrap(); |
| 495 | assert_eq!( |
| 496 | &resumed["tape"].as_array().unwrap()[..saved["tape"].as_array().unwrap().len()], |
| 497 | saved["tape"].as_array().unwrap() |
| 498 | ); |
| 499 | } |
| 500 | } |
| 501 |