frank/ruvnet--RuView
1
0
Fork 0
mirror of https://github.com/ruvnet/RuView synced 2026-08-05 19:41:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into feat/ruview-auth-cognitum-oauth-verifier

Browse Source 
# Conflicts:
#	v2/crates/wifi-densepose-sensing-server/src/main.rs
This commit is contained in:
Dragan Spiridonov
2026-07-24 15:26:15 +02:00
parent 56327d0931 544b746895
commit 89babb00a9
71 changed files with 10058 additions and 345 deletions
Download Patch File Download Diff File Expand all files Collapse all files
v2/crates/wifi-densepose-sensing-server/src/embedding.rs
-1589
View File
File diff suppressed because it is too large Load Diff
v2/crates/wifi-densepose-sensing-server/src/graph_transformer.rs
-1154
View File
File diff suppressed because it is too large Load Diff
v2/crates/wifi-densepose-sensing-server/src/lib.rs
+9 -5
View File
@@ -14,10 +14,7 @@ pub mod ws_ticket;
pub mod cli;
pub mod dataset;
pub mod edge_registry;
#[allow(dead_code)]
pub mod embedding;
pub mod error_response;
pub mod graph_transformer;
pub mod host_validation;
pub mod introspection;
pub mod matter;
@@ -31,8 +28,6 @@ pub mod semantic;
pub mod rufield_surface;
pub mod rvf_container;
pub mod rvf_pipeline;
pub mod sona;
pub mod sparse_inference;
#[allow(dead_code)]
pub mod trainer;
pub mod vital_signs;
@@ -44,3 +39,12 @@ pub mod vendor_origin_plume;
pub mod vendor_remaining;
/// ADR-270 provider registry and canonical event helpers.
pub mod vendor_rf;
// ADR-185 §3.2/§13: the AETHER pure-compute stack (contrastive embedding,
// CSI-to-pose transformer, SONA, quantization) was hoisted into the std-only
// `wifi-densepose-aether` leaf crate so the Python `[aether]` wheel can bind it
// without this crate's Axum/tokio/worldgraph/ruvector tree. Re-exported here so
// this crate's own code (`crate::embedding`, `crate::graph_transformer`,
// `crate::sona`) and public API (`wifi_densepose_sensing_server::embedding`, …)
// are unchanged.
pub use wifi_densepose_aether::{embedding, graph_transformer, sona, sparse_inference};
v2/crates/wifi-densepose-sensing-server/src/main.rs
+362 -60
View File
@@ -20,8 +20,14 @@ mod multistatic_bridge;
mod mediatek_csi;
mod qualcomm_csi;
mod realtek_radar;
mod path_safety;
pub mod pose;
mod rvf_container;
// ADR-186 (TRAIN-RECONNECT): the in-server training pipeline was written but
// never declared as a module, so it was orphaned / uncompiled. Declaring it
// here compiles it against the real `AppStateInner` and wires its `routes()`
// (including `/ws/train/progress`) into the live router below.
mod training_api;
mod rvf_pipeline;
mod tracker_bridge;
pub mod types;
@@ -1120,11 +1126,13 @@ struct AppStateInner {
recording_current_id: Option<String>,
/// Shutdown signal for the recording writer task.
recording_stop_tx: Option<tokio::sync::watch::Sender<bool>>,
// ── Training fields ─────────────────────────────────────────────────────
/// Training status: "idle", "running", "completed", "failed".
training_status: String,
/// Training configuration, if any.
training_config: Option<serde_json::Value>,
// ── Training fields (ADR-186 TRAIN-RECONNECT) ────────────────────────────
/// Live training state (shared status snapshot + cooperative cancel flag +
/// background task handle) for the in-server trainer in `training_api`.
training_state: training_api::TrainingState,
/// Fan-out channel the background training job publishes progress JSON to;
/// the `/ws/train/progress` WebSocket handler subscribes to it.
training_progress_tx: broadcast::Sender<String>,
// ── Adaptive classifier (environment-tuned) ──────────────────────────
/// Trained adaptive model (loaded from data/adaptive_model.json or trained at runtime).
adaptive_model: Option<adaptive_classifier::AdaptiveModel>,
@@ -1248,6 +1256,87 @@ const FRAME_HISTORY_CAPACITY: usize = 100;
type SharedState = Arc<RwLock<AppStateInner>>;
#[cfg(test)]
impl AppStateInner {
/// Minimal, dependency-free `AppStateInner` for in-process router tests
/// (ADR-186 P6). Uses the same field constructors as the real state seeding
/// in `main()` but with trivial values and no CLI/config inputs, so tests can
/// build the training router without the full server boot.
pub(crate) fn minimal() -> Self {
AppStateInner {
latest_update: None,
rssi_history: VecDeque::new(),
frame_history: VecDeque::new(),
tick: 0,
source: "test".to_string(),
last_esp32_frame: None,
latest_realtek_radar: None,
last_realtek_frame: None,
latest_mediatek_csi: None,
last_mediatek_frame: None,
latest_qualcomm_csi: None,
last_qualcomm_frame: None,
latest_vendor_rf: BTreeMap::new(),
tx: broadcast::channel::<String>(16).0,
intro: wifi_densepose_sensing_server::introspection::IntrospectionState::new(),
intro_tx: broadcast::channel::<String>(16).0,
total_detections: 0,
start_time: std::time::Instant::now(),
vital_detector: VitalSignDetector::new(10.0),
latest_vitals: VitalSigns::default(),
rvf_info: None,
save_rvf_path: None,
progressive_loader: None,
active_sona_profile: None,
model_loaded: false,
smoothed_person_score: 0.0,
prev_person_count: 0,
smoothed_motion: 0.0,
current_motion_level: "absent".to_string(),
debounce_counter: 0,
debounce_candidate: "absent".to_string(),
baseline_motion: 0.0,
baseline_frames: 0,
smoothed_hr: 0.0,
smoothed_br: 0.0,
smoothed_hr_conf: 0.0,
smoothed_br_conf: 0.0,
hr_buffer: VecDeque::with_capacity(8),
br_buffer: VecDeque::with_capacity(8),
edge_vitals: None,
latest_wasm_events: None,
discovered_models: Vec::new(),
active_model_id: None,
recordings: Vec::new(),
recording_active: false,
recording_start_time: None,
recording_current_id: None,
recording_stop_tx: None,
training_state: training_api::TrainingState::default(),
training_progress_tx: broadcast::channel::<String>(256).0,
adaptive_model: None,
node_states: HashMap::new(),
pose_tracker: PoseTracker::new(),
last_tracker_instant: None,
multistatic_fuser: MultistaticFuser::new(),
engine_bridge: engine_bridge::EngineBridge::new(
wifi_densepose_bfld::PrivacyMode::PrivateHome,
1,
"default",
"Default Room",
None,
),
field_model: None,
p95_variance: RollingP95::new(600, 60),
p95_motion_band_power: RollingP95::new(600, 60),
p95_spectral_power: RollingP95::new(600, 60),
dedup_factor: 3.0,
data_dir: std::path::PathBuf::from("data"),
field_surface: Arc::new(RwLock::new(rufield_surface::FieldSurface::from_env())),
}
}
}
// ── ESP32 Edge Vitals Packet (ADR-039, magic 0xC511_0002) ────────────────────
/// Decoded vitals packet from ESP32 edge processing pipeline.
@@ -4973,54 +5062,12 @@ fn scan_recording_files() -> Vec<serde_json::Value> {
}
// ── Training Endpoints ──────────────────────────────────────────────────────
/// GET /api/v1/train/status — get training status.
async fn train_status(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
Json(serde_json::json!({
"status": s.training_status,
"config": s.training_config,
}))
}
/// POST /api/v1/train/start — start a training run.
async fn train_start(
State(state): State<SharedState>,
Json(body): Json<serde_json::Value>,
) -> Json<serde_json::Value> {
let mut s = state.write().await;
if s.training_status == "running" {
return Json(serde_json::json!({
"error": "training already running",
"success": false,
}));
}
s.training_status = "running".to_string();
s.training_config = Some(body.clone());
info!("Training started with config: {}", body);
Json(serde_json::json!({
"success": true,
"status": "running",
"message": "Training pipeline started. Use GET /api/v1/train/status to monitor.",
}))
}
/// POST /api/v1/train/stop — stop the current training run.
async fn train_stop(State(state): State<SharedState>) -> Json<serde_json::Value> {
let mut s = state.write().await;
if s.training_status != "running" {
return Json(serde_json::json!({
"error": "no training in progress",
"success": false,
}));
}
s.training_status = "idle".to_string();
info!("Training stopped");
Json(serde_json::json!({
"success": true,
"status": "idle",
}))
}
//
// ADR-186 (TRAIN-RECONNECT): the former stub handlers here flipped a status
// string and logged one line without ever starting a job (issue #1233). They
// are replaced by the real `training_api` router, merged into the app below,
// which runs the pure-Rust trainer on a background task and streams live
// progress over `/ws/train/progress`.
// ── Adaptive classifier endpoints ────────────────────────────────────────────
@@ -7826,9 +7873,9 @@ async fn main() {
recording_start_time: None,
recording_current_id: None,
recording_stop_tx: None,
// Training
training_status: "idle".to_string(),
training_config: None,
// Training (ADR-186 TRAIN-RECONNECT)
training_state: training_api::TrainingState::default(),
training_progress_tx: broadcast::channel::<String>(256).0,
adaptive_model:
adaptive_classifier::AdaptiveModel::load(&adaptive_classifier::model_path())
.ok()
@@ -8117,10 +8164,12 @@ async fn main() {
.route("/api/v1/recording/start", post(start_recording))
.route("/api/v1/recording/stop", post(stop_recording))
.route("/api/v1/recording/{id}", delete(delete_recording))
// Training endpoints
.route("/api/v1/train/status", get(train_status))
.route("/api/v1/train/start", post(train_start))
.route("/api/v1/train/stop", post(train_stop))
// Training endpoints (ADR-186 TRAIN-RECONNECT): the real in-server
// trainer + `/ws/train/progress` stream. Merged while the router is
// still `Router<SharedState>` (before `.with_state`) so these routes
// share `AppStateInner` and `/api/v1/train/*` sits under the bearer gate
// applied below (like the rest of `/api/v1/*`).
.merge(training_api::routes())
// Adaptive classifier endpoints
.route("/api/v1/adaptive/train", post(adaptive_train))
.route("/api/v1/adaptive/status", get(adaptive_status))
@@ -9369,3 +9418,256 @@ async fn oauth_status(
"scope": session.as_ref().map(|s| s.scope.clone()),
}))
}
#[cfg(test)]
mod adr186_http_tests {
//! ADR-186 P6: HTTP-level tests that build the real `training_api` router
//! and drive it in-process, guarding against the module being orphaned again
//! (`training_api::routes()` cannot compile unless the module is declared).
use super::*;
use axum::body::Body;
use axum::http::{Request, StatusCode};
use tower::ServiceExt;
/// Serializes tests that read/toggle the process-global
/// `RUVIEW_DISABLE_SERVER_TRAINING` env var, so the disabled-path test cannot
/// flip enablement while an enabled-path test is mid-request.
static TRAIN_ENV_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
fn test_state() -> SharedState {
Arc::new(RwLock::new(AppStateInner::minimal()))
}
/// The `/ws/train/progress` route is registered and reaches the WebSocket
/// handler (issue #1233 was a 404). Over `oneshot` there is no real socket to
/// upgrade, so axum returns 426 Upgrade Required — which still distinguishes a
/// wired WS endpoint (426) from an orphaned/absent route (404). The genuine
/// 101 handshake is asserted by `ws_train_progress_live_101_and_frame`.
#[tokio::test]
async fn ws_train_progress_route_is_wired_not_404() {
let app = training_api::routes().with_state(test_state());
let req = Request::builder()
.uri("/ws/train/progress")
.header("connection", "upgrade")
.header("upgrade", "websocket")
.header("sec-websocket-version", "13")
.header("sec-websocket-key", "dGhlIHNhbXBsZSBub25jZQ==")
.body(Body::empty())
.unwrap();
let resp = app.oneshot(req).await.unwrap();
assert_ne!(resp.status(), StatusCode::NOT_FOUND, "route must not 404");
assert_eq!(
resp.status(),
StatusCode::UPGRADE_REQUIRED,
"a wired WS route returns 426 under oneshot — got {}",
resp.status()
);
}
/// ADR-186 §7 acceptance: over a real socket, `/ws/train/progress` completes a
/// genuine 101 WebSocket handshake and, after a `POST /api/v1/train/start`,
/// delivers at least one real `progress` frame to the connected client.
#[tokio::test]
async fn ws_train_progress_live_101_and_frame() {
use futures_util::StreamExt;
use tokio::io::AsyncWriteExt;
use tokio_tungstenite::tungstenite::Message as TMsg;
let _env_lock = TRAIN_ENV_LOCK.lock().unwrap(); // enablement must stay ON
let shared = test_state();
{
let mut s = shared.write().await;
for i in 0..40 {
let sub: Vec<f64> = (0..56)
.map(|k| 10.0 + ((i as f64) * 0.3 + (k as f64) * 0.1).sin() * 2.0)
.collect();
s.frame_history.push_back(sub);
}
}
// Serve the training router on an ephemeral port.
let app = training_api::routes().with_state(shared.clone());
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
let _ = axum::serve(listener, app).await;
});
// A successful `connect_async` IS the 101 handshake (it errors otherwise).
let (mut ws, resp) =
tokio_tungstenite::connect_async(format!("ws://{addr}/ws/train/progress"))
.await
.expect("WebSocket handshake should succeed (101)");
assert_eq!(resp.status().as_u16(), 101, "handshake must be 101");
// Drive training via a real HTTP POST over a fresh TCP connection.
let body = r#"{"dataset_ids":[],"config":{"epochs":3,"batch_size":8,"warmup_epochs":1,"early_stopping_patience":10}}"#;
let req = format!(
"POST /api/v1/train/start HTTP/1.1\r\nHost: {addr}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
body.len(),
body
);
let mut post = tokio::net::TcpStream::connect(addr).await.unwrap();
post.write_all(req.as_bytes()).await.unwrap();
post.flush().await.unwrap();
// Read WS frames until a `progress` frame arrives (or a 10s ceiling).
let mut got_progress = false;
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(10);
while tokio::time::Instant::now() < deadline {
match tokio::time::timeout(std::time::Duration::from_secs(2), ws.next()).await {
Ok(Some(Ok(TMsg::Text(txt)))) => {
if let Ok(v) = serde_json::from_str::<serde_json::Value>(&txt) {
if v.get("type").and_then(|t| t.as_str()) == Some("progress") {
got_progress = true;
break;
}
}
}
Ok(Some(Ok(_))) => {}
Ok(Some(Err(_))) | Ok(None) => break,
Err(_) => {}
}
}
assert!(
got_progress,
"should receive a real progress frame over the live WS after POST start"
);
// NOTE: deliberately no directory-diff cleanup here. `data/models` is
// gitignored, and deleting by dir-diff would race concurrent model-writing
// tests (it could remove a `.rvf` another test is asserting exists).
}
/// Full HTTP round-trip: POST /api/v1/train/start → poll /api/v1/train/status
/// until completion → a real `.rvf` model artifact exists on disk, and real
/// progress frames were streamed on the broadcast channel.
#[tokio::test]
async fn http_train_start_produces_model_and_streams() {
let _env_lock = TRAIN_ENV_LOCK.lock().unwrap(); // enablement must stay ON
let shared = test_state();
// Seed synthetic frames so training's fallback path has data (no files).
{
let mut s = shared.write().await;
for i in 0..40 {
let sub: Vec<f64> = (0..56)
.map(|k| 10.0 + ((i as f64) * 0.3 + (k as f64) * 0.1).sin() * 2.0)
.collect();
s.frame_history.push_back(sub);
}
}
let mut progress_rx = {
let s = shared.read().await;
s.training_progress_tx.subscribe()
};
let models_dir = std::path::PathBuf::from(training_api::MODELS_DIR);
let before: std::collections::HashSet<std::path::PathBuf> = std::fs::read_dir(&models_dir)
.into_iter()
.flatten()
.flatten()
.map(|e| e.path())
.collect();
let app = training_api::routes().with_state(shared.clone());
// POST start.
let body = serde_json::json!({
"dataset_ids": [],
"config": {"epochs": 3, "batch_size": 8, "warmup_epochs": 1, "early_stopping_patience": 10}
});
let req = Request::builder()
.method("POST")
.uri("/api/v1/train/start")
.header("content-type", "application/json")
.body(Body::from(body.to_string()))
.unwrap();
let resp = app.clone().oneshot(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK, "start should be accepted");
// Poll status until the job reports completion.
let mut completed = false;
for _ in 0..250 {
let req = Request::builder()
.uri("/api/v1/train/status")
.body(Body::empty())
.unwrap();
let resp = app.clone().oneshot(req).await.unwrap();
let bytes = axum::body::to_bytes(resp.into_body(), 65536).await.unwrap();
let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
// Status also carries the P5 enablement flag.
assert_eq!(v.get("enabled"), Some(&serde_json::Value::Bool(true)));
if v.get("active") == Some(&serde_json::Value::Bool(false))
&& v.get("phase").and_then(|p| p.as_str()) == Some("completed")
{
completed = true;
break;
}
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
}
assert!(completed, "training should reach the completed phase");
// Real progress frames were streamed.
let mut saw_progress = false;
while progress_rx.try_recv().is_ok() {
saw_progress = true;
}
assert!(saw_progress, "expected streamed progress frames over the WS channel");
// A new .rvf artifact was written by the run.
let after: std::collections::HashSet<std::path::PathBuf> = std::fs::read_dir(&models_dir)
.into_iter()
.flatten()
.flatten()
.map(|e| e.path())
.collect();
let new_models: Vec<_> = after
.difference(&before)
.filter(|p| p.extension().and_then(|e| e.to_str()) == Some("rvf"))
.cloned()
.collect();
assert!(
!new_models.is_empty(),
"training should write a new .rvf model artifact under {}",
models_dir.display()
);
// No deletion here: removing by dir-diff would race concurrent
// model-writing tests. `data/models` is gitignored.
}
/// P5 fallback guarantee: with server training disabled, POST start returns a
/// structured `{enabled:false, cli:...}` 409 — never a silent success.
#[tokio::test]
async fn http_train_start_disabled_returns_structured_409() {
// Serialize against the enabled-path tests so our env toggle can't race
// their in-flight requests.
let _env_lock = TRAIN_ENV_LOCK.lock().unwrap();
std::env::set_var("RUVIEW_DISABLE_SERVER_TRAINING", "1");
let app = training_api::routes().with_state(test_state());
let body = serde_json::json!({"dataset_ids": [], "config": {"epochs": 1}});
let req = Request::builder()
.method("POST")
.uri("/api/v1/train/start")
.header("content-type", "application/json")
.body(Body::from(body.to_string()))
.unwrap();
let resp = app.oneshot(req).await.unwrap();
let status = resp.status();
let bytes = axum::body::to_bytes(resp.into_body(), 65536).await.unwrap();
let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
std::env::remove_var("RUVIEW_DISABLE_SERVER_TRAINING");
assert_eq!(status, StatusCode::CONFLICT, "disabled start must be 4xx/409");
assert_eq!(v.get("enabled"), Some(&serde_json::Value::Bool(false)));
assert_eq!(
v.get("cli").and_then(|c| c.as_str()),
Some("wifi-densepose train-room"),
"must point at the CLI fallback, never a silent success"
);
assert_ne!(
v.get("success"),
Some(&serde_json::Value::Bool(true)),
"must never claim success:true when disabled"
);
}
}
v2/crates/wifi-densepose-sensing-server/src/sona.rs
-838
View File
@@ -1,838 +0,0 @@
//! SONA online adaptation: LoRA + EWC++ for WiFi-DensePose (ADR-023 Phase 5).
//!
//! Enables rapid low-parameter adaptation to changing WiFi environments without
//! catastrophic forgetting. All arithmetic uses `f32`, no external dependencies.
use std::collections::VecDeque;
// ── LoRA Adapter ────────────────────────────────────────────────────────────
/// Low-Rank Adaptation layer storing factorised delta `scale * A * B`.
#[derive(Debug, Clone)]
pub struct LoraAdapter {
pub a: Vec<Vec<f32>>, // (in_features, rank)
pub b: Vec<Vec<f32>>, // (rank, out_features)
pub scale: f32, // alpha / rank
pub in_features: usize,
pub out_features: usize,
pub rank: usize,
}
impl LoraAdapter {
pub fn new(in_features: usize, out_features: usize, rank: usize, alpha: f32) -> Self {
Self {
a: vec![vec![0.0f32; rank]; in_features],
b: vec![vec![0.0f32; out_features]; rank],
scale: alpha / rank.max(1) as f32,
in_features,
out_features,
rank,
}
}
/// Compute `scale * input * A * B`, returning a vector of length `out_features`.
#[allow(clippy::needless_range_loop)]
pub fn forward(&self, input: &[f32]) -> Vec<f32> {
assert_eq!(input.len(), self.in_features);
let mut hidden = vec![0.0f32; self.rank];
for (i, &x) in input.iter().enumerate() {
for r in 0..self.rank {
hidden[r] += x * self.a[i][r];
}
}
let mut output = vec![0.0f32; self.out_features];
for r in 0..self.rank {
for j in 0..self.out_features {
output[j] += hidden[r] * self.b[r][j];
}
}
for v in output.iter_mut() {
*v *= self.scale;
}
output
}
/// Full delta weight matrix `scale * A * B`, shape (in_features, out_features).
#[allow(clippy::needless_range_loop)]
pub fn delta_weights(&self) -> Vec<Vec<f32>> {
let mut delta = vec![vec![0.0f32; self.out_features]; self.in_features];
for i in 0..self.in_features {
for r in 0..self.rank {
let a_val = self.a[i][r];
for j in 0..self.out_features {
delta[i][j] += a_val * self.b[r][j];
}
}
}
for row in delta.iter_mut() {
for v in row.iter_mut() {
*v *= self.scale;
}
}
delta
}
/// Add LoRA delta to base weights in place.
pub fn merge_into(&self, base_weights: &mut [Vec<f32>]) {
let delta = self.delta_weights();
for (rb, rd) in base_weights.iter_mut().zip(delta.iter()) {
for (w, &d) in rb.iter_mut().zip(rd.iter()) {
*w += d;
}
}
}
/// Subtract LoRA delta from base weights in place.
pub fn unmerge_from(&self, base_weights: &mut [Vec<f32>]) {
let delta = self.delta_weights();
for (rb, rd) in base_weights.iter_mut().zip(delta.iter()) {
for (w, &d) in rb.iter_mut().zip(rd.iter()) {
*w -= d;
}
}
}
/// Trainable parameter count: `rank * (in_features + out_features)`.
pub fn n_params(&self) -> usize {
self.rank * (self.in_features + self.out_features)
}
/// Reset A and B to zero.
pub fn reset(&mut self) {
for row in self.a.iter_mut() {
for v in row.iter_mut() {
*v = 0.0;
}
}
for row in self.b.iter_mut() {
for v in row.iter_mut() {
*v = 0.0;
}
}
}
}
// ── EWC++ Regularizer ───────────────────────────────────────────────────────
/// Elastic Weight Consolidation++ regularizer with running Fisher average.
#[derive(Debug, Clone)]
pub struct EwcRegularizer {
pub lambda: f32,
pub decay: f32,
pub fisher_diag: Vec<f32>,
pub reference_params: Vec<f32>,
}
impl EwcRegularizer {
pub fn new(lambda: f32, decay: f32) -> Self {
Self {
lambda,
decay,
fisher_diag: Vec::new(),
reference_params: Vec::new(),
}
}
/// Diagonal Fisher via numerical central differences: F_i = grad_i^2.
pub fn compute_fisher(
params: &[f32],
loss_fn: impl Fn(&[f32]) -> f32,
n_samples: usize,
) -> Vec<f32> {
let eps = 1e-4f32;
let n = params.len();
let mut fisher = vec![0.0f32; n];
let samples = n_samples.max(1);
for _ in 0..samples {
let mut p = params.to_vec();
for i in 0..n {
let orig = p[i];
p[i] = orig + eps;
let lp = loss_fn(&p);
p[i] = orig - eps;
let lm = loss_fn(&p);
p[i] = orig;
let g = (lp - lm) / (2.0 * eps);
fisher[i] += g * g;
}
}
for f in fisher.iter_mut() {
*f /= samples as f32;
}
fisher
}
/// Online update: `F = decay * F_old + (1-decay) * F_new`.
pub fn update_fisher(&mut self, new_fisher: &[f32]) {
if self.fisher_diag.is_empty() {
self.fisher_diag = new_fisher.to_vec();
return;
}
assert_eq!(self.fisher_diag.len(), new_fisher.len());
for (old, &nv) in self.fisher_diag.iter_mut().zip(new_fisher.iter()) {
*old = self.decay * *old + (1.0 - self.decay) * nv;
}
}
/// Penalty: `0.5 * lambda * sum(F_i * (theta_i - theta_i*)^2)`.
pub fn penalty(&self, current_params: &[f32]) -> f32 {
if self.reference_params.is_empty() || self.fisher_diag.is_empty() {
return 0.0;
}
let n = current_params
.len()
.min(self.reference_params.len())
.min(self.fisher_diag.len());
let mut sum = 0.0f32;
#[allow(clippy::needless_range_loop)]
for i in 0..n {
let d = current_params[i] - self.reference_params[i];
sum += self.fisher_diag[i] * d * d;
}
0.5 * self.lambda * sum
}
/// Gradient of penalty: `lambda * F_i * (theta_i - theta_i*)`.
pub fn penalty_gradient(&self, current_params: &[f32]) -> Vec<f32> {
if self.reference_params.is_empty() || self.fisher_diag.is_empty() {
return vec![0.0f32; current_params.len()];
}
let n = current_params
.len()
.min(self.reference_params.len())
.min(self.fisher_diag.len());
let mut grad = vec![0.0f32; current_params.len()];
for i in 0..n {
grad[i] =
self.lambda * self.fisher_diag[i] * (current_params[i] - self.reference_params[i]);
}
grad
}
/// Save current params as the new reference point.
pub fn consolidate(&mut self, params: &[f32]) {
self.reference_params = params.to_vec();
}
}
// ── Configuration & Types ───────────────────────────────────────────────────
/// SONA adaptation configuration.
#[derive(Debug, Clone)]
pub struct SonaConfig {
pub lora_rank: usize,
pub lora_alpha: f32,
pub ewc_lambda: f32,
pub ewc_decay: f32,
pub adaptation_lr: f32,
pub max_steps: usize,
pub convergence_threshold: f32,
pub temporal_consistency_weight: f32,
}
impl Default for SonaConfig {
fn default() -> Self {
Self {
lora_rank: 4,
lora_alpha: 8.0,
ewc_lambda: 5000.0,
ewc_decay: 0.99,
adaptation_lr: 0.001,
max_steps: 50,
convergence_threshold: 1e-4,
temporal_consistency_weight: 0.1,
}
}
}
/// Single training sample for online adaptation.
#[derive(Debug, Clone)]
pub struct AdaptationSample {
pub csi_features: Vec<f32>,
pub target: Vec<f32>,
}
/// Result of a SONA adaptation run.
#[derive(Debug, Clone)]
pub struct AdaptationResult {
pub adapted_params: Vec<f32>,
pub steps_taken: usize,
pub final_loss: f32,
pub converged: bool,
pub ewc_penalty: f32,
}
/// Saved environment-specific adaptation profile.
#[derive(Debug, Clone)]
pub struct SonaProfile {
pub name: String,
pub lora_a: Vec<Vec<f32>>,
pub lora_b: Vec<Vec<f32>>,
pub fisher_diag: Vec<f32>,
pub reference_params: Vec<f32>,
pub adaptation_count: usize,
}
// ── SONA Adapter ────────────────────────────────────────────────────────────
/// Full SONA system: LoRA adapter + EWC++ regularizer for online adaptation.
#[derive(Debug, Clone)]
pub struct SonaAdapter {
pub config: SonaConfig,
pub lora: LoraAdapter,
pub ewc: EwcRegularizer,
pub param_count: usize,
pub adaptation_count: usize,
}
impl SonaAdapter {
pub fn new(config: SonaConfig, param_count: usize) -> Self {
let lora = LoraAdapter::new(param_count, 1, config.lora_rank, config.lora_alpha);
let ewc = EwcRegularizer::new(config.ewc_lambda, config.ewc_decay);
Self {
config,
lora,
ewc,
param_count,
adaptation_count: 0,
}
}
/// Run gradient descent with LoRA + EWC on the given samples.
pub fn adapt(&mut self, base_params: &[f32], samples: &[AdaptationSample]) -> AdaptationResult {
assert_eq!(base_params.len(), self.param_count);
if samples.is_empty() {
return AdaptationResult {
adapted_params: base_params.to_vec(),
steps_taken: 0,
final_loss: 0.0,
converged: true,
ewc_penalty: self.ewc.penalty(base_params),
};
}
let lr = self.config.adaptation_lr;
let (mut prev_loss, mut steps, mut converged) = (f32::MAX, 0usize, false);
let out_dim = samples[0].target.len();
let in_dim = samples[0].csi_features.len();
for step in 0..self.config.max_steps {
steps = step + 1;
let df = self.lora_delta_flat();
let eff: Vec<f32> = base_params
.iter()
.zip(df.iter())
.map(|(&b, &d)| b + d)
.collect();
let (dl, dg) = Self::mse_loss_grad(&eff, samples, in_dim, out_dim);
let ep = self.ewc.penalty(&eff);
let eg = self.ewc.penalty_gradient(&eff);
let total = dl + ep;
if (prev_loss - total).abs() < self.config.convergence_threshold {
converged = true;
prev_loss = total;
break;
}
prev_loss = total;
let gl = df.len().min(dg.len()).min(eg.len());
let mut tg = vec![0.0f32; gl];
for i in 0..gl {
tg[i] = dg[i] + eg[i];
}
self.update_lora(&tg, lr);
}
let df = self.lora_delta_flat();
let adapted: Vec<f32> = base_params
.iter()
.zip(df.iter())
.map(|(&b, &d)| b + d)
.collect();
let ewc_penalty = self.ewc.penalty(&adapted);
self.adaptation_count += 1;
AdaptationResult {
adapted_params: adapted,
steps_taken: steps,
final_loss: prev_loss,
converged,
ewc_penalty,
}
}
pub fn save_profile(&self, name: &str) -> SonaProfile {
SonaProfile {
name: name.to_string(),
lora_a: self.lora.a.clone(),
lora_b: self.lora.b.clone(),
fisher_diag: self.ewc.fisher_diag.clone(),
reference_params: self.ewc.reference_params.clone(),
adaptation_count: self.adaptation_count,
}
}
pub fn load_profile(&mut self, profile: &SonaProfile) {
self.lora.a = profile.lora_a.clone();
self.lora.b = profile.lora_b.clone();
self.ewc.fisher_diag = profile.fisher_diag.clone();
self.ewc.reference_params = profile.reference_params.clone();
self.adaptation_count = profile.adaptation_count;
}
fn lora_delta_flat(&self) -> Vec<f32> {
self.lora
.delta_weights()
.into_iter()
.map(|r| r[0])
.collect()
}
fn mse_loss_grad(
params: &[f32],
samples: &[AdaptationSample],
in_dim: usize,
out_dim: usize,
) -> (f32, Vec<f32>) {
let n = samples.len() as f32;
let ws = in_dim * out_dim;
let mut grad = vec![0.0f32; params.len()];
let mut loss = 0.0f32;
for s in samples {
let (inp, tgt) = (&s.csi_features, &s.target);
let mut pred = vec![0.0f32; out_dim];
#[allow(clippy::needless_range_loop)]
for j in 0..out_dim {
for i in 0..in_dim.min(inp.len()) {
let idx = j * in_dim + i;
if idx < ws && idx < params.len() {
pred[j] += params[idx] * inp[i];
}
}
}
for j in 0..out_dim.min(tgt.len()) {
let e = pred[j] - tgt[j];
loss += e * e;
#[allow(clippy::needless_range_loop)]
for i in 0..in_dim.min(inp.len()) {
let idx = j * in_dim + i;
if idx < ws && idx < grad.len() {
grad[idx] += 2.0 * e * inp[i] / n;
}
}
}
}
(loss / n, grad)
}
#[allow(clippy::needless_range_loop)]
fn update_lora(&mut self, grad: &[f32], lr: f32) {
let (scale, rank) = (self.lora.scale, self.lora.rank);
if self.lora.b.iter().all(|r| r.iter().all(|&v| v == 0.0)) && rank > 0 {
self.lora.b[0][0] = 1.0;
}
for i in 0..self.lora.in_features.min(grad.len()) {
for r in 0..rank {
self.lora.a[i][r] -= lr * grad[i] * scale * self.lora.b[r][0];
}
}
for r in 0..rank {
let mut g = 0.0f32;
for i in 0..self.lora.in_features.min(grad.len()) {
g += grad[i] * scale * self.lora.a[i][r];
}
self.lora.b[r][0] -= lr * g;
}
}
}
// ── Environment Detector ────────────────────────────────────────────────────
/// CSI baseline drift information.
#[derive(Debug, Clone)]
pub struct DriftInfo {
pub magnitude: f32,
pub duration_frames: usize,
pub baseline_mean: f32,
pub current_mean: f32,
}
/// Detects environmental drift in CSI statistics (>3 sigma from baseline).
#[derive(Debug, Clone)]
pub struct EnvironmentDetector {
window_size: usize,
means: VecDeque<f32>,
variances: VecDeque<f32>,
baseline_mean: f32,
baseline_var: f32,
baseline_std: f32,
baseline_set: bool,
drift_frames: usize,
}
impl EnvironmentDetector {
pub fn new(window_size: usize) -> Self {
Self {
window_size: window_size.max(2),
means: VecDeque::with_capacity(window_size),
variances: VecDeque::with_capacity(window_size),
baseline_mean: 0.0,
baseline_var: 0.0,
baseline_std: 0.0,
baseline_set: false,
drift_frames: 0,
}
}
pub fn update(&mut self, csi_mean: f32, csi_var: f32) {
self.means.push_back(csi_mean);
self.variances.push_back(csi_var);
while self.means.len() > self.window_size {
self.means.pop_front();
}
while self.variances.len() > self.window_size {
self.variances.pop_front();
}
if !self.baseline_set && self.means.len() >= self.window_size {
self.reset_baseline();
}
if self.drift_detected() {
self.drift_frames += 1;
} else {
self.drift_frames = 0;
}
}
pub fn drift_detected(&self) -> bool {
if !self.baseline_set || self.means.is_empty() {
return false;
}
let dev = (self.current_mean() - self.baseline_mean).abs();
let thr = if self.baseline_std > f32::EPSILON {
3.0 * self.baseline_std
} else {
f32::EPSILON * 100.0
};
dev > thr
}
pub fn reset_baseline(&mut self) {
if self.means.is_empty() {
return;
}
let n = self.means.len() as f32;
self.baseline_mean = self.means.iter().sum::<f32>() / n;
let var = self
.means
.iter()
.map(|&m| (m - self.baseline_mean).powi(2))
.sum::<f32>()
/ n;
self.baseline_var = var;
self.baseline_std = var.sqrt();
self.baseline_set = true;
self.drift_frames = 0;
}
pub fn drift_info(&self) -> DriftInfo {
let cm = self.current_mean();
let abs_dev = (cm - self.baseline_mean).abs();
let magnitude = if self.baseline_std > f32::EPSILON {
abs_dev / self.baseline_std
} else if abs_dev > f32::EPSILON {
abs_dev / f32::EPSILON
} else {
0.0
};
DriftInfo {
magnitude,
duration_frames: self.drift_frames,
baseline_mean: self.baseline_mean,
current_mean: cm,
}
}
fn current_mean(&self) -> f32 {
if self.means.is_empty() {
0.0
} else {
self.means.iter().sum::<f32>() / self.means.len() as f32
}
}
}
// ── Temporal Consistency Loss ───────────────────────────────────────────────
/// Penalises large velocity between consecutive outputs: `sum((c-p)^2) / dt`.
pub struct TemporalConsistencyLoss;
impl TemporalConsistencyLoss {
pub fn compute(prev_output: &[f32], curr_output: &[f32], dt: f32) -> f32 {
if dt <= 0.0 {
return 0.0;
}
let n = prev_output.len().min(curr_output.len());
let mut sq = 0.0f32;
for i in 0..n {
let d = curr_output[i] - prev_output[i];
sq += d * d;
}
sq / dt
}
}
// ── Tests ───────────────────────────────────────────────────────────────────
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn lora_adapter_param_count() {
let lora = LoraAdapter::new(64, 32, 4, 8.0);
assert_eq!(lora.n_params(), 4 * (64 + 32));
}
#[test]
fn lora_adapter_forward_shape() {
let lora = LoraAdapter::new(8, 4, 2, 4.0);
assert_eq!(lora.forward(&[1.0f32; 8]).len(), 4);
}
#[test]
fn lora_adapter_zero_init_produces_zero_delta() {
let delta = LoraAdapter::new(8, 4, 2, 4.0).delta_weights();
assert_eq!(delta.len(), 8);
for row in &delta {
assert_eq!(row.len(), 4);
for &v in row {
assert_eq!(v, 0.0);
}
}
}
#[test]
fn lora_adapter_merge_unmerge_roundtrip() {
let mut lora = LoraAdapter::new(3, 2, 1, 2.0);
lora.a[0][0] = 1.0;
lora.a[1][0] = 2.0;
lora.a[2][0] = 3.0;
lora.b[0][0] = 0.5;
lora.b[0][1] = -0.5;
let mut base = vec![vec![10.0, 20.0], vec![30.0, 40.0], vec![50.0, 60.0]];
let orig = base.clone();
lora.merge_into(&mut base);
assert_ne!(base, orig);
lora.unmerge_from(&mut base);
for (rb, ro) in base.iter().zip(orig.iter()) {
for (&b, &o) in rb.iter().zip(ro.iter()) {
assert!((b - o).abs() < 1e-5, "roundtrip failed: {b} vs {o}");
}
}
}
#[test]
fn lora_adapter_rank_1_outer_product() {
let mut lora = LoraAdapter::new(3, 2, 1, 1.0); // scale=1
lora.a[0][0] = 1.0;
lora.a[1][0] = 2.0;
lora.a[2][0] = 3.0;
lora.b[0][0] = 4.0;
lora.b[0][1] = 5.0;
let d = lora.delta_weights();
let expected = [[4.0, 5.0], [8.0, 10.0], [12.0, 15.0]];
for (i, row) in expected.iter().enumerate() {
for (j, &v) in row.iter().enumerate() {
assert!((d[i][j] - v).abs() < 1e-6);
}
}
}
#[test]
fn lora_scale_factor() {
assert!((LoraAdapter::new(8, 4, 4, 16.0).scale - 4.0).abs() < 1e-6);
assert!((LoraAdapter::new(8, 4, 2, 8.0).scale - 4.0).abs() < 1e-6);
}
#[test]
fn ewc_fisher_positive() {
let fisher = EwcRegularizer::compute_fisher(
&[1.0f32, -2.0, 0.5],
|p: &[f32]| p.iter().map(|&x| x * x).sum::<f32>(),
1,
);
assert_eq!(fisher.len(), 3);
for &f in &fisher {
assert!(f >= 0.0, "Fisher must be >= 0, got {f}");
}
}
#[test]
fn ewc_penalty_zero_at_reference() {
let mut ewc = EwcRegularizer::new(5000.0, 0.99);
let p = vec![1.0, 2.0, 3.0];
ewc.fisher_diag = vec![1.0; 3];
ewc.consolidate(&p);
assert!(ewc.penalty(&p).abs() < 1e-10);
}
#[test]
fn ewc_penalty_positive_away_from_reference() {
let mut ewc = EwcRegularizer::new(5000.0, 0.99);
ewc.fisher_diag = vec![1.0; 3];
ewc.consolidate(&[1.0, 2.0, 3.0]);
let pen = ewc.penalty(&[2.0, 3.0, 4.0]);
assert!(pen > 0.0); // 0.5 * 5000 * 3 = 7500
assert!((pen - 7500.0).abs() < 1e-3, "expected ~7500, got {pen}");
}
#[test]
fn ewc_penalty_gradient_direction() {
let mut ewc = EwcRegularizer::new(100.0, 0.99);
let r = vec![1.0, 2.0, 3.0];
ewc.fisher_diag = vec![1.0; 3];
ewc.consolidate(&r);
let c = vec![2.0, 4.0, 5.0];
let grad = ewc.penalty_gradient(&c);
for (i, &g) in grad.iter().enumerate() {
assert!(g * (c[i] - r[i]) > 0.0, "gradient[{i}] wrong sign");
}
}
#[test]
fn ewc_online_update_decays() {
let mut ewc = EwcRegularizer::new(1.0, 0.5);
ewc.update_fisher(&[10.0, 20.0]);
assert!((ewc.fisher_diag[0] - 10.0).abs() < 1e-6);
ewc.update_fisher(&[0.0, 0.0]);
assert!((ewc.fisher_diag[0] - 5.0).abs() < 1e-6); // 0.5*10 + 0.5*0
assert!((ewc.fisher_diag[1] - 10.0).abs() < 1e-6); // 0.5*20 + 0.5*0
}
#[test]
fn ewc_consolidate_updates_reference() {
let mut ewc = EwcRegularizer::new(1.0, 0.99);
ewc.consolidate(&[1.0, 2.0]);
assert_eq!(ewc.reference_params, vec![1.0, 2.0]);
ewc.consolidate(&[3.0, 4.0]);
assert_eq!(ewc.reference_params, vec![3.0, 4.0]);
}
#[test]
fn sona_config_defaults() {
let c = SonaConfig::default();
assert_eq!(c.lora_rank, 4);
assert!((c.lora_alpha - 8.0).abs() < 1e-6);
assert!((c.ewc_lambda - 5000.0).abs() < 1e-3);
assert!((c.ewc_decay - 0.99).abs() < 1e-6);
assert!((c.adaptation_lr - 0.001).abs() < 1e-6);
assert_eq!(c.max_steps, 50);
assert!((c.convergence_threshold - 1e-4).abs() < 1e-8);
assert!((c.temporal_consistency_weight - 0.1).abs() < 1e-6);
}
#[test]
fn sona_adapter_converges_on_simple_task() {
let cfg = SonaConfig {
lora_rank: 1,
lora_alpha: 1.0,
ewc_lambda: 0.0,
ewc_decay: 0.99,
adaptation_lr: 0.01,
max_steps: 200,
convergence_threshold: 1e-6,
temporal_consistency_weight: 0.0,
};
let mut adapter = SonaAdapter::new(cfg, 1);
let samples: Vec<_> = (1..=5)
.map(|i| {
let x = i as f32;
AdaptationSample {
csi_features: vec![x],
target: vec![2.0 * x],
}
})
.collect();
let r = adapter.adapt(&[0.0f32], &samples);
assert!(
r.final_loss < 1.0,
"loss should decrease, got {}",
r.final_loss
);
assert!(r.steps_taken > 0);
}
#[test]
fn sona_adapter_respects_max_steps() {
let cfg = SonaConfig {
max_steps: 5,
convergence_threshold: 0.0,
..SonaConfig::default()
};
let mut a = SonaAdapter::new(cfg, 4);
let s = vec![AdaptationSample {
csi_features: vec![1.0, 0.0, 0.0, 0.0],
target: vec![1.0],
}];
assert_eq!(a.adapt(&[0.0; 4], &s).steps_taken, 5);
}
#[test]
fn sona_profile_save_load_roundtrip() {
let mut a = SonaAdapter::new(SonaConfig::default(), 8);
a.lora.a[0][0] = 1.5;
a.lora.b[0][0] = -0.3;
a.ewc.fisher_diag = vec![1.0, 2.0, 3.0];
a.ewc.reference_params = vec![0.1, 0.2, 0.3];
a.adaptation_count = 42;
let p = a.save_profile("test-env");
assert_eq!(p.name, "test-env");
assert_eq!(p.adaptation_count, 42);
let mut a2 = SonaAdapter::new(SonaConfig::default(), 8);
a2.load_profile(&p);
assert!((a2.lora.a[0][0] - 1.5).abs() < 1e-6);
assert!((a2.lora.b[0][0] - (-0.3)).abs() < 1e-6);
assert_eq!(a2.ewc.fisher_diag.len(), 3);
assert!((a2.ewc.fisher_diag[2] - 3.0).abs() < 1e-6);
assert_eq!(a2.adaptation_count, 42);
}
#[test]
fn environment_detector_no_drift_initially() {
assert!(!EnvironmentDetector::new(10).drift_detected());
}
#[test]
fn environment_detector_detects_large_shift() {
let mut d = EnvironmentDetector::new(10);
for _ in 0..10 {
d.update(10.0, 0.1);
}
assert!(!d.drift_detected());
for _ in 0..10 {
d.update(50.0, 0.1);
}
assert!(d.drift_detected());
assert!(
d.drift_info().magnitude > 3.0,
"magnitude = {}",
d.drift_info().magnitude
);
}
#[test]
fn environment_detector_reset_baseline() {
let mut d = EnvironmentDetector::new(10);
for _ in 0..10 {
d.update(10.0, 0.1);
}
for _ in 0..10 {
d.update(50.0, 0.1);
}
assert!(d.drift_detected());
d.reset_baseline();
assert!(!d.drift_detected());
}
#[test]
fn temporal_consistency_zero_for_static() {
let o = vec![1.0, 2.0, 3.0];
assert!(TemporalConsistencyLoss::compute(&o, &o, 0.033).abs() < 1e-10);
}
}
v2/crates/wifi-densepose-sensing-server/src/sparse_inference.rs
-1016
View File
File diff suppressed because it is too large Load Diff
v2/crates/wifi-densepose-sensing-server/src/training_api.rs
+530 -205
View File
@@ -26,22 +26,23 @@
use std::collections::VecDeque;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use axum::{
extract::{
ws::{Message, WebSocket, WebSocketUpgrade},
State,
},
response::{IntoResponse, Json},
http::StatusCode,
response::{IntoResponse, Json, Response},
routing::{get, post},
Router,
};
use serde::{Deserialize, Serialize};
use tokio::sync::{broadcast, RwLock};
use tokio::sync::broadcast;
use tracing::{error, info, warn};
use crate::recording::{RecordedFrame, RECORDINGS_DIR};
use crate::rvf_container::RvfBuilder;
// ── Constants ────────────────────────────────────────────────────────────────
@@ -49,6 +50,28 @@ use crate::rvf_container::RvfBuilder;
/// Directory for trained model output.
pub const MODELS_DIR: &str = "data/models";
/// Directory the training loop reads recorded CSI datasets from. Each
/// `dataset_id` maps to `{RECORDINGS_DIR}/{dataset_id}.csi.jsonl`.
pub const RECORDINGS_DIR: &str = "data/recordings";
/// Monotonic per-process counter appended to exported model filenames so two
/// runs that complete in the same wall-clock microsecond still get distinct
/// paths (prevents silent overwrite; keeps concurrent runs from colliding).
static MODEL_ID_SEQ: AtomicU64 = AtomicU64::new(0);
/// Build a process-unique model id `trained-{type}-{ts_micros}-{seq}`. A
/// second-resolution timestamp alone collided for runs finishing in the same
/// second (silent overwrite); microseconds + the monotonic counter guarantee
/// uniqueness even for same-microsecond concurrent completions.
fn next_model_id(training_type: &str) -> String {
format!(
"trained-{}-{}-{}",
training_type,
chrono::Utc::now().format("%Y%m%d_%H%M%S_%6f"),
MODEL_ID_SEQ.fetch_add(1, Ordering::Relaxed)
)
}
/// Number of COCO keypoints.
const N_KEYPOINTS: usize = 17;
/// Dimensions per keypoint in the target vector (x, y, z).
@@ -67,6 +90,25 @@ const N_GLOBAL_FEATURES: usize = 3;
// ── Types ────────────────────────────────────────────────────────────────────
/// A single recorded CSI frame line, as stored in the `.csi.jsonl` datasets the
/// training loop consumes.
///
/// This mirrors the on-disk JSONL schema and is intentionally self-contained so
/// the trainer does not couple to the (separate, orphaned) `recording.rs`
/// module. Only the fields the feature extractor needs are read; `rssi` /
/// `noise_floor` / `features` are carried for schema fidelity.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RecordedFrame {
pub timestamp: f64,
pub subcarriers: Vec<f64>,
#[serde(default)]
pub rssi: f64,
#[serde(default)]
pub noise_floor: f64,
#[serde(default)]
pub features: serde_json::Value,
}
/// Training configuration submitted with a start request.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TrainingConfig {
@@ -229,24 +271,45 @@ pub struct TrainingProgress {
}
/// Runtime training state stored in `AppStateInner`.
///
/// `status` and `cancel` are shared handles (not owned snapshots) so the
/// background training job can update progress and observe stop requests
/// **without holding a reference to the full `AppStateInner`**. That decoupling
/// is what makes the training core ([`run_training_job`]) unit-testable in
/// isolation from the ~60-field server state.
pub struct TrainingState {
/// Current status snapshot.
pub status: TrainingStatus,
/// Handle to the background training task (for cancellation).
/// Live status snapshot, shared with the running training job.
pub status: Arc<Mutex<TrainingStatus>>,
/// Cooperative stop flag; `stop_training` sets it and the job loop observes it.
pub cancel: Arc<AtomicBool>,
/// Handle to the background training task.
pub task_handle: Option<tokio::task::JoinHandle<()>>,
}
impl Default for TrainingState {
fn default() -> Self {
Self {
status: TrainingStatus::default(),
status: Arc::new(Mutex::new(TrainingStatus::default())),
cancel: Arc::new(AtomicBool::new(false)),
task_handle: None,
}
}
}
impl TrainingState {
/// Clone of the current status snapshot.
pub fn snapshot(&self) -> TrainingStatus {
self.status.lock().unwrap().clone()
}
/// Whether a training job is currently active.
pub fn is_active(&self) -> bool {
self.status.lock().unwrap().active
}
}
/// Shared application state type.
pub type AppState = Arc<RwLock<super::AppStateInner>>;
pub type AppState = Arc<tokio::sync::RwLock<super::AppStateInner>>;
/// Feature normalization statistics computed from the training set.
/// Stored alongside the model weights inside the .rvf container so that
@@ -317,11 +380,11 @@ async fn load_recording_frames(dataset_ids: &[String]) -> Vec<RecordedFrame> {
all_frames
}
/// Attempt to collect frames from the live frame_history buffer in AppState.
/// Each `Vec<f64>` in frame_history is a subcarrier amplitude vector.
async fn load_frames_from_history(state: &AppState) -> Vec<RecordedFrame> {
let s = state.read().await;
let history: &VecDeque<Vec<f64>> = &s.frame_history;
/// Build fallback training frames from a snapshot of the live `frame_history`
/// buffer. Each `Vec<f64>` is one frame's subcarrier amplitude vector. Passed as
/// an owned snapshot (not a live `AppState` borrow) so the training core stays
/// state-free and independently testable.
fn frames_from_history(history: &[Vec<f64>]) -> Vec<RecordedFrame> {
history
.iter()
.enumerate()
@@ -938,13 +1001,15 @@ fn deterministic_shuffle(n: usize, seed: u64) -> Vec<usize> {
/// linear model via mini-batch gradient descent.
///
/// On completion, exports a `.rvf` container with real calibrated weights.
async fn real_training_loop(
state: AppState,
async fn run_training_job(
status: Arc<Mutex<TrainingStatus>>,
cancel: Arc<AtomicBool>,
progress_tx: broadcast::Sender<String>,
config: TrainingConfig,
dataset_ids: Vec<String>,
history_snapshot: Vec<Vec<f64>>,
training_type: &str,
) {
) -> Option<PathBuf> {
let total_epochs = config.epochs;
let patience = config.early_stopping_patience;
let mut best_pck = 0.0f64;
@@ -978,7 +1043,7 @@ async fn real_training_loop(
let mut frames = load_recording_frames(&dataset_ids).await;
if frames.is_empty() {
info!("No recordings found for dataset_ids; falling back to live frame_history");
frames = load_frames_from_history(&state).await;
frames = frames_from_history(&history_snapshot);
}
if frames.len() < 10 {
@@ -999,11 +1064,12 @@ async fn real_training_loop(
if let Ok(json) = serde_json::to_string(&fail) {
let _ = progress_tx.send(json);
}
let mut s = state.write().await;
s.training_state.status.active = false;
s.training_state.status.phase = "failed".to_string();
s.training_state.task_handle = None;
return;
{
let mut st = status.lock().unwrap();
st.active = false;
st.phase = "failed".to_string();
}
return None;
}
info!("Loaded {} frames for training", frames.len());
@@ -1079,13 +1145,10 @@ async fn real_training_loop(
// ── Phase 5: Training loop ───────────────────────────────────────────────
for epoch in 1..=total_epochs {
// Check cancellation.
{
let s = state.read().await;
if !s.training_state.status.active {
info!("Training cancelled at epoch {epoch}");
break;
}
// Check cancellation (cooperative stop flag set by `stop_training`).
if cancel.load(Ordering::Relaxed) {
info!("Training cancelled at epoch {epoch}");
break;
}
let phase = if epoch <= config.warmup_epochs {
@@ -1245,10 +1308,10 @@ async fn real_training_loop(
let remaining = total_epochs.saturating_sub(epoch);
let eta_secs = (remaining as f64 * secs_per_epoch) as u64;
// Update shared state.
// Update the shared status snapshot (read by GET /api/v1/train/status).
{
let mut s = state.write().await;
s.training_state.status = TrainingStatus {
let mut st = status.lock().unwrap();
*st = TrainingStatus {
active: true,
epoch,
total_epochs,
@@ -1297,15 +1360,12 @@ async fn real_training_loop(
// ── Phase 6: Export .rvf model ───────────────────────────────────────────
let completed_phase;
{
let s = state.read().await;
completed_phase = if s.training_state.status.active {
"completed"
} else {
"cancelled"
};
}
let completed_phase = if cancel.load(Ordering::Relaxed) {
"cancelled"
} else {
"completed"
};
let mut written_rvf: Option<PathBuf> = None;
// Emit completion message.
let completion = TrainingProgress {
@@ -1326,11 +1386,7 @@ async fn real_training_loop(
if let Err(e) = tokio::fs::create_dir_all(MODELS_DIR).await {
error!("Failed to create models directory: {e}");
} else {
let model_id = format!(
"trained-{}-{}",
training_type,
chrono::Utc::now().format("%Y%m%d_%H%M%S")
);
let model_id = next_model_id(training_type);
let rvf_path = PathBuf::from(MODELS_DIR).join(format!("{model_id}.rvf"));
let mut builder = RvfBuilder::new();
@@ -1407,28 +1463,32 @@ async fn real_training_loop(
}),
);
if let Err(e) = builder.write_to_file(&rvf_path) {
error!("Failed to write trained model RVF: {e}");
} else {
info!(
"Trained model saved: {} ({} params, pck_torso_h@0.2={:.4})",
rvf_path.display(),
total_params,
best_pck
);
match builder.write_to_file(&rvf_path) {
Err(e) => {
error!("Failed to write trained model RVF: {e}");
}
Ok(()) => {
info!(
"Trained model saved: {} ({} params, pck_torso_h@0.2={:.4})",
rvf_path.display(),
total_params,
best_pck
);
written_rvf = Some(rvf_path);
}
}
}
}
// Mark training as inactive.
// Mark training as inactive in the shared status snapshot.
{
let mut s = state.write().await;
s.training_state.status.active = false;
s.training_state.status.phase = completed_phase.to_string();
s.training_state.task_handle = None;
let mut st = status.lock().unwrap();
st.active = false;
st.phase = completed_phase.to_string();
}
info!("Real {training_type} training finished: phase={completed_phase}");
written_rvf
}
// ── Public inference function ────────────────────────────────────────────────
@@ -1559,56 +1619,151 @@ fn default_keypoints() -> Vec<[f64; 4]> {
vec![[320.0, 240.0, 0.0, 0.0]; N_KEYPOINTS]
}
// ── Server-training enablement gate (ADR-186 P5) ─────────────────────────────
/// Env var that opts a deployment out of in-server training (e.g. the
/// lightweight appliance image without recordings). When set truthy, the start
/// endpoints return a structured `enabled:false` response pointing at the CLI —
/// never a silent `success:true` no-op.
const DISABLE_ENV: &str = "RUVIEW_DISABLE_SERVER_TRAINING";
/// Whether in-server training is enabled for this deployment.
fn server_training_enabled() -> bool {
training_enabled_from_env(std::env::var(DISABLE_ENV).ok().as_deref())
}
/// Pure decision (unit-testable without touching process env): enabled unless
/// the flag is a truthy disable value.
fn training_enabled_from_env(flag: Option<&str>) -> bool {
match flag {
Some(v) => {
let v = v.trim();
!(v == "1" || v.eq_ignore_ascii_case("true") || v.eq_ignore_ascii_case("yes"))
}
None => true,
}
}
/// Structured, honest "server training is off for this build — use the CLI"
/// response (HTTP 409). Guarantees no silent no-op in the disabled config.
fn disabled_response() -> Response {
(
StatusCode::CONFLICT,
Json(serde_json::json!({
"status": "error",
"enabled": false,
"reason": "In-server training is disabled for this deployment.",
"cli": "wifi-densepose train-room",
// `detail` is surfaced verbatim by the dashboard's API client.
"detail": "In-server training is disabled on this build. Train from the CLI: wifi-densepose train-room",
})),
)
.into_response()
}
// ── Axum handlers ────────────────────────────────────────────────────────────
async fn start_training(
State(state): State<AppState>,
Json(body): Json<StartTrainingRequest>,
) -> Json<serde_json::Value> {
// Check if training is already active.
{
let s = state.read().await;
if s.training_state.status.active {
return Json(serde_json::json!({
"status": "error",
"message": "Training is already active. Stop it first.",
"current_epoch": s.training_state.status.epoch,
"total_epochs": s.training_state.status.total_epochs,
}));
}
) -> Response {
if !server_training_enabled() {
return disabled_response();
}
let config = body.config.clone();
let dataset_ids = body.dataset_ids.clone();
match spawn_training_job(&state, config, body.dataset_ids.clone(), "supervised").await {
Ok(()) => Json(serde_json::json!({
"status": "started",
"type": "supervised",
"dataset_ids": body.dataset_ids,
"config": body.config,
}))
.into_response(),
Err(active) => Json(active_error(&active)).into_response(),
}
}
// Mark training as active and spawn background task.
let progress_tx;
{
/// Snapshot of the already-running job returned when a start is rejected.
fn active_error(snap: &TrainingStatus) -> serde_json::Value {
serde_json::json!({
"status": "error",
"message": "Training is already active. Stop it first.",
"current_epoch": snap.epoch,
"total_epochs": snap.total_epochs,
})
}
/// Seed the shared status, snapshot `frame_history`, and spawn the background
/// training job. Returns `Err(current_status)` if a job is already active.
///
/// Centralises the single-job guard + spawn used by the supervised, pretrain,
/// and LoRA start handlers so they cannot diverge.
/// Atomically claim the single training slot.
///
/// Checks `active` and sets it `true` **in one `status` lock scope**, so two
/// concurrent callers cannot both observe the slot free — the first claims it,
/// the second gets `Err(current_status)`. Returns the seeded status on success.
///
/// This is the fix for a TOCTOU race: the previous code checked `is_active()`
/// under a `state` READ lock, released it, and only afterward set `active`.
/// A `tokio::RwLock` read lock is shared, so two starts could both hold it, both
/// see the slot inactive, both proceed — spawning two jobs that then share and
/// overwrite one status/cancel and orphan a task handle. The claim's atomicity
/// lives on the `status` mutex, not the coarse `state` lock, which also keeps it
/// unit-testable without a full `AppState`.
fn claim_training_slot(
status: &Mutex<TrainingStatus>,
config: &TrainingConfig,
) -> Result<(), TrainingStatus> {
let mut st = status.lock().unwrap();
if st.active {
return Err(st.clone());
}
*st = TrainingStatus {
active: true,
total_epochs: config.epochs,
lr: config.learning_rate,
patience_remaining: config.early_stopping_patience,
phase: "initializing".to_string(),
..Default::default()
};
Ok(())
}
async fn spawn_training_job(
state: &AppState,
config: TrainingConfig,
dataset_ids: Vec<String>,
training_type: &'static str,
) -> Result<(), TrainingStatus> {
// Grab the shared handles under a read lock; the RwLock is only guarding
// access to the Arcs, not the single-job decision.
let (progress_tx, status, cancel, history_snapshot) = {
let s = state.read().await;
progress_tx = s.training_progress_tx.clone();
}
(
s.training_progress_tx.clone(),
s.training_state.status.clone(),
s.training_state.cancel.clone(),
s.frame_history.iter().cloned().collect::<Vec<_>>(),
)
};
{
let mut s = state.write().await;
s.training_state.status = TrainingStatus {
active: true,
epoch: 0,
total_epochs: config.epochs,
train_loss: 0.0,
val_pck: 0.0,
val_oks: 0.0,
lr: config.learning_rate,
best_pck: 0.0,
best_epoch: 0,
patience_remaining: config.early_stopping_patience,
eta_secs: None,
phase: "initializing".to_string(),
};
}
// Atomic check-and-set on the status mutex. This — not the read lock above —
// is what serialises concurrent starts (see `claim_training_slot`).
claim_training_slot(&status, &config)?;
cancel.store(false, Ordering::Relaxed);
let state_clone = state.clone();
let handle = tokio::spawn(async move {
real_training_loop(state_clone, progress_tx, config, dataset_ids, "supervised").await;
run_training_job(
status,
cancel,
progress_tx,
config,
dataset_ids,
history_snapshot,
training_type,
)
.await;
});
{
@@ -1616,57 +1771,58 @@ async fn start_training(
s.training_state.task_handle = Some(handle);
}
Json(serde_json::json!({
"status": "started",
"type": "supervised",
"dataset_ids": body.dataset_ids,
"config": body.config,
}))
Ok(())
}
async fn stop_training(State(state): State<AppState>) -> Json<serde_json::Value> {
let mut s = state.write().await;
if !s.training_state.status.active {
let s = state.read().await;
if !s.training_state.is_active() {
return Json(serde_json::json!({
"status": "error",
"message": "No training is currently active.",
}));
}
s.training_state.status.active = false;
s.training_state.status.phase = "stopping".to_string();
// The background task checks the active flag and will exit.
// We do not abort the handle -- we let it finish the current batch gracefully.
// Set the cooperative stop flag; the background job observes it between
// epochs and exits gracefully after the current batch. We do not abort the
// task handle.
s.training_state.cancel.store(true, Ordering::Relaxed);
{
let mut st = s.training_state.status.lock().unwrap();
st.phase = "stopping".to_string();
}
let snap = s.training_state.snapshot();
info!("Training stop requested");
Json(serde_json::json!({
"status": "stopping",
"epoch": s.training_state.status.epoch,
"best_pck": s.training_state.status.best_pck,
"epoch": snap.epoch,
"best_pck": snap.best_pck,
}))
}
async fn training_status(State(state): State<AppState>) -> Json<serde_json::Value> {
let s = state.read().await;
Json(serde_json::to_value(&s.training_state.status).unwrap_or_default())
let mut value = serde_json::to_value(s.training_state.snapshot()).unwrap_or_default();
// Surface the enablement flag so the dashboard can honestly disable the
// Start button (with a CLI tooltip) without first firing a POST (ADR-186 P5).
if let Some(obj) = value.as_object_mut() {
obj.insert(
"enabled".to_string(),
serde_json::Value::Bool(server_training_enabled()),
);
}
Json(value)
}
async fn start_pretrain(
State(state): State<AppState>,
Json(body): Json<PretrainRequest>,
) -> Json<serde_json::Value> {
{
let s = state.read().await;
if s.training_state.status.active {
return Json(serde_json::json!({
"status": "error",
"message": "Training is already active. Stop it first.",
}));
}
) -> Response {
if !server_training_enabled() {
return disabled_response();
}
let config = TrainingConfig {
epochs: body.epochs,
learning_rate: body.lr,
@@ -1675,56 +1831,26 @@ async fn start_pretrain(
..Default::default()
};
let progress_tx;
{
let s = state.read().await;
progress_tx = s.training_progress_tx.clone();
match spawn_training_job(&state, config, body.dataset_ids.clone(), "pretrain").await {
Ok(()) => Json(serde_json::json!({
"status": "started",
"type": "pretrain",
"epochs": body.epochs,
"lr": body.lr,
"dataset_ids": body.dataset_ids,
}))
.into_response(),
Err(active) => Json(active_error(&active)).into_response(),
}
{
let mut s = state.write().await;
s.training_state.status = TrainingStatus {
active: true,
total_epochs: body.epochs,
phase: "initializing".to_string(),
..Default::default()
};
}
let state_clone = state.clone();
let dataset_ids = body.dataset_ids.clone();
let handle = tokio::spawn(async move {
real_training_loop(state_clone, progress_tx, config, dataset_ids, "pretrain").await;
});
{
let mut s = state.write().await;
s.training_state.task_handle = Some(handle);
}
Json(serde_json::json!({
"status": "started",
"type": "pretrain",
"epochs": body.epochs,
"lr": body.lr,
"dataset_ids": body.dataset_ids,
}))
}
async fn start_lora_training(
State(state): State<AppState>,
Json(body): Json<LoraTrainRequest>,
) -> Json<serde_json::Value> {
{
let s = state.read().await;
if s.training_state.status.active {
return Json(serde_json::json!({
"status": "error",
"message": "Training is already active. Stop it first.",
}));
}
) -> Response {
if !server_training_enabled() {
return disabled_response();
}
let config = TrainingConfig {
epochs: body.epochs,
learning_rate: 0.0005, // lower LR for LoRA
@@ -1735,42 +1861,19 @@ async fn start_lora_training(
..Default::default()
};
let progress_tx;
{
let s = state.read().await;
progress_tx = s.training_progress_tx.clone();
match spawn_training_job(&state, config, body.dataset_ids.clone(), "lora").await {
Ok(()) => Json(serde_json::json!({
"status": "started",
"type": "lora",
"base_model_id": body.base_model_id,
"profile_name": body.profile_name,
"rank": body.rank,
"epochs": body.epochs,
"dataset_ids": body.dataset_ids,
}))
.into_response(),
Err(active) => Json(active_error(&active)).into_response(),
}
{
let mut s = state.write().await;
s.training_state.status = TrainingStatus {
active: true,
total_epochs: body.epochs,
phase: "initializing".to_string(),
..Default::default()
};
}
let state_clone = state.clone();
let dataset_ids = body.dataset_ids.clone();
let handle = tokio::spawn(async move {
real_training_loop(state_clone, progress_tx, config, dataset_ids, "lora").await;
});
{
let mut s = state.write().await;
s.training_state.task_handle = Some(handle);
}
Json(serde_json::json!({
"status": "started",
"type": "lora",
"base_model_id": body.base_model_id,
"profile_name": body.profile_name,
"rank": body.rank,
"epochs": body.epochs,
"dataset_ids": body.dataset_ids,
}))
}
// ── WebSocket handler for training progress ──────────────────────────────────
@@ -1792,8 +1895,11 @@ async fn handle_train_ws_client(mut socket: WebSocket, state: AppState) {
// Send current status immediately.
{
let s = state.read().await;
if let Ok(json) = serde_json::to_string(&s.training_state.status) {
let snapshot = {
let s = state.read().await;
s.training_state.snapshot()
};
if let Ok(json) = serde_json::to_string(&snapshot) {
let msg = serde_json::json!({
"type": "status",
"data": serde_json::from_str::<serde_json::Value>(&json).unwrap_or_default(),
@@ -1869,6 +1975,60 @@ mod tests {
assert_eq!(status.phase, "idle");
}
#[test]
fn claim_training_slot_admits_exactly_one_concurrent_start() {
// Regression test for the single-job TOCTOU race. Many threads race to
// claim one slot at the same instant (a barrier maximises contention);
// the status mutex must admit EXACTLY ONE. A split check-then-set (the
// old shape) would let several through under load — verified by
// temporarily reverting the atomicity, which drops this from 1.
use std::sync::atomic::{AtomicUsize, Ordering as O};
use std::sync::{Arc, Barrier};
let status = Arc::new(Mutex::new(TrainingStatus::default()));
let config = TrainingConfig::default();
let winners = Arc::new(AtomicUsize::new(0));
const N: usize = 32;
let barrier = Arc::new(Barrier::new(N));
let mut handles = Vec::with_capacity(N);
for _ in 0..N {
let status = status.clone();
let config = config.clone();
let winners = winners.clone();
let barrier = barrier.clone();
handles.push(std::thread::spawn(move || {
barrier.wait();
if claim_training_slot(&status, &config).is_ok() {
winners.fetch_add(1, O::SeqCst);
}
}));
}
for h in handles {
h.join().unwrap();
}
assert_eq!(
winners.load(O::SeqCst),
1,
"exactly one concurrent start may claim the single training slot"
);
assert!(
status.lock().unwrap().active,
"the slot must be marked active after a successful claim"
);
}
#[test]
fn claim_training_slot_rejects_when_already_active() {
let status = Arc::new(Mutex::new(TrainingStatus::default()));
let config = TrainingConfig::default();
assert!(claim_training_slot(&status, &config).is_ok(), "first claim wins");
let err = claim_training_slot(&status, &config)
.expect_err("second claim must be refused while active");
assert!(err.active, "the rejection carries the active status");
}
#[test]
fn training_progress_serializes() {
let progress = TrainingProgress {
@@ -2132,4 +2292,169 @@ mod tests {
assert_eq!(parsed.n_features, 2);
assert_eq!(parsed.mean, vec![1.0, 2.0]);
}
/// Build a small deterministic set of synthetic CSI frames with enough
/// variation that feature extraction is non-degenerate.
fn synthetic_history(n: usize, n_sub: usize) -> Vec<Vec<f64>> {
(0..n)
.map(|i| {
(0..n_sub)
.map(|k| 10.0 + ((i as f64) * 0.3 + (k as f64) * 0.1).sin() * 2.0)
.collect()
})
.collect()
}
/// ADR-186 P3/P6 end-to-end: the real (state-free) training core must
/// (a) stream real progress events over the broadcast channel and
/// (b) actually write a `.rvf` model artifact on completion — not merely
/// flip a status flag. This is the regression guard that keeps the trainer
/// wired (the module was previously orphaned / uncompiled — ADR-186 §1.3).
#[tokio::test]
async fn training_job_streams_real_progress_and_writes_model() {
let history = synthetic_history(40, 56);
let (tx, mut rx) = broadcast::channel::<String>(1024);
let status = Arc::new(Mutex::new(TrainingStatus::default()));
let cancel = Arc::new(AtomicBool::new(false));
let config = TrainingConfig {
epochs: 3,
batch_size: 8,
warmup_epochs: 1,
early_stopping_patience: 10,
..Default::default()
};
// Empty dataset_ids → falls back to the in-memory history snapshot, so
// this test does not depend on the recordings directory.
let rvf = run_training_job(
status.clone(),
cancel,
tx,
config,
Vec::new(),
history,
"supervised",
)
.await;
// (b) A real model artifact was produced and exists on disk.
let rvf_path = rvf.expect("training must produce an .rvf model artifact");
assert!(
rvf_path.exists(),
"rvf artifact should exist at {}",
rvf_path.display()
);
// (a) Real progress frames were streamed, at least one carrying an epoch.
let mut n_frames = 0usize;
let mut saw_epoch = false;
let mut saw_completed = false;
while let Ok(msg) = rx.try_recv() {
n_frames += 1;
let v: serde_json::Value = serde_json::from_str(&msg).unwrap();
if v.get("epoch").and_then(|e| e.as_u64()).unwrap_or(0) >= 1 {
saw_epoch = true;
}
if v.get("phase").and_then(|p| p.as_str()) == Some("completed") {
saw_completed = true;
}
}
assert!(n_frames > 0, "expected streamed progress frames, got none");
assert!(saw_epoch, "expected at least one epoch-tagged progress frame");
assert!(saw_completed, "expected a terminal 'completed' progress frame");
// Final shared status reflects genuine completion, not just a flag flip:
// real epochs ran (the loop wrote per-epoch status) and a finite loss was
// computed from the real gradient-descent pass.
let final_status = status.lock().unwrap().clone();
assert!(!final_status.active, "job should be inactive when finished");
assert_eq!(final_status.phase, "completed");
assert!(
final_status.epoch >= 1,
"at least one real training epoch should have run"
);
assert!(
final_status.train_loss.is_finite(),
"a finite training loss should have been computed"
);
// Keep the test hermetic — remove the artifact it wrote.
let _ = std::fs::remove_file(&rvf_path);
}
/// ADR-186 P4 (path safety): a `dataset_id` containing directory traversal
/// is rejected before any file is opened, so the loader returns no frames
/// rather than reading an arbitrary file.
#[tokio::test]
async fn load_recording_frames_rejects_path_traversal() {
let frames = load_recording_frames(&["../../etc/passwd".to_string()]).await;
assert!(
frames.is_empty(),
"path-traversal dataset_id must yield no frames"
);
}
/// Exported model ids must be unique per call — a second-resolution
/// timestamp alone collided for runs finishing in the same wall-clock second
/// (silently overwriting each other's `.rvf`, which also flaked the
/// concurrent model-writing tests on CI). Guards against regressing the
/// filename scheme back to non-unique.
#[test]
fn model_ids_are_unique_per_call() {
let ids: Vec<String> = (0..1000).map(|_| next_model_id("supervised")).collect();
let unique: std::collections::HashSet<&String> = ids.iter().collect();
assert_eq!(unique.len(), ids.len(), "every model id must be distinct");
assert!(ids[0].starts_with("trained-supervised-"));
}
/// ADR-186 P5: the enablement gate is enabled by default and only disabled
/// by an explicit truthy opt-out, so a `--no-default-features` / default
/// build always has server training ON (no silent regression to disabled).
#[test]
fn training_enablement_gate() {
assert!(training_enabled_from_env(None), "default is enabled");
assert!(training_enabled_from_env(Some("0")), "0 keeps it enabled");
assert!(training_enabled_from_env(Some("")), "empty keeps it enabled");
assert!(!training_enabled_from_env(Some("1")), "1 disables");
assert!(!training_enabled_from_env(Some("true")), "true disables");
assert!(!training_enabled_from_env(Some("YES")), "case-insensitive");
assert!(!training_enabled_from_env(Some(" 1 ")), "trims whitespace");
}
/// A job that is cancelled before it starts still exits cleanly and reports
/// the `cancelled` terminal phase (drives `stop_training`'s cooperative flag).
#[tokio::test]
async fn training_job_honors_cancellation() {
let history = synthetic_history(40, 56);
let (tx, _rx) = broadcast::channel::<String>(1024);
let status = Arc::new(Mutex::new(TrainingStatus::default()));
let cancel = Arc::new(AtomicBool::new(true)); // pre-cancelled
let config = TrainingConfig {
epochs: 50,
batch_size: 8,
warmup_epochs: 1,
early_stopping_patience: 10,
..Default::default()
};
let rvf = run_training_job(
status.clone(),
cancel,
tx,
config,
Vec::new(),
history,
"supervised",
)
.await;
// Cancelled before the first epoch → no model, terminal phase cancelled.
assert!(rvf.is_none(), "cancelled run should not export a model");
let final_status = status.lock().unwrap().clone();
assert!(!final_status.active);
assert_eq!(final_status.phase, "cancelled");
}
}