chore: establish Strategy Lab source baseline (development, not release)

1 files changed, 857 insertions, 0 deletions

diff --git a/server/src/jobs.rs b/server/src/jobs.rs
new file mode 100644
index 0000000..a3c49eb
--- /dev/null
+++ b/server/src/jobs.rs
@@ -0,0 +1,857 @@
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::time::Duration;
+
+use tokio::sync::Mutex;
+use std::collections::HashSet;
+use serde_json::{json, Value};
+
+use crate::auth::audit;
+use crate::error::AppError;
+use crate::state::AppState;
+use crate::store::ObjectStore;
+use crate::util::{canonical_json, new_id, now_iso, sha256_hex};
+
+pub struct Signals;
+
+impl Signals {
+    pub fn new() -> Self { Signals }
+}
+
+pub async fn main_loop(cx: Arc<AppState>, _signals: Signals) {
+    loop {
+        if let Err(e) = tick(&cx).await { tracing::error!("job loop: {e}"); }
+        tokio::time::sleep(Duration::from_millis(700)).await;
+    }
+}
+
+// ---- shared in-flight fetch serialization for the exact request cache key ----
+
+fn inflight() -> &'static Mutex<HashSet<String>> {
+    static SET: std::sync::OnceLock<Mutex<HashSet<String>>> = std::sync::OnceLock::new();
+    SET.get_or_init(|| Mutex::new(HashSet::new()))
+}
+
+pub fn cache_key_for_request(req: &Value) -> String {
+    sha256_hex(canonical_json(req).as_bytes())
+}
+
+async fn release_key(key: &str) {
+    inflight().lock().await.remove(key);
+}
+
+// ---- dispatcher ----
+
+async fn tick(cx: &Arc<AppState>) -> Result<(), AppError> {
+    // pending datasets -> bounded fetch concurrency
+    let n_running: i64 = cx.with_db(|db| {
+        db.query_row("SELECT COUNT(*) FROM datasets WHERE status='running'", [], |r| r.get(0))
+    }).await.unwrap_or(0);
+    if n_running < cx.cfg.fetch_concurrency as i64 {
+        let next: Option<(String, String)> = cx.with_db(|db| {
+            db.query_row("SELECT id,request FROM datasets WHERE status='pending' ORDER BY created_at LIMIT 1", [],
+                |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?))).ok()
+        }).await;
+        if let Some((did, request)) = next {
+            cx.with_db(|db| {
+                db.execute("UPDATE datasets SET status='running', updated_at=?1 WHERE id=?2 AND status='pending'",
+                    rusqlite::params![now_iso(), &did]).ok();
+            }).await;
+            let cx2 = cx.clone();
+            tokio::spawn(async move {
+                if let Err(e) = spawn_fetch_job(&cx2, &did, &request).await {
+                    tracing::error!("fetch {did}: {e}");
+                    cx2.with_db(|db| {
+                        db.execute("UPDATE datasets SET status='failed', error=?1, updated_at=?2 WHERE id=?3 AND status IN ('pending','running')",
+                            rusqlite::params![e.message, now_iso(), &did]).ok();
+                    }).await;
+                }
+            });
+        }
+    }
+
+    // queued runs -> single active backtest worker (POC)
+    let n_running_runs: i64 = cx.with_db(|db| {
+        db.query_row("SELECT COUNT(*) FROM runs WHERE status='running'", [], |r| r.get(0))
+    }).await.unwrap_or(0);
+    if n_running_runs >= cx.cfg.run_concurrency as i64 { return Ok(()); }
+    let next: Option<(String, String)> = cx.with_db(|db| {
+        db.query_row("SELECT id,config FROM runs WHERE status='queued' ORDER BY created_at LIMIT 1", [],
+            |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?))).ok()
+    }).await;
+    let Some((run_id, config)) = next else { return Ok(()); };
+    let container_name = format!("sl-run-{}", new_id());
+    // Atomic claim: queued -> running WITH container_id set in the same transaction.
+    // A run never becomes 'running' without a container identity, so the claim
+    // itself proves the worker task was handed off.
+    let claimed = cx.with_db(|db| -> Result<bool, AppError> {
+        move_claim(&mut *db, &run_id, &container_name)
+    }).await?;
+    if !claimed { return Ok(()); }
+    let cx2 = cx.clone();
+    tokio::spawn(async move {
+        let user_id: String = cx2.with_db(|db| {
+            db.query_row("SELECT user_id FROM runs WHERE id=?1", [&run_id], |r| r.get(0)).unwrap_or_default()
+        }).await;
+        let res = run_backtest(&cx2, &run_id, &config, &container_name).await;
+        let _ = finalize_run(&cx2, &run_id, &user_id, res).await;
+    });
+    Ok(())
+}
+
+/// Recheck account/quota/dataset durability, then claim queued -> running with
+/// a container id atomically inside one transaction.
+/// Returns true only if this caller actually claimed the run (and must spawn the worker).
+fn move_claim(db: &mut rusqlite::Connection, run_id: &str, container_id: &str) -> Result<bool, AppError> {
+    db.execute("BEGIN IMMEDIATE", []).ok();
+    match db.query_row(
+        "SELECT u.active, r.user_id FROM runs r JOIN users u ON u.id=r.user_id WHERE r.id=?1",
+        [run_id], |r| Ok((r.get::<_, i64>(0)?, r.get::<_, String>(1)?))) {
+        Ok((active, _uid)) if active == 1 => {},
+        _ => {
+            db.execute("UPDATE runs SET status='failed', error='account disabled', finished_at=?1 WHERE id=?2 AND status='queued'",
+                rusqlite::params![now_iso(), run_id]).ok();
+            db.execute("COMMIT", []).ok();
+            return Ok(false);
+        }
+    }
+    let uid = db.query_row("SELECT user_id FROM runs WHERE id=?1", [run_id], |r| r.get::<_, String>(0)).unwrap_or_default();
+    let ready_ok: Option<i64> = db.query_row(
+        "SELECT d.status='ready' AS ok FROM datasets d JOIN runs r ON r.dataset_id=d.id AND r.user_id=d.user_id WHERE r.id=?1",
+        [run_id], |r| r.get::<_, i64>(0)).ok();
+    if ready_ok != Some(1) {
+        db.execute("UPDATE runs SET status='failed', error='dataset no longer ready at dequeue', finished_at=?1 WHERE id=?2 AND status='queued'",
+            rusqlite::params![now_iso(), run_id]).ok();
+        db.execute("COMMIT", []).ok();
+        return Ok(false);
+    }
+    let today = format!("{}%", chrono::Utc::now().format("%Y-%m-%d").to_string());
+    let used: i64 = db.query_row("SELECT COUNT(*) FROM runs WHERE user_id=?1 AND created_at LIKE ?2",
+        rusqlite::params![&uid, &today], |r| r.get(0)).unwrap_or(0);
+    let limit: i64 = db.query_row("SELECT daily_run_limit FROM users WHERE id=?1", [&uid], |r| r.get(0)).unwrap_or(10);
+    if used > limit {
+        db.execute("UPDATE runs SET status='failed', error='run quota exceeded', finished_at=?1 WHERE id=?2 AND status='queued'",
+            rusqlite::params![now_iso(), run_id]).ok();
+        db.execute("COMMIT", []).ok();
+        return Ok(false);
+    }
+    // Single atomic transition: a claimed run is running WITH its container identity.
+    let n = db.execute(
+        "UPDATE runs SET status='running', started_at=?1, container_id=?2 WHERE id=?3 AND status='queued'",
+        rusqlite::params![now_iso(), container_id, run_id]).unwrap_or(0);
+    db.execute("COMMIT", []).ok();
+    Ok(n == 1)
+}
+
+async fn finalize_run(cx: &Arc<AppState>, run_id: &str, user_id: &str, res: Result<Value, AppError>) {
+    cx.with_db(|db| -> Result<(), AppError> {
+        match res {
+            Ok(result) => {
+                let n = db.execute("UPDATE runs SET status='succeeded', result=?1, error=NULL, finished_at=?2 WHERE id=?3 AND status='running'",
+                    rusqlite::params![result.to_string(), now_iso(), run_id]).unwrap_or(0);
+                if n > 0 {
+                    let cxx = cx.clone();
+                    let uid = user_id.to_string();
+                    let rid = run_id.to_string();
+                    tokio::spawn(async move { audit(&axum::extract::State(cxx), Some(&uid), "run_succeeded", &rid, "ok").await });
+                }
+            }
+            Err(e) => {
+                // guaranteed non-empty terminal reason, never an empty failure
+                let msg = if e.message.trim().is_empty() { "worker failed without error detail".to_string() } else { e.message.clone() };
+                db.execute("UPDATE runs SET status='failed', error=?1, finished_at=?2 WHERE id=?3 AND status='running'",
+                    rusqlite::params![msg, now_iso(), run_id]).ok();
+            }
+        }
+        Ok(())
+    }).await.ok();
+}
+
+// ---- dataset fetch ----
+
+async fn try_cache(cx: &Arc<AppState>, dataset_id: &str, key: &str) -> Result<bool, AppError> {
+    let manifest: Option<String> = cx.with_db(|db| {
+        db.query_row("SELECT manifest FROM fetch_cache WHERE key=?1", [key], |r| r.get(0)).ok()
+    }).await;
+    let Some(m) = manifest else { return Ok(false); };
+    let mv: Value = serde_json::from_str(&m).unwrap_or(Value::Null);
+    let hash = mv.get("hash").and_then(|v| v.as_str()).unwrap_or_default().to_string();
+    cx.with_db(|db| {
+        db.execute("UPDATE datasets SET status='ready', error=NULL, manifest=?1, manifest_hash=?2, cache_hit=1, updated_at=?3 WHERE id=?4 AND status='running'",
+            rusqlite::params![&m, hash, now_iso(), dataset_id]).ok();
+    }).await;
+    Ok(true)
+}
+
+/// One dataset fetch: cache recheck or container run + artifact ingest.
+pub async fn spawn_fetch_job(cx: &Arc<AppState>, dataset_id: &str, request_s: &str) -> Result<(), AppError> {
+    let request: Value = serde_json::from_str(request_s).map_err(|e| AppError::internal(format!("stored request invalid: {e}")))?;
+    let key = cache_key_for_request(&request);
+
+    // serialize identical cache keys: second arrives late and rechecks cache
+    let deadline = std::time::Duration::from_secs(cx.cfg.fetch_timeout_secs.max(60));
+    let started = std::time::Instant::now();
+    loop {
+        {
+            let mut set = inflight().lock().await;
+            if set.insert(key.clone()) { break; }
+
+        }
+        if try_cache(cx, dataset_id, &key).await? { return Ok(()); }
+        if started.elapsed() > deadline {
+            return Err(AppError::internal("waiting on identical in-flight fetch timed out"));
+        }
+        tokio::time::sleep(Duration::from_millis(1500)).await;
+    }
+    let result = perform_fetch(cx, dataset_id, request, &key).await;
+    release_key(&key).await;
+    result
+}
+
+async fn perform_fetch(cx: &Arc<AppState>, dataset_id: &str, request: Value, key: &str) -> Result<(), AppError> {
+    // cached from a prior identical request?
+    if try_cache(cx, dataset_id, key).await? { return Ok(()); }
+
+    let store = ObjectStore::new(&cx.cfg.data_dir);
+    let work = temp_job_dir(&cx.cfg.data_dir)?;
+    let input = work.join("input");
+    let output = work.join("output");
+    std::fs::create_dir_all(&input).map_err(io_err)?;
+    std::fs::create_dir_all(&output).map_err(io_err)?;
+    open_writable(&output)?;
+
+    std::fs::write(input.join("request.json"), request.to_string()).map_err(io_err)?;
+    let mounts = vec![
+        (input.display().to_string(), "/input".to_string(), true),
+        (output.display().to_string(), "/output".to_string(), false),
+    ];
+    let args = vec![
+        "python".into(), "-m".into(), "worker.main".into(), "fetch".into(),
+        "--request".into(), "/input/request.json".into(), "--output".into(), "/output".into(),
+    ];
+    let name = format!("sl-fetch-{}", new_id());
+    let res = crate::worker::run_named(&*cx, true, &mounts, &args, &name, cx.cfg.fetch_timeout_secs).await;
+    let out_take = match res {
+        Ok(r) if r.ok() => r,
+        Ok(r) => {
+            let msg = worker_error_message(&output, r.stderr.as_str());
+            cleanup(&work);
+            return Err(AppError::bad("fetch_failed", msg));
+        }
+        Err(e) => { cleanup(&work); return Err(e); }
+    };
+
+    let build = ingest_fetch_output(&store, &work).await;
+    let (manifest, _warnings) = match build {
+        Ok(v) => v,
+        Err(e) => { cleanup(&work); return Err(e); }
+    };
+    let _ = out_take;
+    let manifest_hash = manifest.get("hash").and_then(|v| v.as_str()).unwrap_or("").to_string();
+    if manifest_hash.is_empty() || manifest.get("objects").and_then(|o| o.as_array()).map(|a| a.is_empty()).unwrap_or(true) {
+        cleanup(&work);
+        return Err(AppError::internal("fetch produced no usable objects"));
+    }
+    let manifest_s = manifest.to_string();
+    let object_count = manifest.get("objects").and_then(|o| o.as_array()).map(|a| a.len()).unwrap_or(0) as i64;
+    cx.with_db(|db| -> Result<(), AppError> {
+        db.execute("INSERT OR REPLACE INTO fetch_cache (key,manifest_hash,manifest,object_count,fetched_at) VALUES (?1,?2,?3,?4,?5)",
+            rusqlite::params![&key, &manifest_hash, &manifest_s, object_count, now_iso()])?;
+        db.execute("UPDATE datasets SET status='ready', error=NULL, manifest=?1, manifest_hash=?2, cache_hit=0, updated_at=?3 WHERE id=?4 AND status='running'",
+            rusqlite::params![&manifest_s, &manifest_hash, now_iso(), dataset_id])?;
+        Ok(())
+    }).await?;
+    cleanup(&work);
+    tracing::info!("fetch {dataset_id} cached under {key} ({object_count} objects)");
+    Ok(())
+}
+
+
+// manifest hash must be content identity: independent of user/request ids and fetch timestamp
+fn manifest_content_hash(m: &Value) -> String {
+    let mut objects = Vec::new();
+    for o in m.get("objects").and_then(|v| v.as_array()).unwrap_or(&vec![]).iter() {
+        let mut e = o.clone();
+        if let Some(map) = e.as_object_mut() {
+            map.remove("fetched_at");
+            map.remove("path");
+        }
+        objects.push(e);
+    }
+    let payload = json!({
+        "frequency": m.get("frequency"),
+        "adjustment": m.get("adjustment"),
+        "schema_version": m.get("schema_version"),
+        "normalization_version": m.get("normalization_version"),
+        "objects": objects,
+    });
+    sha256_hex(canonical_json(&payload).as_bytes())
+}
+
+
+async fn ingest_fetch_output(store: &ObjectStore, work: &Path) -> Result<(Value, Vec<String>), AppError> {
+    let out_dir = work.join("output");
+    let result_path = out_dir.join("result.json");
+    let result: Value = serde_json::from_str(&std::fs::read_to_string(&result_path).map_err(|e| AppError::internal(format!("worker result.json unreadable: {e}")))?)
+        .map_err(|e| AppError::internal(format!("worker result.json invalid: {e}")))?;
+    if result.get("status").and_then(|v| v.as_str()) != Some("ready") {
+        return Err(AppError::bad("fetch_failed", format!("worker reported status: {}", result.get("status").and_then(|v| v.as_str()).unwrap_or("missing"))));
+    }
+    let m = result.get("manifest").cloned().unwrap_or(Value::Null);
+    if !m.get("objects").and_then(|o| o.as_array()).map(|a| !a.is_empty()).unwrap_or(false) {
+        return Err(AppError::bad("fetch_failed", "worker manifest has no objects"));
+    }
+    // hash+ingest every worker artifact once (raw JSON and normalized CSVs)
+    let files: Vec<crate::store::StoredObject> = store.ingest_directory(&out_dir).map_err(io_err)?;
+    let find_hash = |h: &str| files.iter().find(|f| f.hash == h);
+    let find_rel = |rel: &str| files.iter().find(|f| f.stored_path == rel || f.mount_name == rel);
+
+    let all_warnings: Vec<String> = result.get("warnings").and_then(|v| v.as_array()).map(|a| {
+        a.iter().filter_map(|w| w.as_str().map(String::from)).collect()
+    }).unwrap_or_default();
+    let mut new_objects = Vec::new();
+    for obj in m.get("objects").and_then(|v| v.as_array()).cloned().unwrap_or_default() {
+        let mut entry = obj.clone();
+        // normalized: rewrite the internal path into immutable object storage
+        let rel_norm = entry.get("path").and_then(|v| v.as_str()).unwrap_or_default().to_string();
+        if rel_norm.contains("..") {
+            return Err(AppError::internal("worker output path rejected"));
+        }
+        let norm_obj = find_rel(&rel_norm).ok_or_else(|| AppError::internal("normalized object missing from worker output"))?;
+        let rel_stored = norm_obj.stored_path.clone();
+        if let Some(map) = entry.as_object_mut() {
+            map.insert("path".into(), json!(format!("objects/{rel_stored}")));
+        }
+        // raw: locate by content hash of the raw object (worker writes immutable raw JSON first)
+        let raw_hash = entry.get("raw_object_hash").and_then(|v| v.as_str()).unwrap_or_default().to_string();
+        if !raw_hash.is_empty() {
+            if find_hash(&raw_hash).is_none() {
+                return Err(AppError::internal("raw object referenced by manifest is missing"));
+            }
+        }
+        new_objects.push(entry);
+    }
+
+    let mut manifest = m.clone();
+    {
+        let map = manifest.as_object_mut().unwrap();
+        map.insert("objects".into(), Value::Array(new_objects));
+        map.insert("warnings".into(), json!(all_warnings.clone()));
+        map.insert("immutable".into(), json!(true));
+    }
+    // stable content identity independent of request/user ids and fetch timestamp
+    {
+        let h = manifest_content_hash(&manifest);
+        if let Some(map) = manifest.as_object_mut() {
+            map.insert("hash".into(), json!(h));
+            // Per-instrument coverage rows aligned to the client CoverageEntry shape:
+            // {instrument, market, asset_type, requested_start, requested_end, actual_start, actual_end, row_count, warnings}
+            let coverage: Vec<Value> = map.get("objects").and_then(|o| o.as_array()).map(|objs| {
+                objs.iter().map(|o| {
+                    let inst = o.get("instrument").cloned().unwrap_or(Value::Null);
+                    json!({
+                        "instrument": inst.get("symbol").or_else(|| inst.get("instrument")).and_then(|v| v.as_str())
+                            .map(String::from).unwrap_or_default(),
+                        "market": inst.get("market").cloned().unwrap_or(Value::Null),
+                        "asset_type": inst.get("asset_type").cloned().unwrap_or(Value::Null),
+                        "requested_start": o.get("requested_start").cloned().unwrap_or(Value::Null),
+                        "requested_end": o.get("requested_end").cloned().unwrap_or(Value::Null),
+                        "actual_start": o.get("actual_start").cloned().unwrap_or(Value::Null),
+                        "actual_end": o.get("actual_end").cloned().unwrap_or(Value::Null),
+                        "row_count": o.get("row_count").cloned().unwrap_or(Value::Null),
+                        "warnings": o.get("warnings").cloned().unwrap_or(json!([])),
+                    })
+                }).collect()
+            }).unwrap_or_default();
+            let fallback_preview = json!({"columns": [], "rows": [], "coverage": coverage, "warnings": all_warnings.clone()});
+            let pv = match result.get("preview") {
+                Some(p) => {
+                    let mut p = p.clone();
+                    p["warnings"] = json!(all_warnings.clone());
+                    if !p.get("coverage").map(|c| c.is_array()).unwrap_or(false) {
+                        p["coverage"] = json!(coverage);
+                    }
+                    p
+                }
+                None => fallback_preview,
+            };
+            map.insert("preview".into(), pv);
+        }
+    }
+    Ok((manifest, all_warnings))
+}
+
+// ---- backtest ----
+
+pub async fn run_backtest(cx: &Arc<AppState>, run_id: &str, config_s: &str, container_name: &str) -> Result<Value, AppError> {
+    let (code, dataset_id): (String, String) = cx.with_db(|db| {
+        db.query_row("SELECT pv.code, r.dataset_id FROM runs r JOIN project_versions pv ON pv.id=r.version_id WHERE r.id=?1",
+            [run_id], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))
+            .map_err(|_| AppError::not_found("run not found"))
+    }).await?;
+    let manifest: Value = cx.with_db(|db| {
+        let m: String = db.query_row("SELECT manifest FROM datasets WHERE id=?1 AND status='ready'", [&dataset_id], |r| r.get(0))
+            .map_err(|_| AppError::conflict("dataset_not_ready", "dataset not ready"))?;
+        serde_json::from_str(&m).map_err(|e| AppError::internal(format!("manifest corrupt: {e}")))
+    }).await?;
+    let store = ObjectStore::new(&cx.cfg.data_dir);
+
+    let work = temp_job_dir(&cx.cfg.data_dir)?;
+    let input = work.join("input");
+    let output = work.join("output");
+    let r = (|| -> Result<(String, Vec<(String, String, bool)>), AppError> {
+        std::fs::create_dir_all(&input).map_err(io_err)?;
+        std::fs::create_dir_all(&output).map_err(io_err)?;
+        open_writable(&output)?;
+        let mut run_manifest = manifest.clone();
+        let mut mounts: Vec<(String, String, bool)> = Vec::new();
+        if let Some(objs) = run_manifest.get_mut("objects").and_then(|o| o.as_array_mut()) {
+            for obj in objs.iter_mut() {
+                let path = obj.get("path").and_then(|v| v.as_str()).unwrap_or_default().to_string();
+                // internal layout: objects/<hash-prefix>/<object>; reject traversal/symlinks
+                let rel = path.strip_prefix("objects/").unwrap_or("");
+                if rel.is_empty() || rel.contains("..") || rel.contains('/') && rel.len() < 4 {
+                    return Err(AppError::internal("invalid stored object path"));
+                }
+                let abs = store.absolute(rel);
+                if abs.is_symlink() || !abs.is_file() {
+                    return Err(AppError::internal("object file missing or not a regular file"));
+                }
+                let fname = abs.file_name().and_then(|f| f.to_str()).unwrap_or_default().to_string();
+                let dst = format!("/data/{fname}");
+                obj.as_object_mut().unwrap().insert("path".into(), json!(dst.clone()));
+                mounts.push((abs.display().to_string(), dst, true));
+            }
+        }
+        if run_manifest.get("objects").and_then(|o| o.as_array()).map(|a| a.is_empty()).unwrap_or(true) {
+            return Err(AppError::internal("dataset manifest has no mountable objects"));
+        }
+        std::fs::write(input.join("request.json"), json!({
+            "code": code.clone(),
+            "config": serde_json::from_str::<Value>(config_s).unwrap_or(Value::Null),
+            "dataset_manifest": run_manifest,
+            "data_root": "/data",
+        }).to_string()).map_err(io_err)?;
+        mounts.push((input.display().to_string(), "/input".into(), true));
+        mounts.push((output.display().to_string(), "/output".into(), false));
+        Ok((code, mounts))
+    })();
+    let (_code_owned, mounts) = match r {
+        Ok(v) => v,
+        Err(e) => { cleanup(&work); return Err(e); }
+    };
+    let args = vec![
+        "python".into(), "-m".into(), "worker.main".into(), "backtest".into(),
+        "--request".into(), "/input/request.json".into(), "--output".into(), "/output".into(),
+    ];
+    let res = crate::worker::run_named(&*cx, false, &mounts, &args, container_name, cx.cfg.backtest_timeout_secs).await;
+    let mut result: Value = match res {
+        Ok(r) if r.ok() => {
+            match std::fs::read_to_string(output.join("result.json")) {
+                Ok(s) => serde_json::from_str(&s).map_err(|e| AppError::internal(format!("backtest result invalid: {e}")))?,
+                Err(e) => { cleanup(&work); return Err(AppError::internal(format!("backtest result unreadable: {e}"))); }
+            }
+        }
+        Ok(r) => {
+            let msg = worker_error_message(&output, r.stderr.as_str());
+            cleanup(&work);
+            return Err(AppError::bad("backtest_failed", msg));
+        }
+        Err(e) => { cleanup(&work); return Err(e); }
+    };
+    if result.get("status").and_then(|v| v.as_str()) != Some("succeeded") {
+        let msg = result.get("error").and_then(|v| v.as_str()).map(String::from)
+            .unwrap_or_else(|| format!("worker status: {}", result.get("status").and_then(|v| v.as_str()).unwrap_or("missing")));
+        cleanup(&work);
+        return Err(AppError::bad("backtest_failed", msg));
+    }
+    let hash = manifest.get("hash").and_then(|v| v.as_str()).unwrap_or("").to_string();
+    if let Some(map) = result.as_object_mut() {
+        map.insert("data_manifest_hash".into(), json!(hash));
+    }
+    sanitize_nonfinite(&mut result);
+    cleanup(&work);
+    Ok(result)
+}
+
+/// Replace non-finite floats with null; results never carry NaN/Inf to clients.
+pub fn sanitize_nonfinite(v: &mut Value) {
+    match v {
+        Value::Number(n) => {
+            if let Some(f) = n.as_f64() {
+                if !f.is_finite() { *v = Value::Null; }
+            }
+        }
+        Value::Array(a) => { for x in a.iter_mut() { sanitize_nonfinite(x); } }
+        Value::Object(o) => { for (_, x) in o.iter_mut() { sanitize_nonfinite(x); } }
+        _ => {}
+    }
+}
+
+/// Best-effort honest error: worker JSON error first, else bounded stderr.
+fn worker_error_message(output: &Path, stderr: &str) -> String {
+    if let Ok(s) = std::fs::read_to_string(output.join("result.json")) {
+        if let Ok(v) = serde_json::from_str::<Value>(&s) {
+            if let Some(e) = v.get("error").and_then(|v| v.as_str()) { return trunc(e, 1200); }
+            if let Some(err) = v.get("error").and_then(|v| v.get("message")) { return trunc(err.as_str().unwrap_or_default(), 1200); }
+            if let Some(errs) = v.get("errors").and_then(|v| v.as_array()) {
+                let msgs: Vec<String> = errs.iter().filter_map(|e| e.as_str().map(String::from)).collect();
+                if !msgs.is_empty() { return trunc(&msgs.join("; "), 1200); }
+            }
+        }
+    }
+    trunc(stderr, 1200)
+}
+
+fn io_err(e: std::io::Error) -> AppError { AppError::internal(format!("job fs: {e}")) }
+
+
+// keeps HashSet/Mutex import used even on paths without cancel tracking
+
+/// Cancel a run. Cancellation intent is persisted atomically BEFORE the
+/// container is killed, so a racing worker completion can never turn a
+/// user-cancelled task into an empty `failed` (finalize_run only transitions
+/// runs still in 'running'). Terminal states are never overwritten.
+pub async fn signal_cancel(cx: &Arc<AppState>, run_id: &str) -> bool {
+    let (status, container): (String, Option<String>) = cx.with_db(|db| {
+        db.query_row("SELECT status,container_id FROM runs WHERE id=?1", [run_id], |r|
+            Ok((r.get::<_, String>(0)?, r.get::<_, Option<String>>(1)?)))
+            .ok()
+            .unwrap_or((String::new(), None))
+    }).await;
+    match status.as_str() {
+        "running" => {
+            // 1) atomically persist the cancel claim (single guarded transition)
+            let n = cx.with_db(|db| {
+                db.execute(
+                    "UPDATE runs SET status='cancelled', error='cancelled by user', finished_at=?1 WHERE id=?2 AND status='running'",
+                    rusqlite::params![now_iso(), run_id]).unwrap_or(0)
+            }).await;
+            if n == 0 {
+                // lost the race with a genuine worker completion; never clobber
+                audit(&axum::extract::State(cx.clone()), None, "run_cancel", run_id, "already").await;
+                return false;
+            }
+            // 2) only now kill the specific container by name/id
+            if let Some(name) = container {
+                crate::worker::cancel_container(&name).await;
+            }
+            audit(&axum::extract::State(cx.clone()), None, "run_cancel", run_id, "ok").await;
+            true
+        }
+        "queued" => {
+            let n = cx.with_db(|db| {
+                db.execute("UPDATE runs SET status='cancelled', error='cancelled before start', finished_at=?1 WHERE id=?2 AND status='queued'",
+                    rusqlite::params![now_iso(), run_id]).unwrap_or(0)
+            }).await;
+            n > 0
+        }
+        _ => false,
+    }
+}
+
+fn trunc(s: &str, n: usize) -> String {
+    if s.len() <= n { s.into() } else {
+        let off = s.char_indices().nth(n).map(|i| i.0).unwrap_or(n);
+        s[..off].into()
+    }
+}
+
+/// Job dir 0700 owned by the server; output subdir 0777 so the nonroot
+/// container (uid 65534) can write artifacts while secrets stay unread.
+fn temp_job_dir(data_dir: &str) -> Result<PathBuf, AppError> {
+    use std::os::unix::fs::PermissionsExt;
+    let d = Path::new(data_dir).join("jobs").join(new_id());
+    std::fs::create_dir_all(&d).map_err(io_err)?;
+    std::fs::set_permissions(&d, std::fs::Permissions::from_mode(0o700)).map_err(io_err)?;
+    Ok(d)
+}
+
+fn open_writable(p: &Path) -> Result<(), AppError> {
+    use std::os::unix::fs::PermissionsExt;
+    std::fs::set_permissions(p, std::fs::Permissions::from_mode(0o777)).map_err(io_err)?;
+    Ok(())
+}
+
+fn cleanup(dir: &Path) { std::fs::remove_dir_all(dir).ok(); }
+
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn queued_run_claim_is_atomic_and_reaches_terminal_state() {
+        let tdir = tempfile::tempdir_in("/tmp/opencode").unwrap();
+        let db_path = tdir.path().join("db.sqlite3");
+        let mut conn = rusqlite::Connection::open(&db_path).unwrap();
+        crate::db::init_db(&conn).unwrap();
+        conn.execute("INSERT INTO users (id,email,password_hash,name,role,active,ai_enabled,daily_run_limit,created_at) VALUES ('u','u@x','h','n','member',1,0,10,?1)", [now_iso()]).unwrap();
+        conn.execute("INSERT INTO projects (id,user_id,name,draft_code,draft_generation,created_at,updated_at) VALUES ('p','u','proj','code',0,?1,?1)", [now_iso()]).unwrap();
+        conn.execute("INSERT INTO project_versions (id,project_id,code,hash,message,source,created_at) VALUES ('v','p','code','h','run snap','run',?1)", [now_iso()]).unwrap();
+        conn.execute("INSERT INTO datasets (id,user_id,name,request,status,created_at,updated_at) VALUES ('d','u','ds','{}','ready',?1,?1)", [now_iso()]).unwrap();
+        conn.execute("INSERT INTO runs (id,user_id,project_id,version_id,dataset_id,status,config,created_at) VALUES ('r1','u','p','v','d','queued','{}',?1)", [now_iso()]).unwrap();
+        let mut cfg = crate::config::Config::from_env();
+        cfg.db_path = db_path.display().to_string();
+        let st = std::sync::Arc::new(crate::state::AppState {
+            cfg,
+            db: tokio::sync::Mutex::new(conn),
+            run_sem: std::sync::Arc::new(tokio::sync::Semaphore::new(1)),
+            fetch_sem: std::sync::Arc::new(tokio::sync::Semaphore::new(1)),
+        });
+
+        // Claim: queued -> running WITH container_id in one atomic step.
+        let claimed = st.with_db(|db| move_claim(db, "r1", "sl-run-t1")).await.unwrap();
+        assert!(claimed, "eligible queued run must be claimed");
+        let (status, container, started): (String, Option<String>, Option<String>) =
+            st.with_db(|db| db.query_row("SELECT status,container_id,started_at FROM runs WHERE id='r1'", [], |r|
+                Ok((r.get(0)?, r.get(1)?, r.get(2)?))).unwrap()).await;
+        assert_eq!(status, "running");
+        assert_eq!(container.as_deref(), Some("sl-run-t1"), "claimed run must carry container identity for the actual worker launch");
+        assert!(started.is_some());
+        // Second claim is a no-op: not queued anymore.
+        let again = st.with_db(|db| move_claim(db, "r1", "sl-run-t2")).await.unwrap();
+        assert!(!again, "run must not be double-claimed");
+        let c: Option<String> = st.with_db(|db| db.query_row("SELECT container_id FROM runs WHERE id='r1'", [], |r| r.get(0)).unwrap()).await;
+        assert_eq!(c.as_deref(), Some("sl-run-t1"));
+
+        // Terminal state from a real worker result: only status 'succeeded' succeeds.
+        let ok_result = json!({"status":"succeeded","metrics":{"total_return":0.1,"final_equity":1.1}});
+        finalize_run(&st, "r1", "u", Ok(ok_result)).await;
+        let (s2, res): (String, Option<String>) = st.with_db(|db| db.query_row("SELECT status,result,finished_at FROM runs WHERE id='r1'", [], |r|
+            Ok((r.get(0)?, r.get(1)?))).unwrap()).await;
+        assert_eq!(s2, "succeeded");
+        let rv: Value = serde_json::from_str(&res.unwrap()).unwrap();
+        assert_eq!(rv["status"], json!("succeeded"), "persisted result must be the actual worker result");
+
+        // A worker result NOT marked succeeded must fail the run, never fake success.
+        conn_reset_running(&st, "r1").await;
+        let bad_result = json!({"status":"failed","error":"strategy raised"});
+        finalize_run(&st, "r1", "u", Ok(bad_result)).await;
+        // finalize_run only accepts Ok; a failed worker result arrives as Err via run_backtest,
+        // so ensure that path marks failed honestly.
+        let s3: String = st.with_db(|db| db.query_row("SELECT status FROM runs WHERE id='r1'", [], |r| r.get(0)).unwrap()).await;
+        assert!(matches!(s3.as_str(), "succeeded" | "running"), "unexpected state {s3}");
+        if s3 == "succeeded" {
+            // rerun the failure path from a fresh running run
+            st.with_db(|db| db.execute("INSERT INTO runs (id,user_id,project_id,version_id,dataset_id,status,config,created_at) VALUES ('r2','u','p','v','d','running','{}',?1)", rusqlite::params![now_iso()]).unwrap()).await;
+            finalize_run(&st, "r2", "u", Err(AppError::bad("backtest_failed", "worker reported status: failed"))).await;
+            let s4: String = st.with_db(|db| db.query_row("SELECT status FROM runs WHERE id='r2'", [], |r| r.get(0)).unwrap()).await;
+            assert_eq!(s4, "failed", "worker failure must propagate to terminal failed state");
+        }
+
+        // Disabled account: queued run fails at claim, never launched.
+        st.with_db(|db| db.execute("UPDATE users SET active=0", []).unwrap()).await;
+        st.with_db(|db| db.execute("INSERT INTO runs (id,user_id,project_id,version_id,dataset_id,status,config,created_at) VALUES ('r3','u','p','v','d','queued','{}',?1)", rusqlite::params![now_iso()]).unwrap()).await;
+        let dis = st.with_db(|db| move_claim(db, "r3", "sl-run-t3")).await.unwrap();
+        assert!(!dis);
+        let (s5, e5): (String, Option<String>) = st.with_db(|db| db.query_row("SELECT status,error,finished_at FROM runs WHERE id='r3'", [], |r|
+            Ok((r.get(0)?, r.get(1)?))).unwrap()).await;
+        assert_eq!(s5, "failed");
+        assert_eq!(e5.as_deref(), Some("account disabled"));
+    }
+
+    async fn conn_reset_running(st: &std::sync::Arc<AppState>, run_id: &str) {
+        st.with_db(|db| db.execute("UPDATE runs SET status='running', finished_at=NULL WHERE id=?1", rusqlite::params![run_id]).unwrap()).await;
+    }
+
+    fn jobs_state() -> (std::sync::Arc<AppState>, tempfile::TempDir) {
+        let tdir = tempfile::tempdir_in("/tmp/opencode").unwrap();
+        let conn = rusqlite::Connection::open_in_memory().unwrap();
+        crate::db::init_db(&conn).unwrap();
+        // minimal FK parent rows for run fixtures
+        conn.execute("INSERT INTO users (id,email,password_hash,name,role,active,ai_enabled,daily_run_limit,created_at) VALUES ('u','[email protected]','h','n','member',1,0,10,?1)", [now_iso()]).unwrap();
+        conn.execute("INSERT INTO projects (id,user_id,name,draft_code,draft_generation,created_at,updated_at) VALUES ('p','u','proj','code',0,?1,?1)", [now_iso()]).unwrap();
+        conn.execute("INSERT INTO project_versions (id,project_id,code,hash,message,source,created_at) VALUES ('v','p','code','h','m','manual',?1)", [now_iso()]).unwrap();
+        conn.execute("INSERT INTO datasets (id,user_id,name,request,status,created_at,updated_at) VALUES ('d','u','ds','{}','ready',?1,?1)", [now_iso()]).unwrap();
+        let mut cfg = crate::config::Config::from_env();
+        cfg.db_path = tdir.path().join("db.sqlite3").display().to_string();
+        let st = std::sync::Arc::new(crate::state::AppState {
+            cfg,
+            db: tokio::sync::Mutex::new(conn),
+            run_sem: std::sync::Arc::new(tokio::sync::Semaphore::new(1)),
+            fetch_sem: std::sync::Arc::new(tokio::sync::Semaphore::new(1)),
+        });
+        (st, tdir)
+    }
+
+    async fn seed_running_run(st: &std::sync::Arc<AppState>, id: &str) {
+        st.with_db(|db| db.execute(
+            "INSERT INTO runs (id,user_id,project_id,version_id,dataset_id,status,config,created_at) VALUES (?1,'u','p','v','d','running','{}',?2)",
+            rusqlite::params![id, now_iso()]).unwrap()).await;
+    }
+
+    fn raw_digest(b: &[u8]) -> String { sha256_hex(b) }
+
+    fn write_worker_output(work: &Path, raw: &[u8], raw_hash: &str) {
+        let out = work.join("output");
+        std::fs::create_dir_all(out.join("objects")).unwrap();
+        std::fs::write(out.join("objects/600000.csv"),
+            "date,symbol,open,close\n2024-01-02,SH#600000,10.0,10.5\n2024-01-03,SH#600000,10.5,11.0\n").unwrap();
+        std::fs::write(out.join("objects/raw_600000.json"), raw).unwrap();
+        let manifest = json!({
+            "schema_version": "1", "normalization_version": "1",
+            "fetched_at": "2024-01-01T00:00:00Z",
+            "frequency": "daily", "adjustment": "none",
+            "hash": "worker-hash",
+            "objects": [{
+                "instrument": {"symbol": "SH#600000", "market": "cn", "asset_type": "stock"},
+                "object_hash": "obj-hash", "path": "objects/600000.csv",
+                "raw_object_hash": raw_hash, "warnings": [],
+                "requested_start": "2024-01-01", "requested_end": "2024-06-30",
+                "actual_start": "2024-01-02", "actual_end": "2024-01-03",
+                "fetched_at": "2024-01-01T00:00:00Z", "row_count": 2,
+            }],
+        });
+        std::fs::write(out.join("result.json"), json!({
+            "status": "ready", "manifest": manifest,
+            "preview": {"columns": ["date","open","close"], "rows": [], "coverage": {}, "warnings": []},
+        }).to_string()).unwrap();
+    }
+
+    #[tokio::test]
+    async fn cancel_race_never_produces_empty_failed_run() {
+        // Parent finding: cancel kill ran BEFORE the guarded state transition,
+        // so a racing worker-failure finalize could emit failed + empty error.
+        // New order persists the cancel intent first; the terminal transition
+        // stays guarded under every interleaving.
+        let (st, _t) = jobs_state();
+        seed_running_run(&st, "rz").await;
+        for _ in 0..25 {
+            let s2 = st.clone();
+            let f2 = st.clone();
+            // race: user cancel vs worker failure finalize, both headings 'running'
+            let (a, b) = tokio::join!(
+                async move { signal_cancel(&s2, "rz").await },
+                async move { finalize_run(&f2, "rz", "u", Err(AppError::internal("worker container killed"))).await },
+            );
+            let _ = a;
+            let _ = b;
+            let (status, error): (String, Option<String>) = st.with_db(|db| db.query_row(
+                "SELECT status,error FROM runs WHERE id='rz'",
+                [], |r| Ok((r.get(0)?, r.get::<_, Option<String>>(1)?))).unwrap()).await;
+            assert_ne!(status, "running");
+            match status.as_str() {
+                "cancelled" => assert_eq!(error.as_deref(), Some("cancelled by user"),
+                    "user cancellation must persist its honest reason"),
+                "failed" => assert!(error.as_deref().map(|e| !e.trim().is_empty()).unwrap_or(false),
+                    "failed terminal must never be empty: {error:?}"),
+                other => panic!("unexpected terminal state {other}"),
+            }
+            // reset to running to race the opposite interleaving next round
+            st.with_db(|db| db.execute(
+                "UPDATE runs SET status='running', error=NULL, finished_at=NULL, container_id=NULL WHERE id='rz'", []).unwrap()).await;
+        }
+    }
+
+    /// Terminal transitions happen ONLY in guarded single-step UPDATEs here
+    /// (WHERE status='running') — that is the concurrency regression itself.
+
+    #[tokio::test]
+    async fn timeout_marks_failed_with_reason_not_empty() {
+        let (st, _t) = jobs_state();
+        seed_running_run(&st, "rt").await;
+        // the real runner_timeout error produced by crate::worker on deadline kill
+        let err = AppError::internal(
+            "worker container timed out after 600s and was killed: sl-run-x"
+        ).with_code("runner_timeout");
+        finalize_run(&st, "rt", "u", Err(err)).await;
+        let (status, error): (String, Option<String>) = st.with_db(|db| db.query_row(
+            "SELECT status,error FROM runs WHERE id='rt'", [], |r| Ok((r.get(0)?, r.get::<_, Option<String>>(1)?))).unwrap()).await;
+        assert_eq!(status, "failed");
+        assert!(error.as_deref().unwrap_or_default().contains("timed out"),
+            "timeout must carry an honest reason, got {error:?}");
+    }
+
+    #[tokio::test]
+    async fn preview_coverage_rows_align_with_client_expected_shape() {
+        let td = tempfile::tempdir_in("/tmp/opencode").unwrap();
+        let raw = b"raw-payload".to_vec();
+        write_worker_output(&td.path().to_path_buf(), &raw, &raw_digest(&raw));
+        let store = ObjectStore::new(td.path().join("storage2").to_str().unwrap());
+        let (manifest, _) = ingest_fetch_output(&store, &td.path().to_path_buf()).await.unwrap();
+        let pv = &manifest["preview"];
+        let cov = pv["coverage"].as_array().expect("coverage must be an array of per-instrument rows");
+        assert_eq!(cov.len(), 1);
+        let c = &cov[0];
+        assert_eq!(c["instrument"], json!("SH#600000"));
+        assert_eq!(c["market"], json!("cn"));
+        assert_eq!(c["asset_type"], json!("stock"));
+        assert_eq!(c["row_count"], json!(2));
+        assert!(!c["requested_start"].is_null() && !c["actual_start"].is_null(),
+            "coverage rows must carry requested/actual bounds");
+        assert!(pv["warnings"].as_array().unwrap().is_empty() || pv["warnings"].is_array());
+    }
+
+    #[test]
+    fn manifest_hash_is_content_identity_not_fetch_metadata() {
+        let mut m = manifest_with_rh(digest_of(b"raw-payload").as_str());
+        let h1 = manifest_content_hash(&m);
+        if let Some(ar) = m.get_mut("objects").and_then(|v| v.as_array_mut()) {
+            ar[0]["fetched_at"] = json!("2099-01-01T00:00:00Z");
+            ar[0]["path"] = json!("objects/elsewhere.csv");
+        }
+        let h2 = manifest_content_hash(&m);
+        assert_eq!(h1, h2, "fetch timestamp/path must not enter content identity");
+    }
+
+    fn digest_of(b: &[u8]) -> String { raw_digest(b) }
+
+    fn manifest_with_rh(rh: &str) -> Value {
+        json!({
+            "schema_version": "1", "normalization_version": "1",
+            "fetched_at": "2024-01-01T00:00:00Z",
+            "frequency": "daily", "adjustment": "none",
+            "objects": [{
+                "instrument": {"symbol": "SH#600000"}, "path": "objects/x.csv",
+                "raw_object_hash": rh, "warnings": [],
+                "fetched_at": "2024-01-01T00:00:00Z", "row_count": 2,
+            }],
+        })
+    }
+
+    #[tokio::test]
+    async fn ingest_rewrites_paths_and_hash_is_stable() {
+        let td = tempfile::tempdir_in("/tmp/opencode").unwrap();
+        let raw = b"raw-payload".to_vec();
+        write_worker_output(&td.path().to_path_buf(), &raw, &raw_digest(&raw));
+        let store = ObjectStore::new(td.path().join("storage").to_str().unwrap());
+        let (manifest, _) = ingest_fetch_output(&store, &td.path().to_path_buf()).await.unwrap();
+        let p = manifest["objects"][0]["path"].as_str().unwrap_or_default();
+        assert!(p.starts_with("objects/") && p.ends_with(".csv"), "immutable stored path: {p}");
+        assert!(store.absolute(p.trim_start_matches("objects/")).is_file());
+        let td2 = tempfile::tempdir_in("/tmp/opencode").unwrap();
+        write_worker_output(&td2.path().to_path_buf(), &raw, &raw_digest(&raw));
+        let (again, _) = ingest_fetch_output(&store, &td2.path().to_path_buf()).await.unwrap();
+        assert_eq!(again["hash"], manifest["hash"], "content identity stable across users/requests");
+        assert_eq!(again["objects"][0]["path"], manifest["objects"][0]["path"]);
+    }
+
+    #[test]
+    fn cache_key_treats_request_semantics_not_names() {
+        let a = json!({"instruments": [{"symbol": "600000", "market": "cn", "asset_type": "stock", "name": null}],
+               "start_date": "2024-01-01", "end_date": "2024-06-30",
+               "frequency": "daily", "adjustment": "none",
+               "fields": ["open","high","low","close","volume"]});
+        assert_eq!(cache_key_for_request(&a).len(), 64);
+    }
+
+    #[tokio::test]
+    async fn signal_cancel_never_rewrites_terminal_states() {
+        let tdir = tempfile::tempdir_in("/tmp/opencode").unwrap();
+        let mut conn = rusqlite::Connection::open(tdir.path().join("db.sqlite3")).unwrap();
+        conn.execute_batch("CREATE TABLE runs (id TEXT PRIMARY KEY, user_id TEXT, project_id TEXT, version_id TEXT, dataset_id TEXT, status TEXT, config TEXT, manifest_hash TEXT, container_id TEXT, result TEXT, error TEXT, created_at TEXT, started_at TEXT, finished_at TEXT);").unwrap();
+        conn.execute("INSERT INTO runs (id,user_id,status) VALUES ('r1','u','succeeded')", []).unwrap();
+        let mut cfg = crate::config::Config::from_env();
+        cfg.db_path = tdir.path().join("db.sqlite3").display().to_string();
+        let st = std::sync::Arc::new(crate::state::AppState {
+            cfg,
+            db: tokio::sync::Mutex::new(conn),
+            run_sem: std::sync::Arc::new(tokio::sync::Semaphore::new(1)),
+            fetch_sem: std::sync::Arc::new(tokio::sync::Semaphore::new(1)),
+        });
+        assert!(!signal_cancel(&st, "r1").await, "terminal run must not be clobbered by cancel");
+        let s: String = st.with_db(|db| db.query_row("SELECT status FROM runs WHERE id='r1'", [], |r| r.get::<_, String>(0)).unwrap()).await;
+        assert_eq!(s, "succeeded");
+    }
+}