| 1 | //! Presentation-only companion. Replaces per-view live world ownership; the |
| 2 | //! existing Engine projection, PetNative world and score remain authoritative. |
| 3 | //! No agent, prompt, provider or execution API is available to this process. |
| 4 | use std::collections::{BTreeMap, VecDeque}; |
| 5 | use std::io::{self, Read}; |
| 6 | use std::path::{Path, PathBuf}; |
| 7 | use std::sync::{Arc, Mutex, mpsc}; |
| 8 | use std::time::{Duration, Instant}; |
| 9 | |
| 10 | use axum::{ |
| 11 | Json, Router, |
| 12 | extract::{DefaultBodyLimit, State}, |
| 13 | http::{HeaderMap, StatusCode}, |
| 14 | response::{Html, IntoResponse, Response}, |
| 15 | routing::{get, post}, |
| 16 | }; |
| 17 | use rquickjs::{Context, Runtime}; |
| 18 | use serde::{Deserialize, Serialize}; |
| 19 | use serde_json::{Value, json}; |
| 20 | |
| 21 | use super::{appearance::Appearance, persistence::Store}; |
| 22 | use crate::fleet::files::{WorkspaceFile, same_file}; |
| 23 | |
| 24 | const LEASE: Duration = Duration::from_secs(2); |
| 25 | const MAX_CLIENTS: usize = 4096; |
| 26 | |
| 27 | #[derive(Clone, Serialize, Deserialize)] |
| 28 | #[serde(deny_unknown_fields)] |
| 29 | pub struct Descriptor { |
| 30 | pub version: u8, |
| 31 | pub port: u16, |
| 32 | pub token: String, |
| 33 | pub identity: String, |
| 34 | } |
| 35 | |
| 36 | pub fn directory() -> io::Result<PathBuf> { |
| 37 | if let Some(path) = std::env::var_os("CODEWHALE_PET_HOME") { |
| 38 | return Ok(path.into()); |
| 39 | } |
| 40 | Ok(dirs::home_dir() |
| 41 | .ok_or_else(|| io::Error::other("Home directory unavailable"))? |
| 42 | .join(".codewhale/pet-shared")) |
| 43 | } |
| 44 | |
| 45 | pub fn descriptor(root: &Path) -> io::Result<Descriptor> { |
| 46 | let file = WorkspaceFile::open(root, Path::new("connection.json"), false)?.open_file()?; |
| 47 | let mut text = String::new(); |
| 48 | file.take(4097).read_to_string(&mut text)?; |
| 49 | if text.len() > 4096 { |
| 50 | return Err(io::Error::other("Invalid pet connection")); |
| 51 | } |
| 52 | let d: Descriptor = serde_json::from_str(&text)?; |
| 53 | if d.version != 1 |
| 54 | || d.port == 0 |
| 55 | || d.token.len() != 64 |
| 56 | || !d.token.bytes().all(|b| b.is_ascii_hexdigit()) |
| 57 | || uuid::Uuid::parse_str(&d.identity).is_err() |
| 58 | { |
| 59 | return Err(io::Error::other("Invalid pet connection")); |
| 60 | } |
| 61 | Ok(d) |
| 62 | } |
| 63 | |
| 64 | #[derive(Clone, Serialize, Deserialize)] |
| 65 | struct Receipt { |
| 66 | seq: u64, |
| 67 | hash: String, |
| 68 | cursor: u64, |
| 69 | } |
| 70 | |
| 71 | #[derive(Serialize, Deserialize)] |
| 72 | struct Saved { |
| 73 | version: u8, |
| 74 | identity: String, |
| 75 | token: String, |
| 76 | port: u16, |
| 77 | source: String, |
| 78 | source_revision: u64, |
| 79 | cursor: u64, |
| 80 | clients: BTreeMap<String, Receipt>, |
| 81 | recording: Value, |
| 82 | #[serde(default)] |
| 83 | appearance: Appearance, |
| 84 | } |
| 85 | |
| 86 | #[derive(Clone, Deserialize, Serialize)] |
| 87 | #[serde(deny_unknown_fields)] |
| 88 | pub struct Request { |
| 89 | pub identity: String, |
| 90 | pub client: String, |
| 91 | pub seq: u64, |
| 92 | pub source_revision: u64, |
| 93 | pub action: Action, |
| 94 | } |
| 95 | |
| 96 | #[derive(Clone, Deserialize, Serialize)] |
| 97 | #[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] |
| 98 | pub enum Action { |
| 99 | Interact { food: bool, x: f64, y: f64 }, |
| 100 | Select { source: String }, |
| 101 | Appearance { appearance: Appearance }, |
| 102 | } |
| 103 | |
| 104 | #[derive(Deserialize)] |
| 105 | #[serde(deny_unknown_fields)] |
| 106 | pub struct Producer { |
| 107 | pub identity: String, |
| 108 | pub epoch: String, |
| 109 | pub client: String, |
| 110 | pub source: String, |
| 111 | pub source_revision: u64, |
| 112 | pub seq: u64, |
| 113 | pub waiting: bool, |
| 114 | pub events: Vec<Value>, |
| 115 | } |
| 116 | |
| 117 | #[derive(Deserialize)] |
| 118 | #[serde(deny_unknown_fields)] |
| 119 | pub struct AudioLease { |
| 120 | pub client: String, |
| 121 | pub enabled: bool, |
| 122 | } |
| 123 | |
| 124 | enum Work { |
| 125 | Action(Request, tokio::sync::oneshot::Sender<Result<Value, String>>), |
| 126 | Producer( |
| 127 | Producer, |
| 128 | tokio::sync::oneshot::Sender<Result<Value, String>>, |
| 129 | ), |
| 130 | Audio( |
| 131 | AudioLease, |
| 132 | tokio::sync::oneshot::Sender<Result<Value, String>>, |
| 133 | ), |
| 134 | Export(tokio::sync::oneshot::Sender<Result<Value, String>>), |
| 135 | } |
| 136 | |
| 137 | #[derive(Clone)] |
| 138 | struct Service { |
| 139 | descriptor: Descriptor, |
| 140 | frames: Arc<Mutex<VecDeque<Value>>>, |
| 141 | tx: mpsc::SyncSender<Work>, |
| 142 | } |
| 143 | |
| 144 | fn valid_client(id: &str) -> bool { |
| 145 | uuid::Uuid::parse_str(id).is_ok() |
| 146 | } |
| 147 | fn valid_source(id: &str) -> bool { |
| 148 | !id.is_empty() |
| 149 | && id.len() <= 128 |
| 150 | && id |
| 151 | .bytes() |
| 152 | .all(|b| b.is_ascii_alphanumeric() || b"-_:./".contains(&b)) |
| 153 | } |
| 154 | |
| 155 | fn authorized(headers: &HeaderMap, state: &Service) -> bool { |
| 156 | let origin = format!("http://127.0.0.1:{}", state.descriptor.port); |
| 157 | let host = format!("127.0.0.1:{}", state.descriptor.port); |
| 158 | if headers.get("host").and_then(|v| v.to_str().ok()) != Some(host.as_str()) { |
| 159 | return false; |
| 160 | } |
| 161 | if headers |
| 162 | .get("origin") |
| 163 | .is_some_and(|o| o.as_bytes() != origin.as_bytes()) |
| 164 | { |
| 165 | return false; |
| 166 | } |
| 167 | let bearer = format!("Bearer {}", state.descriptor.token); |
| 168 | let cookie = format!("cw_pet={}", state.descriptor.token); |
| 169 | headers |
| 170 | .get("authorization") |
| 171 | .is_some_and(|v| v.as_bytes() == bearer.as_bytes()) |
| 172 | || headers |
| 173 | .get("cookie") |
| 174 | .and_then(|v| v.to_str().ok()) |
| 175 | .is_some_and(|v| v.split(';').any(|v| v.trim() == cookie)) |
| 176 | } |
| 177 | |
| 178 | fn answer(result: Result<Value, String>) -> Response { |
| 179 | match result { |
| 180 | Ok(value) => Json(value).into_response(), |
| 181 | Err(error) => (StatusCode::CONFLICT, Json(json!({"error": error}))).into_response(), |
| 182 | } |
| 183 | } |
| 184 | |
| 185 | async fn submit( |
| 186 | state: &Service, |
| 187 | make: impl FnOnce(tokio::sync::oneshot::Sender<Result<Value, String>>) -> Work, |
| 188 | ) -> Response { |
| 189 | let (tx, rx) = tokio::sync::oneshot::channel(); |
| 190 | if state.tx.try_send(make(tx)).is_err() { |
| 191 | return StatusCode::SERVICE_UNAVAILABLE.into_response(); |
| 192 | } |
| 193 | match tokio::time::timeout(Duration::from_secs(5), rx).await { |
| 194 | Ok(Ok(result)) => answer(result), |
| 195 | _ => StatusCode::SERVICE_UNAVAILABLE.into_response(), |
| 196 | } |
| 197 | } |
| 198 | |
| 199 | async fn frame( |
| 200 | State(state): State<Service>, |
| 201 | headers: HeaderMap, |
| 202 | axum::extract::Query(query): axum::extract::Query<BTreeMap<String, String>>, |
| 203 | ) -> Response { |
| 204 | if !authorized(&headers, &state) { |
| 205 | return StatusCode::UNAUTHORIZED.into_response(); |
| 206 | } |
| 207 | let Ok(frames) = state.frames.lock() else { |
| 208 | return StatusCode::SERVICE_UNAVAILABLE.into_response(); |
| 209 | }; |
| 210 | let found = if let Some(tick) = query.get("tick") { |
| 211 | let wanted = tick.parse::<u64>().ok(); |
| 212 | frames |
| 213 | .iter() |
| 214 | .find(|f| wanted.is_some() && f["tick"].as_u64() == wanted) |
| 215 | } else { |
| 216 | frames.back() |
| 217 | }; |
| 218 | found.map_or_else( |
| 219 | || StatusCode::NOT_FOUND.into_response(), |
| 220 | |v| Json(v.clone()).into_response(), |
| 221 | ) |
| 222 | } |
| 223 | |
| 224 | async fn action( |
| 225 | State(state): State<Service>, |
| 226 | headers: HeaderMap, |
| 227 | Json(request): Json<Request>, |
| 228 | ) -> Response { |
| 229 | if !authorized(&headers, &state) { |
| 230 | return StatusCode::UNAUTHORIZED.into_response(); |
| 231 | } |
| 232 | submit(&state, |reply| Work::Action(request, reply)).await |
| 233 | } |
| 234 | async fn producer( |
| 235 | State(state): State<Service>, |
| 236 | headers: HeaderMap, |
| 237 | Json(request): Json<Producer>, |
| 238 | ) -> Response { |
| 239 | if !authorized(&headers, &state) { |
| 240 | return StatusCode::UNAUTHORIZED.into_response(); |
| 241 | } |
| 242 | submit(&state, |reply| Work::Producer(request, reply)).await |
| 243 | } |
| 244 | async fn audio( |
| 245 | State(state): State<Service>, |
| 246 | headers: HeaderMap, |
| 247 | Json(request): Json<AudioLease>, |
| 248 | ) -> Response { |
| 249 | if !authorized(&headers, &state) { |
| 250 | return StatusCode::UNAUTHORIZED.into_response(); |
| 251 | } |
| 252 | submit(&state, |reply| Work::Audio(request, reply)).await |
| 253 | } |
| 254 | async fn export(State(state): State<Service>, headers: HeaderMap) -> Response { |
| 255 | if !authorized(&headers, &state) { |
| 256 | return StatusCode::UNAUTHORIZED.into_response(); |
| 257 | } |
| 258 | submit(&state, Work::Export).await |
| 259 | } |
| 260 | async fn attach(State(state): State<Service>, headers: HeaderMap) -> Response { |
| 261 | if !authorized(&headers, &state) { |
| 262 | return StatusCode::UNAUTHORIZED.into_response(); |
| 263 | } |
| 264 | ( |
| 265 | [ |
| 266 | ( |
| 267 | "set-cookie", |
| 268 | format!( |
| 269 | "cw_pet={}; HttpOnly; SameSite=Strict; Path=/", |
| 270 | state.descriptor.token |
| 271 | ), |
| 272 | ), |
| 273 | ("cache-control", "no-store".into()), |
| 274 | ], |
| 275 | Json(json!({"identity":state.descriptor.identity})), |
| 276 | ) |
| 277 | .into_response() |
| 278 | } |
| 279 | |
| 280 | /// Starts only on an explicit pet command or when a view first attaches. |
| 281 | /// The lifetime lock is held across HTTP serving, ticks and all checkpoints. |
| 282 | // `pet serve` is its own console process, never inside the alt-screen: its |
| 283 | // startup line and stop reason are the operator's only output. |
| 284 | #[allow(clippy::print_stdout, clippy::print_stderr)] |
| 285 | pub fn serve(root: PathBuf, requested_port: u16) -> anyhow::Result<()> { |
| 286 | std::fs::create_dir_all(&root)?; |
| 287 | #[cfg(unix)] |
| 288 | { |
| 289 | use std::os::unix::fs::{MetadataExt, PermissionsExt}; |
| 290 | let metadata = std::fs::symlink_metadata(&root)?; |
| 291 | if !metadata.is_dir() || metadata.uid() != unsafe { libc::geteuid() } { |
| 292 | anyhow::bail!("Pet directory must belong to this user and cannot be a link"); |
| 293 | } |
| 294 | std::fs::set_permissions(&root, std::fs::Permissions::from_mode(0o700))?; |
| 295 | } |
| 296 | let lock_path = WorkspaceFile::open(&root, Path::new("owner.lock"), true)?; |
| 297 | let original = lock_path.open_update(true, false)?; |
| 298 | let mut lifetime = fd_lock::RwLock::new(original.try_clone()?); |
| 299 | let _guard = lifetime |
| 300 | .try_write() |
| 301 | .map_err(|_| anyhow::anyhow!("Another pet owner is running"))?; |
| 302 | let mut store = Store::at(&root)?; |
| 303 | let previous = store.load()?; |
| 304 | let mut saved = if let Some(text) = previous { |
| 305 | let value: Saved = serde_json::from_str(&text)?; |
| 306 | if value.version != 1 |
| 307 | || uuid::Uuid::parse_str(&value.identity).is_err() |
| 308 | || value.token.len() != 64 |
| 309 | || !value.token.bytes().all(|b| b.is_ascii_hexdigit()) |
| 310 | || !valid_source(&value.source) |
| 311 | || value.clients.len() > MAX_CLIENTS |
| 312 | || !value.appearance.valid() |
| 313 | { |
| 314 | anyhow::bail!("Invalid shared habitat; the existing file was kept"); |
| 315 | } |
| 316 | value |
| 317 | } else { |
| 318 | Saved { |
| 319 | version: 1, |
| 320 | identity: uuid::Uuid::new_v4().to_string(), |
| 321 | token: format!( |
| 322 | "{}{}", |
| 323 | uuid::Uuid::new_v4().simple(), |
| 324 | uuid::Uuid::new_v4().simple() |
| 325 | ), |
| 326 | port: requested_port, |
| 327 | source: "unattached".into(), |
| 328 | source_revision: 0, |
| 329 | cursor: 0, |
| 330 | clients: BTreeMap::new(), |
| 331 | recording: Value::Null, |
| 332 | appearance: Appearance::default(), |
| 333 | } |
| 334 | }; |
| 335 | let listener = std::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, saved.port))?; |
| 336 | saved.port = listener.local_addr()?.port(); |
| 337 | listener.set_nonblocking(true)?; |
| 338 | let descriptor = Descriptor { |
| 339 | version: 1, |
| 340 | port: saved.port, |
| 341 | token: saved.token.clone(), |
| 342 | identity: saved.identity.clone(), |
| 343 | }; |
| 344 | let connection = WorkspaceFile::open(&root, Path::new("connection.json"), true)?; |
| 345 | let epoch = uuid::Uuid::new_v4().to_string(); |
| 346 | let (tx, rx) = mpsc::sync_channel(128); |
| 347 | let frames = Arc::new(Mutex::new(VecDeque::new())); |
| 348 | let service = Service { |
| 349 | descriptor: descriptor.clone(), |
| 350 | frames: frames.clone(), |
| 351 | tx, |
| 352 | }; |
| 353 | let (ready_tx, ready_rx) = mpsc::sync_channel(1); |
| 354 | let (ended_tx, ended_rx) = tokio::sync::oneshot::channel(); |
| 355 | let owner = std::thread::Builder::new() |
| 356 | .name("pet-owner-world".into()) |
| 357 | .spawn(move || { |
| 358 | let result = run_world( |
| 359 | saved, store, rx, frames, &epoch, &lock_path, &original, ready_tx, |
| 360 | ); |
| 361 | if let Err(error) = result { |
| 362 | eprintln!("Shared pet stopped: {error}"); |
| 363 | } |
| 364 | let _ = ended_tx.send(()); |
| 365 | })?; |
| 366 | ready_rx |
| 367 | .recv_timeout(Duration::from_secs(15))? |
| 368 | .map_err(anyhow::Error::msg)?; |
| 369 | connection.replace(&serde_json::to_vec(&descriptor)?)?; |
| 370 | let router = Router::new() |
| 371 | .route("/", get(|| async { Html(include_str!("shared.html")) })) |
| 372 | .route("/pet-native.js", get(|| async { ([("content-type", "text/javascript")], include_str!("pet-native.js")) })) |
| 373 | .route("/whale-points.tsv", get(|| async { include_str!("../ambient_life/whale-points.tsv") })) |
| 374 | .route("/v1/frame", get(frame)).route("/v1/action", post(action)) |
| 375 | .route("/v1/producer", post(producer)).route("/v1/audio", post(audio)) |
| 376 | .route("/v1/export", get(export)).route("/v1/attach", post(attach)) |
| 377 | .layer(DefaultBodyLimit::max(64 * 1024)) |
| 378 | .layer(axum::middleware::from_fn(|request: axum::extract::Request, next: axum::middleware::Next| async move { |
| 379 | let mut response = next.run(request).await; |
| 380 | for (key, value) in [("cache-control", "no-store"), ("referrer-policy", "no-referrer"), ("x-content-type-options", "nosniff"), ("content-security-policy", "default-src 'none'; script-src 'self' 'unsafe-inline'; style-src 'unsafe-inline'; connect-src 'self'; img-src 'self' blob:; frame-ancestors 'none'; base-uri 'none'; form-action 'none'")] { |
| 381 | response.headers_mut().insert(axum::http::HeaderName::from_static(key), axum::http::HeaderValue::from_static(value)); |
| 382 | } |
| 383 | response |
| 384 | })).with_state(service); |
| 385 | println!( |
| 386 | "Shared pet {} listening on 127.0.0.1:{}", |
| 387 | descriptor.identity, descriptor.port |
| 388 | ); |
| 389 | let rt = tokio::runtime::Builder::new_multi_thread() |
| 390 | .worker_threads(2) |
| 391 | .enable_all() |
| 392 | .build()?; |
| 393 | rt.block_on(async { |
| 394 | axum::serve(tokio::net::TcpListener::from_std(listener)?, router) |
| 395 | .with_graceful_shutdown(async { |
| 396 | tokio::select! { _=tokio::signal::ctrl_c()=>{}, _=ended_rx=>{} } |
| 397 | }) |
| 398 | .await |
| 399 | })?; |
| 400 | let _ = owner.join(); |
| 401 | Ok(()) |
| 402 | } |
| 403 | |
| 404 | fn run_world( |
| 405 | mut saved: Saved, |
| 406 | mut store: Store, |
| 407 | rx: mpsc::Receiver<Work>, |
| 408 | frames: Arc<Mutex<VecDeque<Value>>>, |
| 409 | epoch: &str, |
| 410 | lock_path: &WorkspaceFile, |
| 411 | original: &std::fs::File, |
| 412 | ready: mpsc::SyncSender<Result<(), String>>, |
| 413 | ) -> anyhow::Result<()> { |
| 414 | let runtime = Runtime::new()?; |
| 415 | runtime.set_memory_limit(64 * 1024 * 1024); |
| 416 | runtime.set_max_stack_size(2 * 1024 * 1024); |
| 417 | let deadline = Arc::new(Mutex::new(Instant::now() + Duration::from_secs(10))); |
| 418 | let limit = deadline.clone(); |
| 419 | runtime.set_interrupt_handler(Some(Box::new(move || { |
| 420 | limit.lock().map_or(true, |d| Instant::now() > *d) |
| 421 | }))); |
| 422 | let context = Context::full(&runtime)?; |
| 423 | context |
| 424 | .with(|ctx| -> rquickjs::Result<()> { |
| 425 | let points: Vec<Vec<f64>> = include_str!("../ambient_life/whale-points.tsv") |
| 426 | .lines() |
| 427 | .map(|line| { |
| 428 | line.split_whitespace() |
| 429 | .filter_map(|n| n.parse().ok()) |
| 430 | .collect() |
| 431 | }) |
| 432 | .collect(); |
| 433 | ctx.globals() |
| 434 | .set("points", serde_json::to_string(&points).unwrap())?; |
| 435 | ctx.eval::<(), _>(include_bytes!("pet-native.js").as_slice())?; |
| 436 | ctx.eval::<(), _>("globalThis.pet = new PetNative(points, '', '[]', true)")?; |
| 437 | if !saved.recording.is_null() { |
| 438 | ctx.globals().set("saved", saved.recording.to_string())?; |
| 439 | ctx.eval::<(), _>( |
| 440 | "pet.restoreRecording(saved); pet.resumeEngine(); delete globalThis.saved", |
| 441 | )?; |
| 442 | } |
| 443 | Ok(()) |
| 444 | }) |
| 445 | .map_err(|_| anyhow::anyhow!("Shared habitat could not restore; its file was kept"))?; |
| 446 | let mut producer: Option<(String, u64, Instant, String)> = None; |
| 447 | let mut audio: Option<(String, Instant)> = None; |
| 448 | let mut playback: Option<(super::audio::Output, super::audio_cursor::AudioCursor)> = None; |
| 449 | let mut audio_error = false; |
| 450 | let mut waiting = false; |
| 451 | let mut last_save = Instant::now(); |
| 452 | let mut storage_error = false; |
| 453 | let origin = Instant::now(); |
| 454 | let initial_time: f64 = context.with(|ctx| ctx.eval("JSON.parse(pet.snapshot()).timeMs"))?; |
| 455 | let mut last = origin; |
| 456 | let mut ticks = 0u64; |
| 457 | let mut measurements = VecDeque::<f64>::new(); |
| 458 | save(&context, &mut saved, &mut store)?; |
| 459 | let _ = ready.send(Ok(())); |
| 460 | loop { |
| 461 | *deadline |
| 462 | .lock() |
| 463 | .map_err(|_| anyhow::anyhow!("Clock lock failed"))? = |
| 464 | Instant::now() + Duration::from_secs(5); |
| 465 | let now = Instant::now(); |
| 466 | if !same_file(&lock_path.open_update(false, false)?, original)? { |
| 467 | anyhow::bail!("Owner lock was replaced"); |
| 468 | } |
| 469 | if producer |
| 470 | .as_ref() |
| 471 | .is_some_and(|(_, _, seen, _)| now.duration_since(*seen) > LEASE) |
| 472 | { |
| 473 | producer = None; |
| 474 | waiting = false; |
| 475 | context.with(|ctx| ctx.eval::<(), _>("pet.disconnectEngine()"))?; |
| 476 | } |
| 477 | if audio |
| 478 | .as_ref() |
| 479 | .is_some_and(|(_, seen)| now.duration_since(*seen) > LEASE) |
| 480 | { |
| 481 | audio = None; |
| 482 | } |
| 483 | let elapsed = now.duration_since(last).as_secs_f64(); |
| 484 | if elapsed >= 1.0 / 30.0 { |
| 485 | // A suspended machine advances a bounded amount and marks a gap; |
| 486 | // offline wall time never invents activity or historical sound. |
| 487 | let count = (elapsed * 30.0).floor().min(3.0) as u64; |
| 488 | if elapsed > 0.25 { |
| 489 | producer = None; |
| 490 | waiting = false; |
| 491 | context.with(|ctx| ctx.eval::<(), _>("pet.disconnectEngine()"))?; |
| 492 | } |
| 493 | let started = Instant::now(); |
| 494 | context.with(|ctx| -> rquickjs::Result<()> { |
| 495 | ctx.globals() |
| 496 | .set("at", initial_time + (ticks + count) as f64 * 1000.0 / 30.0)?; |
| 497 | ctx.globals().set("waiting", waiting)?; |
| 498 | ctx.eval::<(), _>("pet.advanceEngine(at,true,waiting)") |
| 499 | })?; |
| 500 | ticks += count; |
| 501 | if audio.is_none() { |
| 502 | playback = None; |
| 503 | } |
| 504 | if audio.is_some() && playback.is_none() { |
| 505 | match super::audio::Output::start() { |
| 506 | Ok(output) => { |
| 507 | let cursor = super::audio_cursor::AudioCursor::new( |
| 508 | output.target(), |
| 509 | initial_time + ticks as f64 * 1000.0 / 30.0, |
| 510 | ); |
| 511 | playback = Some((output, cursor)); |
| 512 | audio_error = false; |
| 513 | } |
| 514 | Err(_) => { |
| 515 | audio = None; |
| 516 | audio_error = true; |
| 517 | } |
| 518 | } |
| 519 | } |
| 520 | if let Some((output, cursor)) = &mut playback { |
| 521 | let target = output.target(); |
| 522 | if output.failed() |
| 523 | || context |
| 524 | .with(|ctx| { |
| 525 | cursor.present( |
| 526 | &ctx, |
| 527 | &target, |
| 528 | initial_time + ticks as f64 * 1000.0 / 30.0, |
| 529 | ) |
| 530 | }) |
| 531 | .is_err() |
| 532 | { |
| 533 | context.with(|ctx| { |
| 534 | let _ = ctx.catch(); |
| 535 | }); |
| 536 | playback = None; |
| 537 | audio = None; |
| 538 | audio_error = true; |
| 539 | } |
| 540 | } |
| 541 | last = if elapsed > 0.25 { |
| 542 | now |
| 543 | } else { |
| 544 | last + Duration::from_secs_f64(count as f64 / 30.0) |
| 545 | }; |
| 546 | let text: String = context.with(|ctx| ctx.eval("pet.presentation()"))?; |
| 547 | let mut frame: Value = serde_json::from_str(&text)?; |
| 548 | frame["version"] = json!(1); |
| 549 | frame["identity"] = json!(saved.identity); |
| 550 | frame["epoch"] = json!(epoch); |
| 551 | frame["tick"] = |
| 552 | json!((frame["timeMs"].as_f64().unwrap_or(0.0) * 30.0 / 1000.0).round() as u64); |
| 553 | frame["cursor"] = json!(saved.cursor); |
| 554 | frame["source"] = json!(saved.source); |
| 555 | frame["sourceRevision"] = json!(saved.source_revision); |
| 556 | // Presentation material keeps missing coverage legible. The core |
| 557 | // pigment, score, particle digest and recording are unchanged. |
| 558 | for key in ["", "still"] { |
| 559 | let pose = if key.is_empty() { |
| 560 | &mut frame |
| 561 | } else { |
| 562 | &mut frame[key] |
| 563 | }; |
| 564 | let hollow = |
| 565 | producer.is_none() || pose["style"]["hollow"].as_bool().unwrap_or(true); |
| 566 | if hollow { |
| 567 | pose["style"]["hollow"] = json!(true); |
| 568 | pose["style"]["r"] = json!(153); |
| 569 | pose["style"]["g"] = json!(176); |
| 570 | pose["style"]["b"] = json!(184); |
| 571 | pose["style"]["alpha"] = |
| 572 | json!(0.68 * pose["state"]["lit"].as_f64().unwrap_or(1.0).max(0.25)); |
| 573 | } |
| 574 | } |
| 575 | for key in ["", "still"] { |
| 576 | let pose = if key.is_empty() { |
| 577 | &mut frame |
| 578 | } else { |
| 579 | &mut frame[key] |
| 580 | }; |
| 581 | if !saved.appearance.event_colors { |
| 582 | for (key, value) in ["r", "g", "b"].into_iter().zip(saved.appearance.particle) { |
| 583 | pose["style"][key] = json!(value); |
| 584 | } |
| 585 | } |
| 586 | pose["style"]["alpha"] = json!( |
| 587 | (pose["style"]["alpha"].as_f64().unwrap_or(0.5) * saved.appearance.brightness) |
| 588 | .clamp(0.0, 1.0) |
| 589 | ); |
| 590 | } |
| 591 | frame["appearance"] = json!(saved.appearance); |
| 592 | frame["producerConnected"] = json!(producer.is_some()); |
| 593 | frame["storageAvailable"] = json!(!storage_error); |
| 594 | frame["audioOwner"] = json!(audio.as_ref().map(|(id, _)| id)); |
| 595 | frame["audioUnavailable"] = json!(audio_error); |
| 596 | measurements.push_back(started.elapsed().as_secs_f64() * 1000.0); |
| 597 | if measurements.len() > 300 { |
| 598 | measurements.pop_front(); |
| 599 | } |
| 600 | frame["performance"] = json!({"worldHz":30,"frames":ticks,"uptimeSeconds":origin.elapsed().as_secs_f64(),"workMs":measurements.back()}); |
| 601 | let mut output = frames |
| 602 | .lock() |
| 603 | .map_err(|_| anyhow::anyhow!("Frame lock failed"))?; |
| 604 | output.push_back(frame); |
| 605 | if output.len() > 16 { |
| 606 | output.pop_front(); |
| 607 | } |
| 608 | } |
| 609 | if last_save.elapsed() >= Duration::from_secs(1) { |
| 610 | storage_error = save(&context, &mut saved, &mut store).is_err(); |
| 611 | last_save = Instant::now(); |
| 612 | } |
| 613 | let work = match rx.recv_timeout(Duration::from_millis(2)) { |
| 614 | Ok(work) => work, |
| 615 | Err(mpsc::RecvTimeoutError::Timeout) => continue, |
| 616 | Err(mpsc::RecvTimeoutError::Disconnected) => { |
| 617 | save(&context, &mut saved, &mut store)?; |
| 618 | return Ok(()); |
| 619 | } |
| 620 | }; |
| 621 | match work { |
| 622 | Work::Export(reply) => { |
| 623 | let result = context |
| 624 | .with(|ctx| super::persistence::export_recording(&ctx, true)) |
| 625 | .map_err(|_| "Recording export failed".to_owned()) |
| 626 | .and_then(|text| { |
| 627 | serde_json::from_slice(&text).map_err(|_| "Recording export failed".into()) |
| 628 | }); |
| 629 | let _ = reply.send(result); |
| 630 | } |
| 631 | Work::Audio(request, reply) => { |
| 632 | let result = if !valid_client(&request.client) { |
| 633 | Err("Invalid view identity".into()) |
| 634 | } else if !request.enabled { |
| 635 | if audio.as_ref().is_some_and(|(id, _)| *id == request.client) { |
| 636 | audio = None; |
| 637 | } |
| 638 | Ok(json!({"granted":false})) |
| 639 | } else if audio.as_ref().is_none_or(|(id, _)| *id == request.client) { |
| 640 | audio = Some((request.client, Instant::now())); |
| 641 | Ok(json!({"granted":true})) |
| 642 | } else { |
| 643 | Ok(json!({"granted":false})) |
| 644 | }; |
| 645 | let _ = reply.send(result); |
| 646 | } |
| 647 | Work::Producer(request, reply) => { |
| 648 | let result = (|| -> Result<Value, String> { |
| 649 | if request.identity != saved.identity |
| 650 | || request.epoch != epoch |
| 651 | || !valid_client(&request.client) |
| 652 | || request.source != saved.source |
| 653 | || request.source_revision != saved.source_revision |
| 654 | || request.events.len() > 64 |
| 655 | { |
| 656 | return Err( |
| 657 | "Source changed; attach at the current frame and discard stale input" |
| 658 | .into(), |
| 659 | ); |
| 660 | } |
| 661 | use sha2::{Digest, Sha256}; |
| 662 | let hash=Sha256::digest(serde_json::to_vec(&json!({"seq":request.seq,"events":request.events,"waiting":request.waiting})).unwrap()).iter().map(|b|format!("{b:02x}")).collect::<String>(); |
| 663 | if let Some((id, seq, _, prior_hash)) = &producer { |
| 664 | if *id != request.client { |
| 665 | return Err("This source already has a producer".into()); |
| 666 | } |
| 667 | if request.seq == *seq && *prior_hash == hash { |
| 668 | return Ok( |
| 669 | json!({"seq":seq,"cursor":saved.cursor,"duplicate":true,"durable":false}), |
| 670 | ); |
| 671 | } |
| 672 | if request.seq != seq + 1 { |
| 673 | producer = None; |
| 674 | waiting = false; |
| 675 | context |
| 676 | .with(|ctx| ctx.eval::<(), _>("pet.disconnectEngine()")) |
| 677 | .map_err(|_| "Coverage reset failed")?; |
| 678 | return Err("Producer gap; reconnect without historical input".into()); |
| 679 | } |
| 680 | } else if request.seq != 0 || !request.events.is_empty() { |
| 681 | return Err( |
| 682 | "Begin a producer lease with sequence zero and no historical events" |
| 683 | .into(), |
| 684 | ); |
| 685 | } |
| 686 | context |
| 687 | .with(|ctx| -> rquickjs::Result<()> { |
| 688 | ctx.globals() |
| 689 | .set("events", serde_json::to_string(&request.events).unwrap())?; |
| 690 | ctx.eval::<(), _>( |
| 691 | "pet.observeEngineBatch(events, JSON.parse(pet.snapshot()).timeMs)", |
| 692 | ) |
| 693 | }) |
| 694 | .map_err(|_| { |
| 695 | context.with(|ctx| { |
| 696 | let _ = ctx.catch(); |
| 697 | }); |
| 698 | "Invalid Engine metadata" |
| 699 | })?; |
| 700 | if !request.events.is_empty() || waiting != request.waiting { |
| 701 | saved.cursor += 1; |
| 702 | } |
| 703 | producer = Some((request.client, request.seq, Instant::now(), hash)); |
| 704 | waiting = request.waiting; |
| 705 | Ok(json!({"seq":request.seq,"cursor":saved.cursor,"durable":false})) |
| 706 | })(); |
| 707 | let _ = reply.send(result); |
| 708 | } |
| 709 | Work::Action(request, reply) => { |
| 710 | let result = (|| -> Result<Value, String> { |
| 711 | use sha2::{Digest, Sha256}; |
| 712 | if request.identity != saved.identity |
| 713 | || !valid_client(&request.client) |
| 714 | || request.seq == 0 |
| 715 | { |
| 716 | return Err("Invalid pet action identity".into()); |
| 717 | } |
| 718 | let hash = Sha256::digest(serde_json::to_vec(&request).unwrap()) |
| 719 | .iter() |
| 720 | .map(|b| format!("{b:02x}")) |
| 721 | .collect::<String>(); |
| 722 | if let Some(receipt) = saved.clients.get(&request.client) { |
| 723 | if request.seq == receipt.seq && hash == receipt.hash { |
| 724 | return Ok(json!({"cursor":receipt.cursor,"duplicate":true})); |
| 725 | } |
| 726 | if request.seq != receipt.seq + 1 { |
| 727 | return Err("Action sequence is stale or has a gap".into()); |
| 728 | } |
| 729 | } else if request.seq != 1 || saved.clients.len() >= MAX_CLIENTS { |
| 730 | return Err( |
| 731 | "Action client is unknown or the retained client limit was reached" |
| 732 | .into(), |
| 733 | ); |
| 734 | } |
| 735 | if request.source_revision != saved.source_revision { |
| 736 | return Err( |
| 737 | "Source changed; review the current source before interacting".into(), |
| 738 | ); |
| 739 | } |
| 740 | // Check storage before accepting an action. A failed commit |
| 741 | // is rolled back together with its idempotency receipt. |
| 742 | save(&context, &mut saved, &mut store) |
| 743 | .map_err(|_| "Pet storage unavailable; action was not accepted")?; |
| 744 | let before = serde_json::to_string(&saved).unwrap(); |
| 745 | match request.action { |
| 746 | Action::Appearance { appearance } => { |
| 747 | if !appearance.valid() { |
| 748 | return Err("Invalid appearance range".into()); |
| 749 | } |
| 750 | saved.appearance = appearance; |
| 751 | } |
| 752 | Action::Interact { food, x, y } => { |
| 753 | if !x.is_finite() || !y.is_finite() || x.abs() > 1.0 || y.abs() > 1.0 { |
| 754 | return Err("Invalid interaction coordinates".into()); |
| 755 | } |
| 756 | context |
| 757 | .with(|ctx| -> rquickjs::Result<()> { |
| 758 | ctx.globals().set("x", x)?; |
| 759 | ctx.globals().set("y", y)?; |
| 760 | ctx.globals() |
| 761 | .set("kind", if food { "food" } else { "attention" })?; |
| 762 | ctx.eval::<(), _>("pet.interact(kind,x,y)") |
| 763 | }) |
| 764 | .map_err(|_| "Interaction failed")?; |
| 765 | } |
| 766 | Action::Select { source } => { |
| 767 | if !valid_source(&source) { |
| 768 | return Err("Invalid source identity".into()); |
| 769 | } |
| 770 | saved.source = source; |
| 771 | saved.source_revision += 1; |
| 772 | producer = None; |
| 773 | waiting = false; |
| 774 | context |
| 775 | .with(|ctx| ctx.eval::<(), _>("pet.disconnectEngine()")) |
| 776 | .map_err(|_| "Source disconnect failed")?; |
| 777 | } |
| 778 | } |
| 779 | saved.cursor += 1; |
| 780 | saved.clients.insert( |
| 781 | request.client, |
| 782 | Receipt { |
| 783 | seq: request.seq, |
| 784 | hash, |
| 785 | cursor: saved.cursor, |
| 786 | }, |
| 787 | ); |
| 788 | if save(&context, &mut saved, &mut store).is_err() { |
| 789 | saved = serde_json::from_str(&before).unwrap(); |
| 790 | context.with(|ctx| -> rquickjs::Result<()> { ctx.globals().set("rollback",saved.recording.to_string())?;ctx.eval::<(),_>("pet.restoreRecording(rollback); pet.disconnectEngine(); delete globalThis.rollback") }).map_err(|_| "Storage rollback failed")?; |
| 791 | producer = None; |
| 792 | waiting = false; |
| 793 | storage_error = true; |
| 794 | return Err("Pet storage unavailable; action was not accepted".into()); |
| 795 | } |
| 796 | storage_error = false; |
| 797 | last_save = Instant::now(); |
| 798 | Ok(json!({"cursor":saved.cursor,"duplicate":false})) |
| 799 | })(); |
| 800 | let _ = reply.send(result); |
| 801 | } |
| 802 | } |
| 803 | } |
| 804 | } |
| 805 | |
| 806 | fn save(context: &Context, saved: &mut Saved, store: &mut Store) -> anyhow::Result<()> { |
| 807 | let (text, segment) = context.with(|ctx| -> rquickjs::Result<(String, bool)> { |
| 808 | let segment: bool = ctx.eval("pet.needsSegment()")?; |
| 809 | Ok(( |
| 810 | ctx.eval(if segment { |
| 811 | "pet.prepareSegment()" |
| 812 | } else { |
| 813 | "pet.recording(true)" |
| 814 | })?, |
| 815 | segment, |
| 816 | )) |
| 817 | })?; |
| 818 | saved.recording = serde_json::from_str(&text)?; |
| 819 | let archive = |
| 820 | if segment { |
| 821 | Some(context.with(|ctx| { |
| 822 | ctx.eval::<String, _>("JSON.stringify(JSON.parse(pet.recording(true)))") |
| 823 | })?) |
| 824 | } else { |
| 825 | None |
| 826 | }; |
| 827 | let tick = saved.recording["checkpoint"]["tick"].as_u64().unwrap_or(0); |
| 828 | store.save_archived( |
| 829 | &serde_json::to_string(saved)?, |
| 830 | archive.as_ref().map(|a| (a.as_bytes(), tick)), |
| 831 | )?; |
| 832 | if segment { |
| 833 | context.with(|ctx| ctx.eval::<(), _>("pet.commitSegment()"))?; |
| 834 | } |
| 835 | Ok(()) |
| 836 | } |
| 837 |