1 files changed, 281 insertions, 0 deletions
diff --git a/server/src/runs.rs b/server/src/runs.rs
new file mode 100644
index 0000000..8bb934f
--- /dev/null
+++ b/server/src/runs.rs
@@ -0,0 +1,281 @@
+use axum::extract::{Path, Query};
+use axum::http::StatusCode;
+use axum::Json;
+use serde_json::{json, Value};
+
+use crate::auth::{audit, AuthUser};
+use crate::config::MAX_CODE_LEN;
+use crate::error::{AppError, AppResult};
+use crate::util::new_id;
+use crate::util::now_iso;
+
+pub use crate::state::Cx;
+
+const RUN_COLS: &str = "id,project_id,version_id,config,dataset_id,status,error,manifest_hash,created_at,result,started_at,finished_at";
+
+fn row_run(r: &rusqlite::Row) -> rusqlite::Result<Value> {
+    let config: String = r.get(3)?;
+    let result: Option<String> = r.get(9)?;
+    Ok(json!({
+        "id": r.get::<_, String>(0)?,
+        "project_id": r.get::<_, String>(1)?,
+        "version_id": r.get::<_, String>(2)?,
+        "config": serde_json::from_str::<Value>(&config).unwrap_or(Value::Null),
+        "dataset_id": r.get::<_, String>(4)?,
+        "status": r.get::<_, String>(5)?,
+        "error": r.get::<_, Option<String>>(6)?,
+        "manifest_hash": r.get::<_, Option<String>>(7)?,
+        "data_manifest_hash": r.get::<_, Option<String>>(7)?,
+        "created_at": r.get::<_, String>(8)?,
+        "result": result.and_then(|s| serde_json::from_str::<Value>(&s).ok()).unwrap_or(Value::Null),
+        "started_at": r.get::<_, Option<String>>(10)?,
+        "finished_at": r.get::<_, Option<String>>(11)?,
+    }))
+}
+
+fn select_run(db: &mut rusqlite::Connection, run_id: &str) -> AppResult<Value> {
+    let cols = RUN_COLS;
+    db.query_row(format!("SELECT {cols} FROM runs WHERE id=?1").as_str(), [run_id], row_run)
+        .map_err(|_| AppError::not_found("run not found"))
+}
+
+async fn assert_run_owner(cx: &Cx, user_id: &str, run_id: &str) -> AppResult<()> {
+    let owner: Option<String> = cx.with_db(|db| {
+        db.query_row("SELECT user_id FROM runs WHERE id=?1", [run_id], |r| r.get(0)).ok()
+    }).await;
+    match owner {
+        Some(u) if u == user_id => Ok(()),
+        _ => Err(AppError::not_found("run not found")),
+    }
+}
+
+pub async fn get(cx: Cx, auth: AuthUser, Path(id): Path<String>) -> AppResult<Json<Value>> {
+    assert_run_owner(&cx, &auth.id, &id).await?;
+    let v = cx.with_db(|db| select_run(db, &id)).await?;
+    Ok(Json(v))
+}
+
+pub async fn list(cx: Cx, auth: AuthUser, q: Query<std::collections::HashMap<String, String>>) -> AppResult<Json<Value>> {
+    let project = q.get("project_id").cloned();
+    let cols = RUN_COLS;
+    let items: Vec<Value> = cx.with_db(|db| -> AppResult<Vec<Value>> {
+        let owned = project.as_deref();
+        let sql = if owned.is_some() {
+            format!("SELECT {cols} FROM runs WHERE user_id=?1 AND project_id=?2 ORDER BY created_at DESC LIMIT 200")
+        } else {
+            format!("SELECT {cols} FROM runs WHERE user_id=?1 ORDER BY created_at DESC LIMIT 200")
+        };
+        let mut st = db.prepare(&sql)?;
+        let mut rows = if let Some(p) = owned {
+            st.query(rusqlite::params![auth.id, p])?
+        } else {
+            st.query([auth.id.clone()])?
+        };
+        let mut out = Vec::new();
+        while let Some(r) = rows.next()? { out.push(row_run(r)?); }
+        Ok(out)
+    }).await?;
+    Ok(Json(json!({ "items": items })))
+}
+
+fn build_config(j: &Value) -> AppResult<Value> {
+    let g = |k: &str, d: f64| j.get(k).and_then(|v| v.as_f64()).unwrap_or(d);
+    let capital = j.get("capital").and_then(|v| v.as_f64());
+    let config = json!({
+        "capital": capital.unwrap_or(1_000_000.0),
+        "commission": g("commission", 0.0003),
+        "slippage": g("slippage", 0.001),
+        "benchmark_symbol": j.get("benchmark_symbol").cloned().unwrap_or(Value::Null),
+        "parameters": j.get("parameters").cloned().unwrap_or(json!({})),
+        "seed": j.get("seed").cloned().unwrap_or(Value::Null),
+    });
+    let cap = config.get("capital").and_then(|v| v.as_f64()).unwrap();
+    if !cap.is_finite() || !(1.0..=1e12).contains(&cap) {
+        return Err(AppError::bad("validation", "capital must be between 1 and 1e12"));
+    }
+    for k in ["commission", "slippage"] {
+        let v = config.get(k).and_then(|v| v.as_f64()).unwrap();
+        if !v.is_finite() || !(0.0..=0.05).contains(&v) {
+            return Err(AppError::bad("validation", format!("{k} must be between 0 and 0.05")));
+        }
+    }
+    if config.get("parameters").and_then(|p| p.as_object()).map(|o| !o.is_empty()).unwrap_or(false) {
+        if serde_json::to_string(config.get("parameters").unwrap()).unwrap_or_default().len() > MAX_CODE_LEN {
+            return Err(AppError::bad("validation", "parameters too large"));
+        }
+    }
+    Ok(config)
+}
+
+/// Insert a queued run inside one transaction with an atomic per-user daily quota check.
+async fn insert_run(cx: &Cx, auth: &AuthUser, project_id: &str, dataset_id: &str, version_id: &str, config: &Value, manifest_hash: &Option<String>) -> AppResult<String> {
+    let uid = auth.id.clone();
+    let cfg_s = config.to_string();
+    let limit = auth.daily_run_limit;
+    let today: String = chrono::Utc::now().format("%Y-%m-%d").to_string();
+    let pattern = format!("{today}%");
+    let id = new_id();
+    let now = now_iso();
+    cx.with_db(|db| -> AppResult<String> {
+        db.execute("BEGIN IMMEDIATE", []).ok();
+        let used: i64 = db.query_row(
+            "SELECT COUNT(*) FROM runs WHERE user_id=?1 AND created_at LIKE ?2",
+            rusqlite::params![&uid, &pattern], |r| r.get(0)).unwrap_or(0);
+        if used >= limit {
+            db.execute("ROLLBACK", []).ok();
+            return Err(AppError::new(axum::http::StatusCode::TOO_MANY_REQUESTS, "run_limit", "daily run quota reached"));
+        }
+        db.execute("INSERT INTO runs (id,user_id,project_id,version_id,dataset_id,status,config,manifest_hash,created_at) VALUES (?1,?2,?3,?4,?5,'queued',?6,?7,?8)",
+            rusqlite::params![&id, &uid, project_id, version_id, dataset_id, &cfg_s, manifest_hash, &now])?;
+        db.execute("COMMIT", []).map_err(AppError::from)?;
+        Ok(id)
+    }).await
+}
+
+pub async fn enqueue(cx: Cx, auth: AuthUser, body: Option<Json<serde_json::Value>>) -> AppResult<(StatusCode, Json<Value>)> {
+    let Json(j) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?;
+    let project_id = j.get("project_id").and_then(|v| v.as_str()).ok_or_else(|| AppError::bad("validation", "project_id required"))?.to_string();
+    let dataset_id = j.get("dataset_id").and_then(|v| v.as_str()).ok_or_else(|| AppError::bad("validation", "dataset_id required"))?.to_string();
+
+    // dataset must exist, belong to the caller and be ready
+    let (ready, manifest_hash): (bool, Option<String>) = cx.with_db(|db| {
+        db.query_row("SELECT status='ready', manifest_hash FROM datasets WHERE id=?1 AND user_id=?2",
+            rusqlite::params![&dataset_id, &auth.id], |r| Ok((r.get::<_, i64>(0)? != 0, r.get(1)?)))
+            .map_err(|_| AppError::not_found("dataset not found"))
+    }).await?;
+    if !ready { return Err(AppError::conflict("dataset_not_ready", "dataset pending or failed; cannot run")); }
+
+    // snapshot + pin draft
+    let draft: String = cx.with_db(|db| {
+        db.query_row("SELECT draft_code FROM projects WHERE id=?1 AND user_id=?2", rusqlite::params![&project_id, &auth.id], |r| r.get(0))
+            .map_err(|_| AppError::not_found("project not found"))
+    }).await?;
+    if draft.trim().is_empty() {
+        return Err(AppError::bad("validation", "draft is empty; cannot run empty strategy"));
+    }
+
+    let config = build_config(&j)?;
+    let version_id = push_run_version(&cx, &auth.id, &project_id, &draft).await?;
+
+    // warnings need explicit acknowledgement
+    let warnings: Value = cx.with_db(|db| {
+        let m: String = db.query_row("SELECT manifest FROM datasets WHERE id=?1", [&dataset_id], |r| r.get(0)).unwrap_or_default();
+        let mj: Value = serde_json::from_str(&m).unwrap_or(Value::Null);
+        Ok::<_, AppError>(mj.get("warnings").cloned().unwrap_or(json!([])))
+    }).await?;
+    let mut warnings = warnings;
+    if let Some(objs) = warnings.as_array_mut() {
+        // explicit unsupported-restriction surfacing for index feeds
+        let has_index: bool = cx.with_db(|db| {
+            let m: String = db.query_row("SELECT manifest FROM datasets WHERE id=?1", [&dataset_id], |r| r.get(0)).unwrap_or_default();
+            let mj: Value = serde_json::from_str(&m).unwrap_or(Value::Null);
+            Ok::<_, AppError>(mj.get("objects").and_then(|o| o.as_array()).map(|a| {
+                a.iter().any(|o| o.get("instrument").and_then(|i| i.get("asset_type")).and_then(|t| t.as_str()) == Some("index"))
+            }).unwrap_or(false))
+        }).await?;
+        if has_index && !objs.iter().any(|w| w.as_str().map(|s| s.contains("index feeds are nontradable")).unwrap_or(false)) {
+            objs.push(json!("index feeds are nontradable research proxies; direct index orders are rejected by the engine"));
+        }
+    }
+    let has_warns = warnings.as_array().map(|a| !a.is_empty()).unwrap_or(false);
+    if has_warns && !j.get("acknowledge_warnings").and_then(|v| v.as_bool()).unwrap_or(false) {
+        return Err(AppError::conflict("warnings_unacknowledged", "dataset has warnings; acknowledge to run")
+            .with_details(json!({"warnings": warnings})));
+    }
+
+    let run_id = insert_run(&cx, &auth, &project_id, &dataset_id, &version_id, &config, &manifest_hash).await?;
+    audit(&cx, Some(&auth.id), "run_create", &run_id, "ok").await;
+    let v = cx.with_db(|db| select_run(db, &run_id)).await?;
+    Ok((StatusCode::ACCEPTED, Json(v)))
+}
+
+/// Snapshot the draft as a pinned run version, reusing an identical hash version.
+async fn push_run_version(cx: &Cx, _user_id: &str, project_id: &str, code: &str) -> AppResult<String> {
+    let hash = crate::util::sha256_hex(code.as_bytes());
+    let code = code.to_string();
+    cx.with_db(|db| -> AppResult<String> {
+        let vid: Option<String> = db.query_row("SELECT id FROM project_versions WHERE project_id=?1 AND hash=?2",
+            rusqlite::params![project_id, &hash], |r| r.get(0)).ok();
+        if let Some(v) = vid { return Ok(v); }
+        let id = new_id();
+        db.execute("INSERT INTO project_versions (id,project_id,code,hash,message,source,created_at) VALUES (?1,?2,?3,?4,'run snapshot','run',?5)",
+            rusqlite::params![&id, project_id, &code, &hash, now_iso()])?;
+        Ok(id)
+    }).await
+}
+
+pub async fn cancel(cx: Cx, auth: AuthUser, Path(id): Path<String>) -> AppResult<Json<Value>> {
+    assert_run_owner(&cx, &auth.id, &id).await?;
+    let status: Option<String> = cx.with_db(|db| {
+        db.query_row("SELECT status FROM runs WHERE id=?1", [&id], |r| r.get(0)).ok()
+    }).await;
+    if matches!(status.as_deref(), Some("queued") | Some("running")) {
+        let res = crate::jobs::signal_cancel(&cx, &id).await;
+        audit(&cx, Some(&auth.id), "run_cancel", &id, if res { "ok" } else { "not_running" }).await;
+    }
+    let v = cx.with_db(|db| select_run(db, &id)).await?;
+    Ok(Json(v))
+}
+
+/// Fresh run pinned to the original code+config+dataset; never refetches data.
+pub async fn rerun(cx: Cx, auth: AuthUser, Path(id): Path<String>, body: Option<Json<serde_json::Value>>) -> AppResult<(StatusCode, Json<Value>)> {
+    assert_run_owner(&cx, &auth.id, &id).await?;
+    let j = body.map(|b| b.0).unwrap_or(json!({}));
+    if j.get("use_original_data").and_then(|v| v.as_bool()) != Some(true) {
+        return Err(AppError::bad("validation", "use_original_data must be true; rerun is pinned to original data"));
+    }
+    let (project_id, version_id, dataset_id, config, manifest_hash): (String, String, String, String, Option<String>) = cx.with_db(|db| {
+        db.query_row("SELECT project_id,version_id,dataset_id,config,manifest_hash FROM runs WHERE id=?1", [&id], |r|
+            Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?, r.get::<_, String>(2)?, r.get::<_, String>(3)?, r.get::<_, Option<String>>(4)?)))
+            .map_err(|_| AppError::not_found("run not found"))
+    }).await?;
+    // dataset must still be ready and owned
+    let ready: bool = cx.with_db(|db| {
+        db.query_row("SELECT status='ready' FROM datasets WHERE id=?1 AND user_id=?2",
+            rusqlite::params![&dataset_id, &auth.id], |r| r.get(0))
+            .map_err(|_| AppError::not_found("dataset not found"))
+    }).await?;
+    if !ready { return Err(AppError::conflict("dataset_not_ready", "original dataset is missing or not ready")); }
+    let cfg_val: Value = serde_json::from_str(&config).unwrap_or(json!({}));
+    let new_id_r = insert_run(&cx, &auth, &project_id, &dataset_id, &version_id, &cfg_val, &manifest_hash).await?;
+    audit(&cx, Some(&auth.id), "run_rerun", &new_id_r, "ok").await;
+    let v = cx.with_db(|db| select_run(db, &new_id_r)).await?;
+    Ok((StatusCode::ACCEPTED, Json(v)))
+}
+
+/// Mark running runs failed after a server restart (queued stays resumable).
+pub async fn cleanup_interrupted(cx: &Cx) {
+    cx.with_db(|db| {
+        db.execute("UPDATE runs SET status='failed', error='interrupted by server restart', finished_at=?1 WHERE status='running'",
+            rusqlite::params![now_iso()]).ok();
+    }).await;
+}
+
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn config_bounds_enforced() {
+        let ok = json!({"capital": 100000, "commission": 0.0003, "slippage": 0.001});
+        assert!(build_config(&ok).is_ok());
+        let bad = json!({"capital": 0.0});
+        assert_eq!(build_config(&bad).unwrap_err().code, "validation");
+        let bad = json!({"capital": 5000, "commission": 0.5});
+        assert_eq!(build_config(&bad).unwrap_err().code, "validation");
+        let big = json!({"capital": 100000, "parameters": {"huge": "x".repeat(MAX_CODE_LEN + 1)}});
+        assert_eq!(build_config(&big).unwrap_err().code, "validation");
+        let defaults = build_config(&json!({})).unwrap();
+        assert_eq!(defaults["commission"], json!(0.0003), "defaults are visible to users");
+    }
+
+    #[test]
+    fn nonfinite_sanitized() {
+        let mut v = json!({"metrics": {"sharpe": f64::NAN, "total_return": 1.0}, "equity": [1.0, f64::INFINITY]});
+        crate::jobs::sanitize_nonfinite(&mut v);
+        let s = serde_json::to_string(&v).unwrap();
+        assert!(!s.contains("NaN") && !s.contains("Infinity"), "results must never carry NaN/Inf");
+        assert_eq!(v["metrics"]["sharpe"], Value::Null);
+    }
+}