From 5eb442b36e1aba5cc79600d3ad1eaebd8074f946 Mon Sep 17 00:00:00 2001 From: andre Date: Sun, 18 Oct 2026 21:16:29 -0400 Subject: [PATCH] feat(sim): discovery file and simulation start options for local companions --- README.md | 2 + backend/src/api/live_simulation.rs | 45 +++++- backend/src/api/mod.rs | 4 + backend/src/api/tests.rs | 91 +++++++++++ backend/src/live_simulation.rs | 245 +++++++++++++++++++++++++++-- desktop-gpui/src/embed.rs | 102 ++++++++++++ desktop-gpui/src/main.rs | 4 + shared/contracts/replay-api.md | 11 ++ 8 files changed, 485 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index 78819e9..e75223f 100644 --- a/README.md +++ b/README.md @@ -40,6 +40,8 @@ Downloaded sessions, caches, logs, and a user-editable `.env` file live here: | macOS | `~/Library/Application Support/interval/data` | | Linux | `~/.local/share/interval/data` | +While the app runs, `server.json` in this directory records the embedded backend's loopback address (`url`), process id and start time, so local companion tools can find it. It is removed on exit; a stale copy after a crash fails `GET /healthz`. The backend only ever listens on `127.0.0.1`. + Set `INTERVAL_DESKTOP_DATA_DIR` in your environment to use another directory. For other settings, edit `.env` there and restart the app. Environment variables take precedence over `.env`. | Variable | Purpose | diff --git a/backend/src/api/live_simulation.rs b/backend/src/api/live_simulation.rs index 79163cc..8c783da 100644 --- a/backend/src/api/live_simulation.rs +++ b/backend/src/api/live_simulation.rs @@ -1,5 +1,9 @@ use super::{ApiError, AppState}; -use crate::{domain::ReplaySnapshot, replay, storage}; +use crate::{ + domain::ReplaySnapshot, + live_simulation::{self, SimulationError, StartOptions}, + replay, storage, +}; use axum::{ extract::{Path, Query, State}, response::sse::{Event, KeepAlive, Sse}, @@ -14,22 +18,53 @@ pub struct LiveSimulationStreamQuery { speed: Option, } +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct SpeedBody { + speed: f64, +} + +impl From for ApiError { + fn from(error: SimulationError) -> Self { + match error { + SimulationError::InvalidRequest(message) => ApiError::BadRequest(message), + SimulationError::NotCached => ApiError::NotFound, + SimulationError::Storage(error) => ApiError::Storage(error), + } + } +} + pub async fn start( State(state): State, Path(session_key): Path, + body: Option>, ) -> Result, ApiError> { if replay::metadata(&state.pool, session_key).await?.is_none() { return Err(ApiError::NotFound); } + let options = body.map_or_else(StartOptions::default, |Json(options)| options); Ok(Json( state .live_simulation - .start(&state.pool, session_key) - .await - .map_err(ApiError::Storage)?, + .start(&state.pool, session_key, options) + .await?, )) } +pub async fn speed( + State(state): State, + Path(session_key): Path, + Json(body): Json, +) -> Result, ApiError> { + let speed = live_simulation::normalize_speed(body.speed)?; + state + .live_simulation + .set_speed(session_key, speed) + .await + .map(Json) + .ok_or(ApiError::NotFound) +} + pub async fn status( State(state): State, Path(session_key): Path, @@ -78,7 +113,7 @@ pub async fn stream( .speed .filter(|value| value.is_finite() && *value > 0.0) .unwrap_or(1.0) - .clamp(0.25, 1_000.0); + .clamp(live_simulation::MIN_SPEED, live_simulation::MAX_SPEED); state.live_simulation.set_speed(session_key, speed).await; let frame_delay = Duration::from_secs_f64((frame_step / speed).clamp(0.01, 2.0)); let replay_events = storage::get_replay_events(&state.pool, session_key).await?; diff --git a/backend/src/api/mod.rs b/backend/src/api/mod.rs index 5cc6cc9..7811d53 100644 --- a/backend/src/api/mod.rs +++ b/backend/src/api/mod.rs @@ -154,6 +154,10 @@ pub fn router(state: AppState) -> Router { "/api/sessions/{session_key}/live-simulation/start", post(live_simulation::start), ) + .route( + "/api/sessions/{session_key}/live-simulation/speed", + post(live_simulation::speed), + ) .route( "/api/sessions/{session_key}/live-simulation/status", get(live_simulation::status), diff --git a/backend/src/api/tests.rs b/backend/src/api/tests.rs index 5b528c4..bb3b7c4 100644 --- a/backend/src/api/tests.rs +++ b/backend/src/api/tests.rs @@ -1944,6 +1944,75 @@ async fn live_simulation_start_status_snapshot_and_stop_use_cached_replay() { assert_eq!(stopped["active"], false); } +#[tokio::test] +async fn live_simulation_start_accepts_lap_and_speed_and_speed_route_rebases() { + let app = seeded_router().await; + + let start = post_json_body( + app.clone(), + "/api/sessions/9839/live-simulation/start", + json!({ "start_lap": 3, "speed": 4.0 }), + StatusCode::OK, + ) + .await; + assert_eq!(start["active"], true); + assert!(start["current_t"].as_f64().unwrap() >= 120.0); + + let snapshot = get_json( + app.clone(), + "/api/sessions/9839/live-simulation/snapshot", + StatusCode::OK, + ) + .await; + assert_eq!(snapshot["race_state"]["lap"], 3); + + let status = post_json_body( + app.clone(), + "/api/sessions/9839/live-simulation/speed", + json!({ "speed": 1.0 }), + StatusCode::OK, + ) + .await; + assert_eq!(status["active"], true); + assert!(status["current_t"].as_f64().unwrap() >= 120.0); + + post_json_body( + app.clone(), + "/api/sessions/9839/live-simulation/speed", + json!({ "speed": -1.0 }), + StatusCode::BAD_REQUEST, + ) + .await; + post_json_body( + app.clone(), + "/api/sessions/9839/live-simulation/start", + json!({ "start_t": 0.0, "start_lap": 1 }), + StatusCode::BAD_REQUEST, + ) + .await; + post_json_body( + app.clone(), + "/api/sessions/9839/live-simulation/start", + json!({ "start_lap": 99 }), + StatusCode::BAD_REQUEST, + ) + .await; + + post_json( + app.clone(), + "/api/sessions/9839/live-simulation/stop", + StatusCode::OK, + ) + .await; + post_json_body( + app, + "/api/sessions/9839/live-simulation/speed", + json!({ "speed": 2.0 }), + StatusCode::NOT_FOUND, + ) + .await; +} + #[tokio::test] async fn live_simulation_requires_cached_replay() { let app = seeded_router().await; @@ -2411,6 +2480,28 @@ async fn get_json(app: Router, uri: &str, expected_status: StatusCode) -> Value serde_json::from_slice(&body).unwrap() } +async fn post_json_body( + app: Router, + uri: &str, + body: Value, + expected_status: StatusCode, +) -> Value { + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri(uri) + .header("content-type", "application/json") + .body(Body::from(body.to_string())) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), expected_status); + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + serde_json::from_slice(&body).unwrap() +} + async fn post_json(app: Router, uri: &str, expected_status: StatusCode) -> Value { let response = app .oneshot( diff --git a/backend/src/live_simulation.rs b/backend/src/live_simulation.rs index da46973..c0e6763 100644 --- a/backend/src/live_simulation.rs +++ b/backend/src/live_simulation.rs @@ -2,10 +2,49 @@ use crate::{ domain::{DataSource, EndpointLinks, LiveSessionStatus, ReplayMetadata, ReplaySnapshot}, replay, }; +use serde::Deserialize; use sqlx::SqlitePool; use std::{collections::HashMap, sync::Arc, time::Instant}; use tokio::sync::Mutex; +pub const MIN_SPEED: f64 = 0.25; +pub const MAX_SPEED: f64 = 1_000.0; + +/// Optional start parameters. Absent fields keep the original behavior: start at +/// `race_start_t` (or `min_t`) at 1x. `start_t` and `start_lap` are mutually exclusive. +#[derive(Debug, Clone, Copy, Default, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct StartOptions { + pub start_t: Option, + pub start_lap: Option, + pub speed: Option, +} + +#[derive(Debug, thiserror::Error)] +pub enum SimulationError { + #[error("{0}")] + InvalidRequest(String), + #[error("cached replay is required")] + NotCached, + #[error(transparent)] + Storage(#[from] anyhow::Error), +} + +impl From for SimulationError { + fn from(error: sqlx::Error) -> Self { + Self::Storage(error.into()) + } +} + +pub fn normalize_speed(speed: f64) -> Result { + if !speed.is_finite() || speed <= 0.0 { + return Err(SimulationError::InvalidRequest(format!( + "speed must be a positive number, got {speed}" + ))); + } + Ok(speed.clamp(MIN_SPEED, MAX_SPEED)) +} + #[derive(Clone, Default)] pub struct LiveSimulationRegistry { sessions: Arc>>, @@ -26,14 +65,25 @@ impl LiveSimulationRegistry { &self, pool: &SqlitePool, session_key: i64, - ) -> anyhow::Result { + options: StartOptions, + ) -> Result { let metadata = replay::metadata(pool, session_key) .await? - .ok_or_else(|| anyhow::anyhow!("cached replay metadata is required"))?; - let start_t = simulation_start_t(&metadata); + .ok_or(SimulationError::NotCached)?; + let speed = normalize_speed(options.speed.unwrap_or(1.0))?; + let start_t = match (options.start_t, options.start_lap) { + (Some(_), Some(_)) => { + return Err(SimulationError::InvalidRequest( + "start_t and start_lap are mutually exclusive".to_string(), + )); + } + (Some(t), None) => clamp_start_t(&metadata, t)?, + (None, Some(lap)) => lap_start_t(pool, session_key, &metadata, lap).await?, + (None, None) => simulation_start_t(&metadata), + }; replay::snapshot_at(pool, session_key, start_t) .await? - .ok_or_else(|| anyhow::anyhow!("cached replay snapshots are required"))?; + .ok_or(SimulationError::NotCached)?; let simulation = LiveSimulation { session_key, @@ -41,7 +91,7 @@ impl LiveSimulationRegistry { started_at: chrono::Utc::now().to_rfc3339(), min_t: start_t, max_t: metadata.max_t, - speed: 1.0, + speed, }; let status = simulation.status(); self.sessions.lock().await.insert(session_key, simulation); @@ -74,13 +124,16 @@ impl LiveSimulationRegistry { .unwrap_or_else(|| inactive_status(session_key)) } - pub async fn set_speed(&self, session_key: i64, speed: f64) { - if let Some(simulation) = self.sessions.lock().await.get_mut(&session_key) { - let current_t = simulation.current_t(); - simulation.min_t = current_t; - simulation.started_instant = Instant::now(); - simulation.speed = speed.max(0.001); - } + /// Rebases the clock so the current simulation time is continuous across the change. + /// Returns `None` when no simulation is running for the session. + pub async fn set_speed(&self, session_key: i64, speed: f64) -> Option { + let mut sessions = self.sessions.lock().await; + let simulation = sessions.get_mut(&session_key)?; + let current_t = simulation.current_t(); + simulation.min_t = current_t; + simulation.started_instant = Instant::now(); + simulation.speed = speed.max(0.001); + Some(simulation.status()) } pub async fn current_snapshot( @@ -150,6 +203,54 @@ fn inactive_status(session_key: i64) -> LiveSessionStatus { } } +fn clamp_start_t(metadata: &ReplayMetadata, t: f64) -> Result { + if !t.is_finite() { + return Err(SimulationError::InvalidRequest(format!( + "start_t must be a finite number, got {t}" + ))); + } + let last_frame = (metadata.max_t - metadata.frame_step_seconds).max(metadata.min_t); + Ok(t.clamp(metadata.min_t, last_frame)) +} + +/// Earliest persisted frame whose leader lap has reached `lap`, found by bisecting the +/// frame index so a long replay costs a handful of snapshot reads instead of a full scan. +async fn lap_start_t( + pool: &SqlitePool, + session_key: i64, + metadata: &ReplayMetadata, + lap: i32, +) -> Result { + if lap < 1 { + return Err(SimulationError::InvalidRequest(format!( + "start_lap must be at least 1, got {lap}" + ))); + } + let step = metadata.frame_step_seconds.max(0.001); + let frame_t = |frame: i64| metadata.min_t + frame as f64 * step; + let lap_at = |frame: i64| async move { + replay::snapshot_at(pool, session_key, frame_t(frame)) + .await + .map(|snapshot| snapshot.map_or(0, |snapshot| snapshot.race_state.lap)) + }; + let last = ((metadata.max_t - metadata.min_t) / step).floor().max(0.0) as i64; + if lap_at(last).await? < lap { + return Err(SimulationError::InvalidRequest(format!( + "start_lap {lap} is beyond the cached replay" + ))); + } + let (mut lo, mut hi) = (0, last); + while lo < hi { + let mid = lo + (hi - lo) / 2; + if lap_at(mid).await? >= lap { + hi = mid; + } else { + lo = mid + 1; + } + } + Ok(frame_t(lo)) +} + fn simulation_start_t(metadata: &ReplayMetadata) -> f64 { if metadata.race_start_t.is_finite() && metadata.race_start_t >= metadata.min_t @@ -173,7 +274,10 @@ mod tests { storage::seed_demo_session(&pool).await.unwrap(); let registry = LiveSimulationRegistry::default(); - let status = registry.start(&pool, 9839).await.unwrap(); + let status = registry + .start(&pool, 9839, StartOptions::default()) + .await + .unwrap(); let snapshot = registry .current_snapshot(&pool, 9839) .await @@ -201,7 +305,10 @@ mod tests { .unwrap(); let registry = LiveSimulationRegistry::default(); - let status = registry.start(&pool, 9839).await.unwrap(); + let status = registry + .start(&pool, 9839, StartOptions::default()) + .await + .unwrap(); let snapshot = registry .current_snapshot(&pool, 9839) .await @@ -212,6 +319,116 @@ mod tests { assert_eq!(snapshot.cursor.t, 60.0); } + async fn demo_pool() -> SqlitePool { + let pool = storage::connect("sqlite::memory:").await.unwrap(); + storage::migrate(&pool).await.unwrap(); + storage::seed_demo_session(&pool).await.unwrap(); + pool + } + + #[tokio::test] + async fn start_t_is_clamped_to_the_last_persisted_frame() { + let pool = demo_pool().await; + let registry = LiveSimulationRegistry::default(); + + let options = StartOptions { + start_t: Some(60.0), + ..StartOptions::default() + }; + let status = registry.start(&pool, 9839, options).await.unwrap(); + assert!(status.current_t.unwrap() >= 60.0); + + let options = StartOptions { + start_t: Some(10_000.0), + ..StartOptions::default() + }; + let status = registry.start(&pool, 9839, options).await.unwrap(); + assert!(status.active); + assert!(status.current_t.unwrap() >= 120.0); + + let options = StartOptions { + start_t: Some(-5.0), + ..StartOptions::default() + }; + let status = registry.start(&pool, 9839, options).await.unwrap(); + assert!(status.current_t.unwrap() >= 0.0); + assert!(status.current_t.unwrap() < 1.0); + } + + #[tokio::test] + async fn start_lap_resolves_the_first_frame_of_that_lap() { + let pool = demo_pool().await; + let registry = LiveSimulationRegistry::default(); + + let options = StartOptions { + start_lap: Some(3), + ..StartOptions::default() + }; + registry.start(&pool, 9839, options).await.unwrap(); + let snapshot = registry + .current_snapshot(&pool, 9839) + .await + .unwrap() + .unwrap(); + assert_eq!(snapshot.race_state.lap, 3); + assert_eq!(snapshot.cursor.t, 120.0); + + let options = StartOptions { + start_lap: Some(99), + ..StartOptions::default() + }; + assert!(matches!( + registry.start(&pool, 9839, options).await, + Err(SimulationError::InvalidRequest(_)) + )); + } + + #[tokio::test] + async fn start_rejects_conflicting_selectors_and_invalid_speed() { + let pool = demo_pool().await; + let registry = LiveSimulationRegistry::default(); + + let options = StartOptions { + start_t: Some(0.0), + start_lap: Some(1), + speed: None, + }; + assert!(matches!( + registry.start(&pool, 9839, options).await, + Err(SimulationError::InvalidRequest(_)) + )); + + let options = StartOptions { + speed: Some(0.0), + ..StartOptions::default() + }; + assert!(matches!( + registry.start(&pool, 9839, options).await, + Err(SimulationError::InvalidRequest(_)) + )); + assert!(!registry.status(9839).await.active); + } + + #[tokio::test] + async fn set_speed_keeps_simulation_time_continuous() { + let pool = demo_pool().await; + let registry = LiveSimulationRegistry::default(); + let options = StartOptions { + start_t: Some(60.0), + speed: Some(4.0), + ..StartOptions::default() + }; + registry.start(&pool, 9839, options).await.unwrap(); + + let before = registry.status(9839).await.current_t.unwrap(); + let status = registry.set_speed(9839, 1.0).await.unwrap(); + let after = status.current_t.unwrap(); + + assert!(after >= before); + assert!(after - before < 1.0); + assert!(registry.set_speed(42, 1.0).await.is_none()); + } + #[tokio::test] async fn inactive_status_is_empty() { let registry = LiveSimulationRegistry::default(); diff --git a/desktop-gpui/src/embed.rs b/desktop-gpui/src/embed.rs index 1fc8a93..f376f72 100644 --- a/desktop-gpui/src/embed.rs +++ b/desktop-gpui/src/embed.rs @@ -10,7 +10,9 @@ use include_dir::{Dir, include_dir}; use interval_backend::server::{self, BoundServer, ServeOptions}; +use serde::{Deserialize, Serialize}; use std::future::Future; +use std::net::SocketAddr; use std::path::{Path, PathBuf}; static TRACK_ASSETS: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../backend/assets/tracks"); @@ -192,3 +194,103 @@ pub async fn serve_embedded( ) .await } + +/// Where local companion tools (an MCP server, a test harness) find the embedded +/// backend. The port is ephemeral, so without this file nothing outside the process can +/// locate it. It holds only the loopback URL plus diagnostics, never a credential, and +/// the bind stays loopback-only. A crash leaves it behind, so readers must treat it as a +/// hint and confirm with `GET /healthz`. +pub const DISCOVERY_FILE: &str = "server.json"; + +#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)] +pub struct Discovery { + pub url: String, + pub pid: u32, + pub started_at: String, +} + +pub fn write_discovery(dir: &Path, addr: SocketAddr) -> std::io::Result<()> { + let record = Discovery { + url: format!("http://{addr}"), + pid: std::process::id(), + started_at: chrono::Utc::now().to_rfc3339(), + }; + // Stage-and-rename so a reader never sees a half-written file. + let staged = dir.join(format!("{DISCOVERY_FILE}.partial")); + std::fs::write(&staged, serde_json::to_vec_pretty(&record)?)?; + std::fs::rename(staged, dir.join(DISCOVERY_FILE)) +} + +/// Removes the file only if this process wrote it, so a newer instance's record is +/// never deleted by an older one shutting down. +pub fn remove_discovery(dir: &Path) { + let path = dir.join(DISCOVERY_FILE); + let owned = std::fs::read(&path) + .ok() + .and_then(|bytes| serde_json::from_slice::(&bytes).ok()) + .is_some_and(|record| record.pid == std::process::id()); + if owned { + let _ = std::fs::remove_file(path); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + struct TempDir(PathBuf); + + impl TempDir { + fn new(tag: &str) -> Self { + let dir = std::env::temp_dir().join(format!( + "interval-discovery-test-{}-{tag}", + std::process::id() + )); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + Self(dir) + } + } + + impl Drop for TempDir { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } + } + + fn read(dir: &Path) -> Discovery { + serde_json::from_slice(&std::fs::read(dir.join(DISCOVERY_FILE)).unwrap()).unwrap() + } + + #[test] + fn discovery_round_trips_the_bound_address_and_is_removed_on_shutdown() { + let dir = TempDir::new("roundtrip"); + let addr: SocketAddr = "127.0.0.1:53124".parse().unwrap(); + write_discovery(&dir.0, addr).unwrap(); + let record = read(&dir.0); + assert_eq!(record.url, "http://127.0.0.1:53124"); + assert_eq!(record.pid, std::process::id()); + assert!(chrono::DateTime::parse_from_rfc3339(&record.started_at).is_ok()); + assert!(!dir.0.join(format!("{DISCOVERY_FILE}.partial")).exists()); + + remove_discovery(&dir.0); + assert!(!dir.0.join(DISCOVERY_FILE).exists()); + } + + #[test] + fn remove_discovery_leaves_another_instances_record_alone() { + let dir = TempDir::new("foreign"); + let foreign = Discovery { + url: "http://127.0.0.1:1".to_string(), + pid: std::process::id().wrapping_add(1), + started_at: "2026-10-18T00:00:00Z".to_string(), + }; + std::fs::write( + dir.0.join(DISCOVERY_FILE), + serde_json::to_vec(&foreign).unwrap(), + ) + .unwrap(); + remove_discovery(&dir.0); + assert_eq!(read(&dir.0), foreign); + } +} diff --git a/desktop-gpui/src/main.rs b/desktop-gpui/src/main.rs index f5e4197..b90a8d3 100644 --- a/desktop-gpui/src/main.rs +++ b/desktop-gpui/src/main.rs @@ -631,6 +631,9 @@ fn main() { .expect("failed to start embedded backend"); let base_url = format!("http://{}", bound.addr); tracing::info!(%base_url, "embedded backend ready"); + if let Err(error) = embed::write_discovery(&data_dir, bound.addr) { + tracing::warn!(%error, "failed to write backend discovery file"); + } tracing::info!("entering gpui run loop"); application().with_assets(assets::Assets).run(move |cx: &mut App| { @@ -728,6 +731,7 @@ fn main() { // `run()` returned: the last window closed and gpui quit. Drain the server so the // sqlite pool closes cleanly, then stop the runtime without waiting on stragglers. let _ = shutdown_tx.send(()); + embed::remove_discovery(&data_dir); let _ = runtime.block_on(async { tokio::time::timeout(std::time::Duration::from_secs(2), bound.task).await }); diff --git a/shared/contracts/replay-api.md b/shared/contracts/replay-api.md index 9f86baa..b01d375 100644 --- a/shared/contracts/replay-api.md +++ b/shared/contracts/replay-api.md @@ -219,6 +219,17 @@ The live stream emits the same `metadata`, `snapshot`, `event`, `end`, and `erro Live simulation starts from `ReplayMetadata.race_start_t` when that value is present and inside replay bounds, so pre-grid and formation-lap time is skipped for normal testing. If `race_start_t` is unavailable, the simulator falls back to `min_t`. +`start` accepts an optional JSON body. Without one, behavior is unchanged. + +```json +{ "start_lap": 30, "speed": 4.0 } +``` + +- `start_t` (seconds) or `start_lap` (leader lap, resolved to the first persisted frame of that lap), never both. `start_t` is clamped to the replay bounds; a `start_lap` beyond the cached replay returns `400`. +- `speed` must be a positive number and is clamped to `[0.25, 1000]`. Default `1.0`. + +`POST /api/sessions/{session_key}/live-simulation/speed` with `{ "speed": 2.0 }` changes the speed of a running simulation and returns its status. The clock is rebased so `current_t` is continuous across the change. Returns `404` when no simulation is running and `400` for an invalid speed. Opening `/live-simulation/stream?speed=` still sets the speed as before. + ### OpenF1 Live OpenF1 live mode is enabled by default. Disable it for offline/dev runs with `INTERVAL_OPENF1_LIVE_ENABLED=false`, `0`, `no`, or `off`. Optional configuration includes `INTERVAL_OPENF1_LIVE_BASE_URL`, `INTERVAL_OPENF1_LIVE_TOKEN`, and `INTERVAL_OPENF1_LIVE_AUTH_HEADER`. With the default `Authorization` header, `INTERVAL_OPENF1_LIVE_TOKEN` accepts either a raw token such as `abc123` or a prefixed value such as `Bearer abc123`.