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

15 files changed, 4924 insertions, 0 deletions

diff --git a/server/src/admin.rs b/server/src/admin.rs
new file mode 100644
index 0000000..d6a861e
--- /dev/null
+++ b/server/src/admin.rs
@@ -0,0 +1,387 @@
+use serde_json::json;
+use axum::Json;
+
+use crate::auth::{audit, hash_password, make_admin_token, AuthUser};
+use crate::error::{AppError, AppResult};
+use crate::state::Cx;
+use crate::util::now_iso;
+
+async fn assert_admin(_cx: &Cx, auth: &AuthUser) -> AppResult<()> {
+    if auth.role != "admin" {
+        return Err(AppError::forbidden("admin only"));
+    }
+    Ok(())
+}
+
+fn user_json(id: &str, email: &str, name: &str, role: &str, active: i64, ai: i64, limit: i64, created: String) -> serde_json::Value {
+    json!({
+        "id": id, "email": email, "name": name, "role": role,
+        "active": active != 0, "ai_enabled": ai != 0,
+        "daily_run_limit": limit, "created_at": created,
+    })
+}
+
+pub async fn users(cx: Cx, auth: AuthUser) -> AppResult<Json<serde_json::Value>> {
+    assert_admin(&cx, &auth).await?;
+    let items: Vec<serde_json::Value> = cx.with_db(|db| {
+        let mut st = db.prepare("SELECT id,email,name,role,active,ai_enabled,daily_run_limit,created_at FROM users ORDER BY created_at ASC")?;
+        let v: Vec<serde_json::Value> = st.query_map([], |r| Ok(user_json(
+            &r.get::<_, String>(0)?, &r.get::<_, String>(1)?, &r.get::<_, String>(2)?, &r.get::<_, String>(3)?,
+            r.get::<_, i64>(4)?, r.get::<_, i64>(5)?, r.get::<_, i64>(6)?, r.get::<_, String>(7)?,
+        )))?.collect::<Result<_, _>>()?;
+        Ok::<_, AppError>(v)
+    }).await?;
+    Ok(Json(json!({ "items": items })))
+}
+
+#[derive(serde::Deserialize)]
+pub struct AdminUserPatch {
+    pub active: Option<bool>,
+    pub role: Option<String>,
+    pub daily_run_limit: Option<i64>,
+    pub ai_enabled: Option<bool>,
+}
+
+pub async fn patch_user(cx: Cx, auth: AuthUser, axum::extract::Path(target): axum::extract::Path<String>, body: Option<axum::Json<AdminUserPatch>>) -> AppResult<Json<serde_json::Value>> {
+    assert_admin(&cx, &auth).await?;
+    let axum::Json(r) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?;
+    if let Some(role) = &r.role {
+        if !matches!(role.as_str(), "admin" | "member") {
+            return Err(AppError::bad("validation", "role must be admin|member"));
+        }
+    }
+    if let Some(n) = r.daily_run_limit {
+        if !(1..=2000).contains(&n) {
+            return Err(AppError::bad("validation", "daily_run_limit must be between 1 and 2000"));
+        }
+    }
+    cx.with_db(|db| -> AppResult<()> {
+        let tx = db.transaction()?;
+        let (trole, tactive): (String, i64) = tx.query_row(
+            "SELECT role, active FROM users WHERE id=?1", [&target], |row| Ok((row.get(0)?, row.get(1)?)))
+            .map_err(|_| AppError::not_found("user not found"))?;
+        let demoting = r.role.as_ref().map(|new| trole == "admin" && new != "admin").unwrap_or(false);
+        let disabling = r.active == Some(false);
+        // Last active admin protection, enforced atomically with the update.
+        if (demoting || disabling) && trole == "admin" && tactive != 0 {
+            let active_admins: i64 = tx.query_row("SELECT COUNT(*) FROM users WHERE role='admin' AND active=1", [], |row| row.get(0))?;
+            if active_admins <= 1 {
+                return Err(AppError::conflict("last_admin", "cannot demote or disable the last active admin"));
+            }
+        }
+        tx.execute(
+            "UPDATE users SET ai_enabled=COALESCE(?1,ai_enabled), daily_run_limit=COALESCE(?2,daily_run_limit), role=COALESCE(?3,role), active=COALESCE(?4,active) WHERE id=?5",
+            rusqlite::params![
+                r.ai_enabled.map(|b| b as i64),
+                r.daily_run_limit,
+                &r.role,
+                r.active.map(|b| b as i64),
+                &target,
+            ])?;
+        // Disabling revokes every session and freezes that user's live runs,
+        // inside the same transaction as the account state flip.
+        if disabling {
+            tx.execute("DELETE FROM sessions WHERE user_id=?1", [&target]).ok();
+            tx.execute(
+                "UPDATE runs SET status='cancelled', error='account disabled', finished_at=?1 WHERE user_id=?2 AND status IN ('queued','running')",
+                rusqlite::params![now_iso(), &target],
+            ).ok();
+        }
+        tx.commit()?;
+        Ok(())
+    }).await?;
+    audit(&cx, Some(&auth.id), "admin_user_update", &target, "ok").await;
+    let v = cx.with_db(|db| {
+        db.query_row("SELECT id,email,name,role,active,ai_enabled,daily_run_limit,created_at FROM users WHERE id=?1", [&target], |r|
+            Ok(user_json(
+                &r.get::<_, String>(0)?, &r.get::<_, String>(1)?, &r.get::<_, String>(2)?, &r.get::<_, String>(3)?,
+                r.get::<_, i64>(4)?, r.get::<_, i64>(5)?, r.get::<_, i64>(6)?, r.get::<_, String>(7)?,
+            ))).map_err(|_| AppError::not_found("user not found"))
+    }).await?;
+    Ok(Json(v))
+}
+
+#[derive(serde::Deserialize)]
+pub struct InvitationReq {
+    pub email: Option<String>,
+    pub role: Option<String>,
+    pub expires_hours: Option<i64>,
+}
+
+pub async fn create_invitation(cx: Cx, auth: AuthUser, body: Option<axum::Json<InvitationReq>>) -> AppResult<(axum::http::StatusCode, Json<serde_json::Value>)> {
+    assert_admin(&cx, &auth).await?;
+    let axum::Json(r) = body.unwrap_or(axum::Json(InvitationReq { email: None, role: None, expires_hours: None }));
+    let role = r.role.unwrap_or_else(|| "member".to_string());
+    // POC: invitations can only mint members; admins are bootstrapped offline.
+    // The invitation's role is stored in DB and registration ignores any
+    // client-supplied role input, so no escalation path exists.
+    if role != "member" {
+        return Err(AppError::bad("validation", "POC invitations can only create member role"));
+    }
+    let hours = r.expires_hours.unwrap_or(168);
+    if !(1..=336).contains(&hours) {
+        return Err(AppError::bad("validation", "expires_hours must be 1-336"));
+    }
+    let (token, expires) = make_admin_token(&cx, "invitations", None, hours, r.email.as_deref()).await?;
+    audit(&cx, Some(&auth.id), "invitation_issued", r.email.as_deref().unwrap_or("open-invite"), "ok").await;
+    Ok((axum::http::StatusCode::CREATED, Json(json!({ "token": token, "expires_at": expires }))))
+}
+
+/// Sanitized invitations list: no token hashes, no secrets.
+pub async fn list_invitations(cx: Cx, auth: AuthUser) -> AppResult<Json<serde_json::Value>> {
+    assert_admin(&cx, &auth).await?;
+    let items: Vec<serde_json::Value> = cx.with_db(|db| {
+        let now = now_iso();
+        let mut st = db.prepare("SELECT id,email,role,expires_at,used_by,created_at FROM invitations WHERE expires_at > ?1 ORDER BY created_at DESC")?;
+        let v = st.query_map([&now], |r| Ok(json!({
+            "id": r.get::<_, String>(0)?,
+            "email": r.get::<_, Option<String>>(1)?,
+            "role": r.get::<_, String>(2)?,
+            "expires_at": r.get::<_, String>(3)?,
+            "used_by": r.get::<_, Option<String>>(4)?,
+            "created_at": r.get::<_, String>(5)?,
+        })))?.collect::<Result<Vec<_>, _>>()?;
+        Ok::<_, AppError>(v)
+    }).await?;
+    Ok(Json(json!({ "items": items })))
+}
+
+pub async fn delete_invitation(cx: Cx, auth: AuthUser, axum::extract::Path(id): axum::extract::Path<String>) -> AppResult<Json<serde_json::Value>> {
+    assert_admin(&cx, &auth).await?;
+    cx.with_db(|db| -> AppResult<()> {
+        if db.execute("DELETE FROM invitations WHERE id=?1", [&id])? == 0 {
+            return Err(AppError::not_found("invitation not found"));
+        }
+        Ok(())
+    }).await?;
+    audit(&cx, Some(&auth.id), "invitation_revoke", &id, "ok").await;
+    Ok(Json(json!({"ok": true})))
+}
+
+/// Admin-issued password reset token, hashed in DB, short-lived single-use.
+/// The raw token is returned once for display; no fake email delivery.
+pub async fn create_reset(cx: Cx, auth: AuthUser, axum::extract::Path(target): axum::extract::Path<String>, _body: Option<axum::Json<serde_json::Value>>) -> AppResult<(axum::http::StatusCode, Json<serde_json::Value>)> {
+    assert_admin(&cx, &auth).await?;
+    cx.with_db(|db| {
+        db.query_row("SELECT 1 FROM users WHERE id=?1", [&target], |row| row.get::<_, i64>(0))
+            .map_err(|_| AppError::not_found("user not found"))?;
+        Ok::<_, AppError>(())
+    }).await?;
+    let hours = 2;
+    let (token, expires) = make_admin_token(&cx, "password_resets", Some(target.as_str()), hours, None).await?;
+    audit(&cx, Some(&auth.id), "admin_password_reset_issued", &target, "ok").await;
+    Ok((axum::http::StatusCode::CREATED, Json(json!({"reset_token": token, "expires_at": expires}))))
+}
+
+/// Sanitized security audit listing: actor/action/target/time/status only.
+/// Sensitive material never reaches this table (see auth::audit) and stored
+/// targets are rendered trimmed, never with password/key/code content.
+pub async fn audit_list(cx: Cx, auth: AuthUser) -> AppResult<Json<serde_json::Value>> {
+    assert_admin(&cx, &auth).await?;
+    let items: Vec<serde_json::Value> = cx.with_db(|db| {
+        let mut st = db.prepare("SELECT ts,actor_id,action,target,status FROM audit ORDER BY seq DESC LIMIT 500")?;
+        let v = st.query_map([], |r| {
+            let ts: String = r.get(0)?;
+            let actor: Option<String> = r.get(1)?;
+            let action: String = r.get(2)?;
+            let target: Option<String> = r.get(3)?;
+            let status: String = r.get(4)?;
+            let items_json = json!({
+                "ts": ts, "actor": actor, "action": action,
+                "target": target.unwrap_or_default(), "status": status,
+            });
+            Ok(items_json)
+        })?.collect::<Result<Vec<_>, _>>()?;
+        Ok::<_, AppError>(v)
+    }).await?;
+    Ok(Json(json!({ "items": items })))
+}
+
+/// Bootstrap the first admin from env. Idempotent: only inserts when no
+/// admin account exists yet.
+pub async fn bootstrap_admin(cx: &Cx) -> AppResult<()> {
+    let cfg = &cx.cfg;
+    let Some(email) = cfg.bootstrap_admin_email.clone() else { return Ok(()); };
+    let Some(password) = cfg.bootstrap_admin_password.clone() else { return Ok(()); };
+    let email = email.trim().to_lowercase();
+    if email.is_empty() || !email.contains('@') || password.len() < 8 {
+        return Ok(());
+    }
+    let exists: i64 = cx.with_db(|db| {
+        db.query_row("SELECT COUNT(*) FROM users WHERE role='admin'", [], |r| r.get(0)).unwrap_or(0)
+    }).await;
+    if exists > 0 { return Ok(()); }
+    let hash = hash_password(&password)?;
+    let id = crate::util::new_id();
+    cx.with_db(|db| {
+        db.execute("INSERT OR IGNORE INTO users (id,email,password_hash,name,role,active,ai_enabled,daily_run_limit,created_at) VALUES (?1,?2,?3,'Administrator','admin',1,1,100,?4)",
+            rusqlite::params![&id, &email, &hash, now_iso()]).ok();
+    }).await;
+    audit(cx, Some(&id), "bootstrap_admin", &id, "ok").await;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::state::AppState;
+    use axum::extract::{Path as P, State};
+    use std::sync::Arc;
+
+    fn handle(s: &Arc<AppState>) -> Cx { State(s.clone()) }
+
+    fn admin_auth() -> AuthUser {
+        AuthUser { id: "admin-x".into(), email: "[email protected]".into(), role: "admin".into(), ai_enabled: true, daily_run_limit: 100, session_id: "sess-admin".into() }
+    }
+
+    fn member_auth() -> AuthUser {
+        AuthUser { id: "member-x".into(), email: "[email protected]".into(), role: "member".into(), ai_enabled: false, daily_run_limit: 10, session_id: "sess-member".into() }
+    }
+
+    fn test_state() -> Arc<AppState> {
+        let conn = rusqlite::Connection::open_in_memory().expect("in-memory db");
+        crate::db::init_db(&conn).expect("schema");
+        let cfg = crate::config::Config {
+            bind_addr: "127.0.0.1:0".into(),
+            canonical_origin: String::new(),
+            secure_cookies: false,
+            db_path: ":memory:".into(),
+            frontend_dir: "frontend/dist".into(),
+            data_dir: std::env::temp_dir().to_string_lossy().into_owned(),
+            worker_image: "test".into(),
+            fetch_timeout_secs: 1,
+            backtest_timeout_secs: 1,
+            run_concurrency: 1,
+            fetch_concurrency: 1,
+            session_hours: 24,
+            bootstrap_admin_email: None,
+            bootstrap_admin_password: None,
+            ai_base_url: "http://127.0.0.1:9".into(),
+            ai_model: "test-model".into(),
+            ai_daily_request_cap: 1,
+            ai_input_token_cap: 1,
+            ai_output_token_cap: 1,
+            ai_enabled_poc: false,
+            default_run_limit_per_day: 10,
+            version: "test".into(),
+        };
+        Arc::new(AppState {
+            cfg,
+            db: tokio::sync::Mutex::new(conn),
+            run_sem: Arc::new(tokio::sync::Semaphore::new(1)),
+            fetch_sem: Arc::new(tokio::sync::Semaphore::new(1)),
+        })
+    }
+
+    /// Insert a user directly and return its id.
+    async fn seed_user(cx: &Cx, email: &str, role: &str, password: &str) -> String {
+        let hash = crate::auth::hash_password(password).unwrap();
+        let uid = crate::util::new_id();
+        cx.with_db(|db| {
+            db.execute(
+                "INSERT INTO users (id,email,password_hash,name,role,active,ai_enabled,daily_run_limit,created_at) VALUES (?1,?2,?3,?4,?5,1,0,10,?6)",
+                rusqlite::params![&uid, email, &hash, "Test User", role, now_iso()]).unwrap();
+        }).await;
+        uid
+    }
+
+    #[tokio::test]
+    async fn member_cannot_list_users_or_audit() {
+        let s = test_state();
+        let u = users(handle(&s), member_auth()).await;
+        assert_eq!(u.err().map(|e| e.status).unwrap_or_default(), axum::http::StatusCode::FORBIDDEN);
+        let a = audit_list(handle(&s), member_auth()).await;
+        assert_eq!(a.err().map(|e| e.status).unwrap_or_default(), axum::http::StatusCode::FORBIDDEN);
+        let inv = create_invitation(handle(&s), member_auth(), None).await;
+        assert_eq!(inv.err().map(|e| e.status).unwrap_or_default(), axum::http::StatusCode::FORBIDDEN);
+    }
+
+    #[tokio::test]
+    async fn last_active_admin_cannot_be_disabled_or_demoted() {
+        let s = test_state();
+        let aid = seed_user(&handle(&s), "[email protected]", "admin", "an-admin-passphrase").await;
+        let res = patch_user(handle(&s), admin_auth(), P(aid.clone()), Some(axum::Json(AdminUserPatch { active: Some(false), role: None, daily_run_limit: None, ai_enabled: None }))).await;
+        assert_eq!(res.err().map(|e| e.code).unwrap_or_default(), "last_admin", "disabling the last admin must be blocked");
+        let res = patch_user(handle(&s), admin_auth(), P(aid.clone()), Some(axum::Json(AdminUserPatch { active: None, role: Some("member".into()), daily_run_limit: None, ai_enabled: None }))).await;
+        assert_eq!(res.err().map(|e| e.code).unwrap_or_default(), "last_admin", "demoting the last admin must be blocked");
+        // With a second active admin, disabling becomes legal.
+        let _ = seed_user(&handle(&s), "[email protected]", "admin", "another-passphrase-2").await;
+        let ok = patch_user(handle(&s), admin_auth(), P(aid.clone()), Some(axum::Json(AdminUserPatch { active: Some(false), role: None, daily_run_limit: None, ai_enabled: None }))).await;
+        assert!(ok.is_ok());
+        let out = ok.unwrap().0;
+        assert_eq!(out["active"], json!(false));
+    }
+
+    #[tokio::test]
+    async fn disabling_user_revokes_sessions_and_cancels_runs() {
+        let s = test_state();
+        let aid = seed_user(&handle(&s), "[email protected]", "member", "a-member-passphrase").await;
+        handle(&s).with_db(|db| {
+            db.execute("INSERT INTO sessions (id,user_id,created_at,expires_at) VALUES ('sess-1',?1,?2,?3)",
+                rusqlite::params![&aid, now_iso(), crate::util::plus_hours(1)]).unwrap();
+            db.execute(
+                "INSERT INTO projects (id,user_id,name,draft_code,draft_generation,created_at,updated_at) VALUES ('pid',?1,'p','',0,?2,?2)",
+                rusqlite::params![&aid, now_iso()]).unwrap();
+            db.execute(
+                "INSERT INTO project_versions (id,project_id,code,hash,message,source,created_at) VALUES ('vid','pid','code','h','m','manual',?1)",
+                [now_iso()]).unwrap();
+            db.execute(
+                "INSERT INTO datasets (id,user_id,name,request,status,created_at,updated_at) VALUES ('did',?1,'ds','{}','ready',?2,?2)",
+                rusqlite::params![&aid, now_iso()]).unwrap();
+            db.execute(
+                "INSERT INTO runs (id,user_id,project_id,version_id,dataset_id,status,config,created_at) VALUES ('rid',?1,'pid','vid','did','running','{}',?2)",
+                rusqlite::params![&aid, now_iso()]).unwrap();
+        }).await;
+        let ok = patch_user(handle(&s), admin_auth(), P(aid.clone()), Some(axum::Json(AdminUserPatch { active: Some(false), role: None, daily_run_limit: None, ai_enabled: None }))).await;
+        assert!(ok.is_ok(), "member can be disabled");
+        let blocked: i64 = handle(&s).with_db(|db| db.query_row(
+            "SELECT COUNT(*) FROM runs WHERE id='rid' AND status='cancelled'", [], |r| r.get(0)).unwrap()).await;
+        assert_eq!(blocked, 1, "live runs must be cancelled");
+        let gone: i64 = handle(&s).with_db(|db| db.query_row(
+            "SELECT COUNT(*) FROM sessions WHERE user_id=?1", [&aid], |r| r.get(0)).unwrap()).await;
+        assert_eq!(gone, 0, "sessions must be revoked");
+    }
+
+    #[tokio::test]
+    async fn invitations_only_mint_members_and_are_hashed() {
+        let s = test_state();
+        let res = create_invitation(handle(&s), admin_auth(), None).await.unwrap();
+        assert_eq!(res.0, axum::http::StatusCode::CREATED);
+        let token = res.1.0["token"].as_str().unwrap().to_string();
+        assert!(!token.contains("password"));
+        // The DB must hold only the sha256 of the token, never the raw token.
+        let hash = crate::util::sha256_hex(token.as_bytes());
+        let hashed_rows: i64 = handle(&s).with_db(|db| db.query_row(
+            "SELECT COUNT(*) FROM invitations WHERE token_hash=?1", [&hash], |r| r.get(0)).unwrap()).await;
+        assert_eq!(hashed_rows, 1);
+        let raw_rows: i64 = handle(&s).with_db(|db| db.query_row(
+            "SELECT COUNT(*) FROM invitations WHERE token_hash=?1", [&token], |r| r.get(0)).unwrap()).await;
+        assert_eq!(raw_rows, 0, "raw tokens must never be stored");
+        // Admin roles via invitation are rejected.
+        let offered = create_invitation(handle(&s), admin_auth(), Some(axum::Json(InvitationReq { email: None, role: Some("admin".into()), expires_hours: None }))).await;
+        assert_eq!(offered.err().map(|e| e.code).unwrap_or_default(), "validation");
+        // Sanitized listing contains no token material.
+        let list = list_invitations(handle(&s), admin_auth()).await.unwrap().0;
+        let raw = serde_json::to_string(&list).unwrap();
+        assert!(!raw.contains(&hash), "listing must not leak token hashes");
+    }
+
+    #[tokio::test]
+    async fn bootstrap_admin_is_idempotent_and_hashes_password() {
+        let mut cfg_state = test_state();
+        {
+            let cfg = &mut Arc::get_mut(&mut cfg_state).unwrap().cfg;
+            cfg.bootstrap_admin_email = Some("[email protected] ".into());
+            cfg.bootstrap_admin_password = Some("boot-admin-passphrase".into());
+        }
+        let cx = handle(&cfg_state);
+        bootstrap_admin(&cx).await.unwrap();
+        bootstrap_admin(&cx).await.unwrap();
+        let count: i64 = cx.with_db(|db| db.query_row(
+            "SELECT COUNT(*) FROM users WHERE role='admin'", [], |r| r.get(0)).unwrap()).await;
+        assert_eq!(count, 1, "bootstrap must not duplicate admins");
+        let hash: String = cx.with_db(|db| db.query_row(
+            "SELECT password_hash FROM users WHERE role='admin'", [], |r| r.get(0)).unwrap()).await;
+        assert!(hash.starts_with("$argon2"), "bootstrap password must be Argon2 hashed");
+    }
+}
diff --git a/server/src/ai.rs b/server/src/ai.rs
new file mode 100644
index 0000000..271fa76
--- /dev/null
+++ b/server/src/ai.rs
@@ -0,0 +1,319 @@
+use axum::extract::Path;
+use axum::http::StatusCode;
+use axum::Json;
+use serde_json::{json, Value};
+
+use crate::auth::{audit, AuthUser};
+use crate::error::{AppError, AppResult};
+use crate::util::new_id;
+use crate::util::now_iso;
+use crate::util::sha256_hex;
+
+use crate::state::Cx;
+
+const ASSIST_TIMEOUT_SECS: u64 = 90;
+pub const MAX_INSTRUCTION_CHARS: usize = 2000;
+
+pub async fn list_ai_usage(cx: Cx, auth: AuthUser) -> AppResult<Json<Value>> {
+    let (items, totals_in, totals_out, n) = cx.with_db(|db| -> AppResult<_> {
+        let mut items = Vec::new();
+        {
+            let mut st = db.prepare("SELECT ts,kind,input_tokens,output_tokens FROM ai_usage WHERE user_id=?1 ORDER BY ts DESC LIMIT 500")?;
+            let mut rows = st.query([auth.id.clone()])?;
+            while let Some(r) = rows.next()? {                items.push(json!({
+                    "ts": r.get::<_, String>(0)?,
+                    "kind": r.get::<_, String>(1)?,
+                    "input_tokens": r.get::<_, i64>(2)?,
+                    "output_tokens": r.get::<_, i64>(3)?,
+                }));
+            }
+        }
+        let (i, o, n): (i64, i64, i64) = db.query_row(
+            "SELECT COALESCE(SUM(input_tokens),0), COALESCE(SUM(output_tokens),0), COUNT(*) FROM ai_usage WHERE user_id=?1",
+            [&auth.id], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))?;
+        Ok((items, i, o, n))
+    }).await?;
+    Ok(Json(json!({
+        "items": items,
+        "totals": {"requests": n, "input_tokens": totals_in, "output_tokens": totals_out, "internal_poc": true},
+    })))
+}
+
+pub async fn assist(cx: Cx, auth: AuthUser, body: Option<Json<serde_json::Value>>) -> AppResult<Json<Value>> {
+    let axum::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 instruction = j.get("instruction").and_then(|v| v.as_str()).ok_or_else(|| AppError::bad("validation", "instruction required"))?.to_string();
+    let expected = j.get("expected_generation").and_then(|v| v.as_i64()).ok_or_else(|| AppError::bad("validation", "expected_generation required"))?;
+    if instruction.trim().is_empty() || instruction.len() > MAX_INSTRUCTION_CHARS {
+        return Err(AppError::bad("validation", "instruction required (max 2000 chars)"));
+    }
+    if !cx.cfg.ai_enabled_poc {
+        return Err(AppError::forbidden("AI assist is not enabled (internal POC flag off)"));
+    }
+    if !auth.ai_enabled {
+        return Err(AppError::forbidden("AI not enabled on this account; contact the administrator"));
+    }
+    let (draft, gen): (String, i64) = cx.with_db(|db| {
+        db.query_row("SELECT draft_code,draft_generation FROM projects WHERE id=?1 AND user_id=?2",
+            rusqlite::params![&project_id, &auth.id], |r| Ok((r.get(0)?, r.get(1)?)))
+            .map_err(|_| AppError::not_found("project not found"))
+    }).await?;
+    if gen != expected {
+        return Err(AppError::conflict("stale_generation", "draft changed; expected generation mismatch"));
+    }
+    // daily request budget, measured from the real ledger
+    let key = std::env::var("OPENCODE_GO_API_KEY").map_err(|_| {
+        AppError::new(StatusCode::INTERNAL_SERVER_ERROR, "ai_not_configured", "[internal] AI provider key not configured in environment")
+    })?;
+    let (used, requests): (i64, i64) = cx.with_db(|db| {
+        let used: i64 = db.query_row("SELECT COUNT(*) FROM ai_requests WHERE user_id=?1 AND created_at LIKE ?2",
+            rusqlite::params![&auth.id, format!("{}%", chrono::Utc::now().format("%Y-%m-%d").to_string())], |r| r.get(0)).unwrap_or(0);
+        let reqs = db.query_row("SELECT COUNT(*) FROM ai_requests WHERE user_id=?1", [&auth.id], |r| r.get(0)).unwrap_or(0);
+        Ok::<_, AppError>((used, reqs))
+    }).await?;
+    let _ = requests;
+    if used >= cx.cfg.ai_daily_request_cap {
+        return Err(AppError::new(StatusCode::TOO_MANY_REQUESTS, "ai_budget", "daily AI request budget reached"));
+    }
+
+    let dataset_summary = summarize_dataset(&cx, &auth.id, &project_id).await?;
+    let system_prompt = "You are a strategy editor for a Backtrader POC. Propose changes to the user strategy code. Reply with one fenced ```python block containing the FULL proposed module, plus a short explanation outside the block. Never fabricate data. Do not execute anything.";
+    // Scope caveat (kept explicit): this integration is an internal POC providing
+    // strategy coding help for own use only; no production use and no commercial
+    // licensing grant is claimed for the upstream model service.
+    let user_prompt = format!("Change requested: {instruction}\n\nAvailable dataset fields (indicator warmup is your code's responsibility): {dataset_summary}\n\nCurrent full strategy source:\n{draft}");
+    let body = json!({
+        "model": cx.cfg.ai_model,
+        "messages": [
+            {"role": "system", "content": system_prompt},
+            {"role": "user", "content": user_prompt},
+        ],
+        "max_tokens": cx.cfg.ai_output_token_cap,
+        "temperature": 0.4,
+    });
+    let id = new_id();
+    let ts = now_iso();
+    cx.with_db(|db| -> AppResult<()> {
+        db.execute("INSERT INTO ai_requests (id,user_id,project_id,instruction,status,base_generation,base_code_hash,created_at) VALUES (?1,?2,?3,?4,'pending',?5,?6,?7)",
+            rusqlite::params![&id, &auth.id, &project_id, &instruction, gen, sha256_hex(draft.as_bytes()), &ts])?;
+        Ok(())
+    }).await?;
+
+    let url = format!("{}/chat/completions", cx.cfg.ai_base_url.trim_end_matches('/'));
+    let client = reqwest::Client::builder().timeout(std::time::Duration::from_secs(ASSIST_TIMEOUT_SECS)).build()
+        .map_err(|_| AppError::internal("http client unavailable"))?;
+    let session_id = ai_session_id(&auth.id, &project_id);
+    let resp = client.post(url)
+        .bearer_auth(&key)
+        .header("Content-Type", "application/json")
+        // Honest app identity for the opencode go gateway (custom coding agents
+        // are an explicitly supported use; no other client identity is claimed).
+        .header("User-Agent", "strategy-lab-coding-assistant/0.1 (internal POC, strategy coding help only)")
+        // Stable per (account, project) conversation id so the upstream can
+        // reuse one session context; never random per request, never the key.
+        .header("x-opencode-session", session_id)
+        .json(&body).send().await;
+
+    let resp: Value = match resp {
+        Ok(r) if r.status().is_success() => r.json().await.unwrap_or(Value::Null),
+        Ok(r) => {
+            let status = r.status();
+            let text = r.text().await.unwrap_or_default();
+            audit(&cx, Some(&auth.id), "ai_error", &id, "fail").await;
+            return Err(AppError::new(StatusCode::BAD_GATEWAY, "ai_upstream_error", format!("model call failed ({status})"))
+                .with_details(json!({"status": status.to_string(), "internal": text.chars().take(500).collect::<String>()})));
+        }
+        Err(e) => {
+            audit(&cx, Some(&auth.id), "ai_error", &id, "error").await;
+            // startup must survive an unreachable provider; user sees a bounded error
+            return Err(AppError::new(StatusCode::BAD_GATEWAY, "ai_unreachable", format!("model endpoint unreachable: {e}")));
+        }
+    };
+    let usage_in: i64 = resp.get("usage").and_then(|u| u.get("prompt_tokens")).and_then(|v| v.as_i64()).unwrap_or(0);
+    let usage_out: i64 = resp.get("usage").and_then(|u| u.get("completion_tokens")).and_then(|v| v.as_i64()).unwrap_or(0);
+    if usage_in > cx.cfg.ai_input_token_cap {
+        return record_failure(&cx, &id, json!({"code": "ai_input_too_large", "input_tokens": usage_in}), "fail").await;
+    }
+    let choice_content = resp.get("choices").and_then(|c| c.get(0)).and_then(|c| c.get("message"))
+        .and_then(|m| m.get("content")).and_then(|c| c.as_str()).unwrap_or("").to_string();
+    let proposed_opt = extract_code_block(&choice_content);
+    let explanation: String = {
+        let without = strip_code_blocks(&choice_content);
+        if without.is_empty() { "Model returned code without explanation.".to_string() } else { without }
+    };
+    let Some(proposed) = proposed_opt else {
+        return record_failure(&cx, &id, json!({"code": "ai_no_code", "message": "model response contained no parseable full code block"}), "fail").await;
+    };
+    let diff = crate::util::unified_diff(&draft, &proposed);
+    let usage = json!({"input_tokens": usage_in, "output_tokens": usage_out, "measured": true});
+    cx.with_db(|db| -> AppResult<()> {
+        db.execute("UPDATE ai_requests SET status='succeeded', model=?1, explanation=?2, proposed_code=?3, diff=?4, usage=?5 WHERE id=?6",
+            rusqlite::params![cx.cfg.ai_model.clone(), explanation, &proposed, &diff, usage.to_string(), &id])?;
+        db.execute("INSERT INTO ai_usage (id,user_id,request_id,ts,kind,input_tokens,output_tokens) VALUES (?1,?2,?3,?4,'request',?5,?6)",
+            rusqlite::params![new_id(), &auth.id, &id, now_iso(), usage_in, usage_out])?;
+        Ok(())
+    }).await?;
+    audit(&cx, Some(&auth.id), "ai_assist", &id, "ok").await;
+    Ok(Json(json!({
+        "id": id,
+        "model": cx.cfg.ai_model,
+        "explanation": explanation,
+        "proposed_code": proposed,
+        "diff": diff,
+        "base_generation": gen,
+        "status": "succeeded",
+        "usage": usage,
+    })))
+}
+
+/// Stable AI gateway session id scoped to owner id + project UUID.
+/// Deterministic so each (owner, project) pair always reuses one upstream
+/// session; it contains no secrets (no API key material) and no email.
+pub fn ai_session_id(owner_id: &str, project_id: &str) -> String {
+    format!("sl-strategy-lab-{}", sha256_hex(format!("{owner_id}:{project_id}").as_bytes()))
+}
+
+async fn record_failure(cx: &Cx, id: &str, err: Value, status: &str) -> AppResult<Json<Value>> {
+    cx.with_db(|db| {
+        db.execute("UPDATE ai_requests SET status='failed', error=?1 WHERE id=?2", rusqlite::params![err.to_string(), id]).ok();
+    }).await;
+    audit(cx, None, "ai_failure", id, status).await;
+    Err(AppError::bad("ai_failure", "AI request failed; see usage ledger"))
+}
+
+fn strip_code_blocks(s: &str) -> String {
+    let mut out = String::new();
+    let mut inn = false;
+    for line in s.lines() {
+        if line.contains("```python") || line.contains("```") { inn = !inn; continue }
+        if !inn { out.push_str(line); out.push('\n'); }
+    }
+    out.trim().to_string()
+}
+
+pub fn extract_code_block(s: &str) -> Option<String> {
+    let mark = "```";
+    let mut in_block = false;
+    let mut block = String::new();
+    for l in s.lines() {
+        if !in_block && l.trim().starts_with(mark) { in_block = true; block.clear(); continue; }
+        if in_block && l.trim().starts_with(mark) { if !block.trim().is_empty() { return Some(block); } in_block = false; block.clear(); continue; }
+        if in_block { block.push_str(l); block.push('\n'); }
+    }
+    Some(block).filter(|b| !b.trim().is_empty())
+}
+
+/// Schema of the most recent ready dataset actually used by runs of this project.
+async fn summarize_dataset(cx: &Cx, user_id: &str, project_id: &str) -> AppResult<String> {
+    let q: Option<String> = cx.with_db(|db| {
+        db.query_row("SELECT d.manifest FROM datasets d JOIN runs r ON r.dataset_id=d.id WHERE r.project_id=?1 AND d.user_id=?2 AND d.status='ready' ORDER BY r.created_at DESC LIMIT 1",
+            rusqlite::params![project_id, user_id], |r| r.get::<_, String>(0)).ok()
+    }).await;
+    let Some(m) = q else { return Ok("No dataset ready in this project yet: columns unknown.".to_string()); };
+    let mj: Value = serde_json::from_str(&m).unwrap_or(Value::Null);
+    let mut cols = Vec::new();
+    if let Some(objs) = mj.get("objects").and_then(|o| o.as_array()) {
+        for o in objs.iter() {
+            if let Some(c) = o.get("columns") {
+                if let Some(arr) = c.as_array() {
+                    let cur: Vec<String> = arr.iter().filter_map(|v| v.as_str().map(String::from)).collect();
+                    if cur.len() > cols.len() { cols = cur; }
+                }
+            }
+        }
+    }
+    Ok(if cols.is_empty() { "Dataset ready but columns not enumerated.".to_string() } else { cols.join(", ") })
+}
+
+/// Accept a proposal and create a new draft and a version based on the AI proposal.
+pub async fn accept(cx: Cx, auth: AuthUser, path: Path<(String,)>, body: Option<Json<serde_json::Value>>) -> AppResult<Json<Value>> {
+    let (aid,) = path.0;
+    let axum::Json(j) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?;
+    let expected = j.get("expected_generation").and_then(|v| v.as_i64()).ok_or_else(|| AppError::bad("validation", "expected_generation required"))?;
+    let (user_id, project_id, proposed, base_gen, base_hash): (String, String, Option<String>, i64, Option<String>) = cx.with_db(|db| {
+        db.query_row("SELECT user_id,project_id,proposed_code,base_generation,base_code_hash FROM ai_requests WHERE id=?1",
+            [&aid], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?)))
+            .map_err(|_| AppError::not_found("AI request not found"))
+    }).await?;
+    if user_id != auth.id { return Err(AppError::not_found("AI request not found")); }
+    let Some(proposed) = proposed else { return Err(AppError::bad("ai_not_ready", "AI request failed or has no code to accept")); };
+    let p: Value = cx.with_db(|db| -> AppResult<Value> {
+        let (draft, gen): (String, i64) = db.query_row("SELECT draft_code,draft_generation FROM projects WHERE id=?1 AND user_id=?2",
+            rusqlite::params![&project_id, &auth.id], |r| Ok((r.get(0)?, r.get(1)?)))
+            .map_err(|_| AppError::not_found("project not found"))?;
+        if let Some(bh) = &base_hash {
+            if sha256_hex(draft.as_bytes()) != *bh {
+                return Err(AppError::conflict("stale_base", "current draft has changed since the proposal base; cannot auto-apply"));
+            }
+        }
+        if gen != base_gen {
+            return Err(AppError::conflict("stale_base", "proposal base does not match current generation"));
+        }
+        if gen != expected {
+            return Err(AppError::conflict("stale_generation", "draft changed; reload first"));
+        }
+        let hash = sha256_hex(proposed.as_bytes());
+        let ts = now_iso();
+        let vid = new_id();
+        db.execute("UPDATE projects SET draft_code=?1, draft_generation=?2, updated_at=?3 WHERE id=?4",
+            rusqlite::params![&proposed, gen + 1, &ts, &project_id])?;
+        db.execute("INSERT INTO project_versions (id,project_id,code,hash,message,source,created_at) VALUES (?1,?2,?3,?4,'accepted AI proposal','ai',?5)",
+            rusqlite::params![&vid, &project_id, &proposed, &hash, &ts])?;
+        db.execute("UPDATE ai_requests SET status='succeeded' WHERE id=?1", [&aid]).ok();
+        let mut st = db.prepare("SELECT id,name,description,draft_code,draft_generation,created_at,updated_at FROM projects WHERE id=?1")?;
+        let mut rows = st.query([&project_id])?;
+        let row = rows.next()?.ok_or_else(|| AppError::internal("project vanished"))?;
+        Ok(json!({
+            "id": row.get::<_, String>(0)?,
+            "name": row.get::<_, String>(1)?,
+            "description": row.get::<_, String>(2)?,
+            "draft_code": row.get::<_, String>(3)?,
+            "draft_generation": row.get::<_, i64>(4)?,
+            "created_at": row.get::<_, String>(5)?,
+            "updated_at": row.get::<_, String>(6)?,
+        }))
+    }).await?;
+    audit(&cx, Some(&auth.id), "ai_accept", &aid, "ok").await;
+    Ok(Json(p))
+}
+
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn extract_takes_first_fenced_full_python_block() {
+        let s = "prose\n```python\nclass Strategy(bt.Strategy):\n    pass\n```\ntrail";
+        assert_eq!(extract_code_block(s).unwrap(), "class Strategy(bt.Strategy):\n    pass\n");
+        assert!(extract_code_block("no fence at all").is_none(), "no hardcoded fake suggestion");
+        assert!(extract_code_block("```\n\n```").is_none(), "empty block rejected");
+    }
+
+    #[test]
+    fn usage_totals_never_invent_costs() {
+        // usage ledger records tokens measured, never price/cost
+        let j = json!({"usage": {"input_tokens": 11, "output_tokens": 7}});
+        assert!(j.get("usage").map(|_| true).unwrap_or(false));
+        assert!(!serde_json::to_string(&j["usage"]).unwrap().contains("cost"));
+    }
+
+    #[test]
+    fn ai_session_headers_are_stable_app_identity_without_secrets() {
+        // Honest app UA: claims only this internal POC coding-assistant identity.
+        let ua = "strategy-lab-coding-assistant/0.1 (internal POC, strategy coding help only)";
+        assert!(ua.starts_with("strategy-lab-coding-assistant/0.1"));
+        assert!(!ua.contains("opencode") && !ua.contains("curl"), "must not impersonate another client identity");
+        // Stable per owner+project: same scope -> same id, different scope -> different id.
+        let owner = "6f1d2b3a-1111-4aaa-9bbb-cccccccccccc";
+        let p1 = "00000000-2222-4333-8444-555555555555";
+        let p2 = "00000000-2222-4333-8444-666666666666";
+        let s1 = ai_session_id(owner, p1);
+        assert_eq!(s1, ai_session_id(owner, p1), "session must be stable across requests for owner+project");
+        assert_ne!(s1, ai_session_id(owner, p2), "session is scoped to the project");
+        assert_ne!(s1, ai_session_id("another-owner", p1), "session is scoped to the owner id");
+        // No secret material ever rides in the session header.
+        assert!(!s1.contains(owner) && !s1.contains(p1));
+        assert_eq!(s1.len(), 16 + 64, "prefix + sha256 hex of owner:project");
+    }
+}
diff --git a/server/src/auth.rs b/server/src/auth.rs
new file mode 100644
index 0000000..8942a83
--- /dev/null
+++ b/server/src/auth.rs
@@ -0,0 +1,713 @@
+use std::sync::Arc;
+use axum::{
+    extract::{FromRequestParts, Path},
+    http::{header, request::Parts, HeaderMap, HeaderValue, StatusCode},
+    response::{IntoResponse, Response},
+    Json,
+};
+use argon2::{Argon2, PasswordHash, PasswordHasher, PasswordVerifier, password_hash::SaltString};
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+
+use crate::error::{AppError, AppResult};
+use crate::state::{AppState, Cx};
+use crate::util::{gen_token, new_id, now_iso, plus_hours, sha256_hex};
+
+pub const COOKIE_NAME: &str = "sl_session";
+
+const USER_COLS_FULL: &str = "id,email,password_hash,name,role,active,ai_enabled,daily_run_limit,created_at";
+
+struct RowUser {
+    id: String, email: String, password_hash: String, name: String,
+    role: String, active: i64, ai_enabled: i64, daily_run_limit: i64, created_at: String,
+}
+
+fn row_user(r: &rusqlite::Row) -> rusqlite::Result<RowUser> {
+    Ok(RowUser {
+        id: r.get(0)?, email: r.get(1)?, password_hash: r.get(2)?, name: r.get(3)?,
+        role: r.get(4)?, active: r.get(5)?, ai_enabled: r.get(6)?, daily_run_limit: r.get(7)?, created_at: r.get(8)?,
+    })
+}
+
+fn user_json(u: &RowUser) -> serde_json::Value {
+    json!({
+        "id": u.id, "email": u.email, "name": u.name, "role": u.role,
+        "active": u.active != 0, "ai_enabled": u.ai_enabled != 0,
+        "daily_run_limit": u.daily_run_limit, "created_at": u.created_at,
+    })
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub struct User {
+    pub id: String,
+    pub email: String,
+    pub name: String,
+    pub role: String,
+    pub active: bool,
+    pub ai_enabled: bool,
+    pub daily_run_limit: i64,
+    pub created_at: String,
+}
+
+#[derive(Clone)]
+pub struct AuthUser {
+    pub id: String,
+    pub email: String,
+    pub role: String,
+    pub ai_enabled: bool,
+    pub daily_run_limit: i64,
+    pub session_id: String,
+}
+
+impl FromRequestParts<Arc<AppState>> for AuthUser {
+    type Rejection = AppError;
+    async fn from_request_parts(parts: &mut Parts, state: &Arc<AppState>) -> Result<Self, Self::Rejection> {
+        let Some(token) = cookie_token(&parts.headers) else {
+            return Err(AppError::unauthorized("authentication required"));
+        };
+        if token.len() < 32 {
+            return Err(AppError::unauthorized("invalid or expired session"));
+        }
+        let id = sha256_hex(token.as_bytes());
+        let now = now_iso();
+        let user = state.with_db(|db| {
+            db.query_row(
+                "SELECT u.id,u.email,u.role,u.ai_enabled,u.daily_run_limit,s.id FROM sessions s JOIN users u ON u.id=s.user_id WHERE s.id=?1 AND s.expires_at > ?2 AND u.active=1",
+                [&id, &now],
+                |r| Ok(AuthUser { id: r.get(0)?, email: r.get(1)?, role: r.get(2)?, ai_enabled: r.get::<_, i64>(3)? != 0, daily_run_limit: r.get(4)?, session_id: r.get(5)? }),
+            ).ok()
+        }).await;
+        user.ok_or_else(|| AppError::unauthorized("invalid or expired session"))
+    }
+}
+
+/// Argon2id password hashing with a fresh random salt per call.
+pub fn hash_password(p: &str) -> AppResult<String> {
+    let salt = SaltString::generate(&mut rand::rngs::OsRng);
+    Argon2::default().hash_password(p.as_bytes(), &salt)
+        .map(|h| h.to_string())
+        .map_err(|e| AppError::internal(format!("hash: {e}")))
+}
+
+pub fn verify_password(hash: &str, p: &str) -> bool {
+    PasswordHash::new(hash).ok()
+        .and_then(|h| Argon2::default().verify_password(p.as_bytes(), &h).ok())
+        .is_some()
+}
+
+pub async fn fetch_user(cx: &Cx, id: &str) -> AppResult<User> {
+    cx.with_db(|db| {
+        db.query_row(
+            "SELECT id,email,name,role,active,ai_enabled,daily_run_limit,created_at FROM users WHERE id=?1",
+            [id],
+            |r| Ok(User { id: r.get(0)?, email: r.get(1)?, name: r.get(2)?, role: r.get(3)?, active: r.get::<_, i64>(4)? != 0, ai_enabled: r.get::<_, i64>(5)? != 0, daily_run_limit: r.get(6)?, created_at: r.get(7)? }),
+        ).map_err(|_| AppError::not_found("user not found"))
+    }).await
+}
+
+pub fn cookie_token(headers: &HeaderMap) -> Option<String> {
+    let raw = headers.get(header::COOKIE)?.to_str().ok()?;
+    raw.split(';').find_map(|c| {
+        let c = c.trim();
+        c.strip_prefix(COOKIE_NAME)
+            .and_then(|v| v.strip_prefix('='))
+            .map(str::to_string)
+            .filter(|v| !v.is_empty())
+    })
+}
+
+/// Issue a random 256-bit session token; only sha256(token) is stored.
+pub async fn session_cookie(cx: &Cx, user_id: &str) -> AppResult<(String, String)> {
+    let token = gen_token();
+    let id = sha256_hex(token.as_bytes());
+    let expires = plus_hours(cx.cfg.session_hours);
+    cx.with_db(|db| -> AppResult<()> {
+        db.execute("INSERT INTO sessions (id,user_id,created_at,expires_at) VALUES (?1,?2,?3,?4)",
+            rusqlite::params![&id, user_id, now_iso(), &expires])?;
+        Ok(())
+    }).await?;
+    let mut cookie = format!("{COOKIE_NAME}={token}; Path=/; HttpOnly; SameSite=Strict; Max-Age={}", cx.cfg.session_hours * 3600);
+    if cx.cfg.secure_cookies {
+        cookie.push_str("; Secure");
+    }
+    Ok((cookie, expires))
+}
+
+pub fn clear_cookie() -> String {
+    format!("{COOKIE_NAME}=; Path=/; HttpOnly; SameSite=Strict; Max-Age=0")
+}
+
+pub async fn revoke_all_sessions(cx: &Cx, user_id: &str) -> AppResult<()> {
+    cx.with_db(|db| -> AppResult<()> {
+        db.execute("DELETE FROM sessions WHERE user_id=?1", [user_id])?;
+        Ok(())
+    }).await
+}
+
+/// Write a sanitized audit event. Callers must never pass passwords, keys,
+/// tokens, strategy source or object payloads; only fixed action labels and
+/// id/email targets, which sanitize_target bounds before storage.
+pub async fn audit(cx: &Cx, actor: Option<&str>, action: &str, target: &str, status: &str) {
+    let target = sanitize_target(target);
+    cx.with_db(|db| {
+        db.execute("INSERT INTO audit (ts,actor_id,action,target,status) VALUES (?1,?2,?3,?4,?5)",
+            rusqlite::params![now_iso(), actor.map(str::to_string), action, target, status]).ok()
+    }).await;
+}
+
+fn sanitize_target(t: &str) -> String {
+    if t.len() > 300 || t.lines().count() > 3 {
+        "redacted-oversized".to_string()
+    } else {
+        t.to_string()
+    }
+}
+
+fn with_cookie(mut resp: Response, cookie: &str) -> Response {
+    if let Ok(hv) = HeaderValue::from_str(cookie) {
+        resp.headers_mut().insert(header::SET_COOKIE, hv);
+    }
+    resp
+}
+
+async fn record_failure(cx: &Cx, email: &str) {
+    cx.with_db(|db| {
+        db.execute("INSERT INTO login_failures (email,failed_at) VALUES (?1,?2)",
+            rusqlite::params![email.to_string(), now_iso()]).ok()
+    }).await;
+}
+
+/// Admin-issued one-time hashed token ("invitations" or "password_resets").
+/// Returns (raw_token shown once, expires_at). Only sha256 is persisted.
+pub async fn make_admin_token(
+    cx: &Cx,
+    table: &str,
+    user_id: Option<&str>,
+    hours: i64,
+    email: Option<&str>,
+) -> AppResult<(String, String)> {
+    let token = gen_token();
+    let th = sha256_hex(token.as_bytes());
+    let expires = plus_hours(hours);
+    let id = new_id();
+    let email = email.map(|e| e.trim().to_lowercase());
+    cx.with_db(|db| -> AppResult<()> {
+        match table {
+            "invitations" => {
+                db.execute("INSERT INTO invitations (id,email,token_hash,role,expires_at,created_at) VALUES (?1,?2,?3,'member',?4,?5)",
+                    rusqlite::params![&id, email, &th, &expires, now_iso()])?;
+            }
+            "password_resets" => {
+                let uid = user_id.ok_or_else(|| AppError::bad("validation", "reset requires target user"))?;
+                db.execute("INSERT INTO password_resets (id,user_id,token_hash,expires_at) VALUES (?1,?2,?3,?4)",
+                    rusqlite::params![&id, uid, &th, &expires])?;
+            }
+            other => return Err(AppError::internal(format!("unknown token table {other}"))),
+        }
+        Ok(())
+    }).await?;
+    Ok((token, expires))
+}
+
+pub async fn purge_expired(cx: &Cx) {
+    cx.with_db(|db| {
+        let now = now_iso();
+        db.execute("DELETE FROM sessions WHERE expires_at <= ?1", [&now]).ok();
+        db.execute("DELETE FROM invitations WHERE expires_at <= ?1", [&now]).ok();
+        db.execute("DELETE FROM password_resets WHERE expires_at <= ?1", [&now]).ok();
+        db.execute("DELETE FROM login_failures WHERE failed_at <= ?1", [&plus_hours(-2)]).ok();
+    }).await;
+}
+
+// ---- handlers ----
+
+#[derive(Deserialize)]
+pub struct LoginReq { pub email: String, pub password: String }
+
+pub async fn login(cx: Cx, body: Option<Json<LoginReq>>) -> AppResult<Response> {
+    let Json(r) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?;
+    let email = r.email.trim().to_lowercase();
+    if email.is_empty() || r.password.is_empty() {
+        return Err(AppError::bad("validation", "email and password required"));
+    }
+    let failures: i64 = cx.with_db(|db| {
+        db.query_row("SELECT COUNT(*) FROM login_failures WHERE email=?1 AND failed_at >= ?2",
+            [&email, &plus_hours(-1)], |row| row.get(0)).unwrap_or(0)
+    }).await;
+    if failures > 20 {
+        audit(&cx, None, "login_throttled", &email, "rate_limited").await;
+        return Err(AppError::new(StatusCode::TOO_MANY_REQUESTS, "rate_limited", "too many failed logins; try again later"));
+    }
+    let Some(u) = cx.with_db(|db| {
+        db.query_row(&format!("SELECT {USER_COLS_FULL} FROM users WHERE email=?1"), [&email], row_user).ok()
+    }).await else {
+        record_failure(&cx, &email).await;
+        audit(&cx, None, "login_failed", &email, "invalid_credentials").await;
+        return Err(AppError::unauthorized("invalid credentials"));
+    };
+    if !verify_password(&u.password_hash, &r.password) {
+        record_failure(&cx, &email).await;
+        audit(&cx, Some(&u.id), "login_failed", &u.id, "invalid_credentials").await;
+        return Err(AppError::unauthorized("invalid credentials"));
+    }
+    if u.active == 0 {
+        audit(&cx, Some(&u.id), "login_denied", &u.email, "disabled").await;
+        return Err(AppError::forbidden("account is disabled"));
+    }
+    purge_expired(&cx).await;
+    let (cookie, _) = session_cookie(&cx, &u.id).await?;
+    audit(&cx, Some(&u.id), "login", &u.id, "ok").await;
+    let resp = Json(user_json(&u)).into_response();
+    Ok(with_cookie(resp, &cookie))
+}
+
+/// Registration payload has NO role field: role comes only from the
+/// admin-issued invitation ("member" only in this POC). serde ignores extra
+/// client fields, so role escalation at registration is impossible.
+#[derive(Deserialize)]
+pub struct RegisterReq {
+    pub invite_token: String,
+    pub name: String,
+    pub email: String,
+    pub password: String,
+}
+
+pub async fn register(cx: Cx, body: Option<Json<RegisterReq>>) -> AppResult<Response> {
+    let Json(r) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?;
+    if r.name.trim().is_empty() || r.name.trim().len() > 200 { return Err(AppError::bad("validation", "name required (max 200)")); }
+    if r.password.len() < 8 || r.password.len() > 256 { return Err(AppError::bad("validation", "password must be 8-256 characters")); }
+    let email = r.email.trim().to_lowercase();
+    if !email.contains('@') || email.len() < 5 || email.len() > 320 { return Err(AppError::bad("validation", "valid email required")); }
+    let pw = hash_password(&r.password)?;
+    let token_hash = sha256_hex(r.invite_token.as_bytes());
+    let day_limit = cx.cfg.default_run_limit_per_day;
+    let user_id: String = cx.with_db(|db| -> AppResult<String> {
+        // One-time atomically consumed invitation with email binding: the
+        // select, binding/role/expiry/used checks, user insert and invite
+        // marking all run inside ONE transaction; the marking (
+        // UPDATE ... WHERE used_by IS NULL) guarantees concurrent retries lose.
+        let now = now_iso();
+        let tx = db.transaction()?;
+        let invite: Option<(String, String, Option<String>)> = tx.query_row(
+            "SELECT id,role,email FROM invitations WHERE token_hash=?1",
+            [&token_hash], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?))).ok();
+        let Some((inv_id, role, inv_email)) = invite else {
+            return Err(AppError::bad("invite_invalid", "invalid, used or expired invitation token"));
+        };
+        if role != "member" {
+            return Err(AppError::bad("invite_invalid", "invitation does not permit this role"));
+        }
+        // Email binding: an invitation addressed to a specific email can only
+        // be consumed with that exact (case-insensitive) email.
+        if let Some(bound) = &inv_email {
+            let bound = bound.trim().to_lowercase();
+            if !bound.is_empty() && bound != email {
+                return Err(AppError::bad("invite_email_mismatch", "invitation is bound to a different email"));
+            }
+        }
+        let expired = tx.query_row("SELECT expires_at FROM invitations WHERE id=?1", [&inv_id], |r| r.get::<_, String>(0))
+            .map(|exp| exp <= now).unwrap_or(true);
+        if expired {
+            return Err(AppError::bad("invite_invalid", "invalid, used or expired invitation token"));
+        }
+        if tx.query_row("SELECT COUNT(*) FROM users WHERE email=?1", [&email], |c| c.get::<_, i64>(0)).unwrap_or(0) > 0 {
+            return Err(AppError::conflict("email_taken", "an account with this email already exists"));
+        }
+        let uid = new_id();
+        tx.execute(
+            "INSERT INTO users (id,email,password_hash,name,role,active,ai_enabled,daily_run_limit,created_at) VALUES (?1,?2,?3,?4,'member',1,0,?5,?6)",
+            rusqlite::params![&uid, &email, &pw, r.name.trim(), day_limit, &now])?;
+        let n = tx.execute("UPDATE invitations SET used_by=?1 WHERE id=?2 AND used_by IS NULL AND expires_at > ?3",
+            rusqlite::params![&uid, &inv_id, &now])?;
+        if n != 1 {
+            return Err(AppError::conflict("invite_used", "invitation token already consumed"));
+        }
+        tx.commit()?;
+        Ok(uid)
+    }).await?;
+    audit(&cx, Some(&user_id), "register", &email, "ok").await;
+    purge_expired(&cx).await;
+    let (cookie, _) = session_cookie(&cx, &user_id).await?;
+    let u = fetch_user(&cx, &user_id).await?;
+    let resp = (StatusCode::CREATED, Json(json!({ "user": u }))).into_response();
+    Ok(with_cookie(resp, &cookie))
+}
+
+pub async fn logout(cx: Cx, auth: AuthUser, headers: HeaderMap) -> AppResult<Response> {
+    if let Some(t) = cookie_token(&headers) {
+        let id = sha256_hex(t.as_bytes());
+        cx.with_db(|db| { db.execute("DELETE FROM sessions WHERE id=?1", [&id]).ok() }).await;
+    }
+    audit(&cx, Some(&auth.id), "logout", &auth.id, "ok").await;
+    let resp = StatusCode::NO_CONTENT.into_response();
+    Ok(with_cookie(resp, &clear_cookie()))
+}
+
+pub async fn me(cx: Cx, auth: AuthUser) -> AppResult<Json<serde_json::Value>> {
+    let u = fetch_user(&cx, &auth.id).await?;
+    Ok(Json(json!({ "user": u })))
+}
+
+#[derive(Deserialize)]
+pub struct ProfileReq { pub name: String }
+
+pub async fn patch_profile(cx: Cx, auth: AuthUser, body: Option<Json<ProfileReq>>) -> AppResult<Json<serde_json::Value>> {
+    let Json(r) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?;
+    let name = r.name.trim().to_string();
+    if name.is_empty() || name.len() > 200 { return Err(AppError::bad("validation", "name required (max 200)")); }
+    cx.with_db(|db| -> AppResult<()> {
+        let n = db.execute("UPDATE users SET name=?1 WHERE id=?2", rusqlite::params![&name, &auth.id])?;
+        if n == 0 { return Err(AppError::not_found("user not found")); }
+        Ok(())
+    }).await?;
+    audit(&cx, Some(&auth.id), "profile_update", &auth.id, "ok").await;
+    let u = fetch_user(&cx, &auth.id).await?;
+    Ok(Json(json!({ "user": u })))
+}
+
+#[derive(Deserialize)]
+pub struct PasswordReq { pub current_password: String, pub new_password: String }
+
+pub async fn change_password(cx: Cx, auth: AuthUser, headers: HeaderMap, body: Option<Json<PasswordReq>>) -> AppResult<Json<serde_json::Value>> {
+    let Json(r) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?;
+    if r.new_password.len() < 8 || r.new_password.len() > 256 {
+        return Err(AppError::bad("validation", "new password must be 8-256 characters"));
+    }
+    let hash: String = cx.with_db(|db| {
+        db.query_row("SELECT password_hash FROM users WHERE id=?1", [&auth.id], |row| row.get(0))
+            .map_err(|_| AppError::unauthorized("authentication required"))
+    }).await?;
+    if !verify_password(&hash, &r.current_password) {
+        audit(&cx, Some(&auth.id), "password_change", &auth.id, "wrong_current_password").await;
+        return Err(AppError::unauthorized("current password incorrect"));
+    }
+    let new_hash = hash_password(&r.new_password)?;
+    // Keep the current session, revoke all others; update + revoke atomically.
+    let keep = cookie_token(&headers).map(|t| sha256_hex(t.as_bytes())).unwrap_or_default();
+    cx.with_db(|db| -> AppResult<()> {
+        let tx = db.transaction()?;
+        tx.execute("UPDATE users SET password_hash=?1 WHERE id=?2", rusqlite::params![&new_hash, &auth.id])?;
+        tx.execute("DELETE FROM sessions WHERE user_id=?1 AND id != ?2", rusqlite::params![&auth.id, &keep])?;
+        tx.commit()?;
+        Ok(())
+    }).await?;
+    audit(&cx, Some(&auth.id), "password_change", &auth.id, "ok").await;
+    Ok(Json(json!({ "ok": true, "other_sessions_revoked": true })))
+}
+
+/// Own-session listing, sanitized: sha256-derived ids only, never raw tokens;
+/// expired sessions are never listed. `current` marks the caller's live session.
+#[derive(Serialize)]
+pub struct SessionInfo { pub id: String, pub created_at: String, pub expires_at: String }
+
+pub async fn list_sessions(cx: Cx, auth: AuthUser) -> AppResult<Json<serde_json::Value>> {
+    let current = auth.session_id.clone();
+    let rows: Vec<(String, String, String)> = cx.with_db(|db| {
+        let now = now_iso();
+        let mut st = db.prepare("SELECT id,created_at,expires_at FROM sessions WHERE user_id=?1 AND expires_at > ?2 ORDER BY created_at DESC")?;
+        let rows = st
+            .query_map(rusqlite::params![&auth.id, &now], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))?
+            .collect::<Result<Vec<_>, _>>()?;
+        Ok(rows)
+    }).await.map_err(|e: rusqlite::Error| AppError::internal(e.to_string()))?;
+    Ok(Json(json!({
+        "items": rows.into_iter().map(|(id, c, e)| json!({
+            "id": id, "created_at": c, "expires_at": e,
+            "current": id == current,
+        })).collect::<Vec<_>>()
+    })))
+}
+
+pub async fn delete_session(cx: Cx, auth: AuthUser, Path(id): Path<String>) -> AppResult<Json<serde_json::Value>> {
+    cx.with_db(|db| -> AppResult<()> {
+        if db.execute("DELETE FROM sessions WHERE id=?1 AND user_id=?2", rusqlite::params![&id, &auth.id])? == 0 {
+            // Never reveal another user's session ids.
+            return Err(AppError::not_found("session not found"));
+        }
+        Ok(())
+    }).await?;
+    audit(&cx, Some(&auth.id), "session_revoke", &auth.id, "ok").await;
+    Ok(Json(json!({"ok": true})))
+}
+
+#[derive(Deserialize)]
+pub struct ResetReq { pub token: String, pub new_password: String }
+
+pub async fn reset_password(cx: Cx, body: Option<Json<ResetReq>>) -> AppResult<Json<serde_json::Value>> {
+    let Json(r) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?;
+    if r.new_password.len() < 8 || r.new_password.len() > 256 {
+        return Err(AppError::bad("validation", "password must be 8-256 characters"));
+    }
+    let new_hash = hash_password(&r.new_password)?;
+    let th = sha256_hex(r.token.as_bytes());
+    // Single-use consumption plus password update plus session revocation in
+    // ONE transaction: a second submission of the same token cannot succeed.
+    let uid: String = cx.with_db(|db| -> AppResult<String> {
+        let tx = db.transaction()?;
+        let (uid, used): (String, i64) = tx.query_row(
+            "SELECT user_id,used FROM password_resets WHERE token_hash=?1",
+            [&th], |row| Ok((row.get(0)?, row.get(1)?)))
+            .map_err(|_| AppError::bad("reset_invalid", "invalid or expired reset token"))?;
+        if used != 0 {
+            return Err(AppError::conflict("reset_used", "reset token already consumed"));
+        }
+        let expired = tx.query_row("SELECT expires_at FROM password_resets WHERE token_hash=?1", [&th], |row| row.get::<_, String>(0))
+            .map(|exp| exp <= now_iso()).unwrap_or(true);
+        if expired {
+            return Err(AppError::bad("reset_invalid", "invalid or expired reset token"));
+        }
+        tx.execute("UPDATE password_resets SET used=1 WHERE token_hash=?1", [&th])?;
+        tx.execute("UPDATE users SET password_hash=?1 WHERE id=?2", rusqlite::params![&new_hash, &uid])?;
+        tx.execute("DELETE FROM sessions WHERE user_id=?1", [&uid])?;
+        tx.commit()?;
+        Ok(uid)
+    }).await?;
+    audit(&cx, Some(&uid), "password_reset", &uid, "ok").await;
+    Ok(Json(json!({"ok": true})))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::db::init_db;
+
+    fn st() -> Arc<AppState> {
+        let conn = rusqlite::Connection::open_in_memory().expect("in-memory db");
+        init_db(&conn).expect("schema");
+        let cfg = crate::config::Config {
+            bind_addr: "127.0.0.1:0".into(),
+            canonical_origin: String::new(),
+            secure_cookies: false,
+            db_path: ":memory:".into(),
+            frontend_dir: "frontend/dist".into(),
+            data_dir: std::env::temp_dir().to_string_lossy().into_owned(),
+            worker_image: "test".into(),
+            fetch_timeout_secs: 1,
+            backtest_timeout_secs: 1,
+            run_concurrency: 1,
+            fetch_concurrency: 1,
+            session_hours: 24,
+            bootstrap_admin_email: None,
+            bootstrap_admin_password: None,
+            ai_base_url: "http://127.0.0.1:9".into(),
+            ai_model: "test-model".into(),
+            ai_daily_request_cap: 1,
+            ai_input_token_cap: 1,
+            ai_output_token_cap: 1,
+            ai_enabled_poc: false,
+            default_run_limit_per_day: 10,
+            version: "test".into(),
+        };
+        Arc::new(AppState {
+            cfg,
+            db: tokio::sync::Mutex::new(conn),
+            run_sem: Arc::new(tokio::sync::Semaphore::new(1)),
+            fetch_sem: Arc::new(tokio::sync::Semaphore::new(1)),
+        })
+    }
+
+    fn handle(s: &Arc<AppState>) -> Cx { axum::extract::State(s.clone()) }
+
+    fn cookie_of(token: &str) -> HeaderMap {
+        let mut hm = HeaderMap::new();
+        hm.insert(header::COOKIE, HeaderValue::from_str(&format!("{COOKIE_NAME}={token}")).unwrap());
+        hm
+    }
+
+    /// Register through the real handler, then resolve the session exactly the
+    /// way the AuthUser extractor does, to produce a test auth context.
+    async fn register_ok(s: &Arc<AppState>, email: &str, name: &str, password: &str) -> (AuthUser, HeaderMap) {
+        let (inv, _) = make_admin_token(&handle(s), "invitations", None, 24, Some(email)).await.unwrap();
+        let body = RegisterReq {
+            invite_token: inv,
+            name: name.into(),
+            email: email.into(),
+            password: password.into(),
+        };
+        let resp = register(handle(s), Some(Json(body))).await.expect("register ok");
+        let cookie = resp.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().to_string();
+        let tok = cookie.split(';').next().unwrap()
+            .strip_prefix(COOKIE_NAME).and_then(|c| c.strip_prefix('='))
+            .expect("cookie contains the raw session token").to_string();
+        let now = now_iso();
+        let au = s.with_db(|db| {
+            db.query_row(
+                "SELECT u.id,u.email,u.role,u.ai_enabled,u.daily_run_limit,s.id FROM sessions s JOIN users u ON u.id=s.user_id WHERE s.id=?1 AND s.expires_at > ?2",
+                [&sha256_hex(tok.as_bytes()), &now],
+                |r| Ok(AuthUser { id: r.get(0)?, email: r.get(1)?, role: r.get(2)?, ai_enabled: r.get::<_, i64>(3)? != 0, daily_run_limit: r.get(4)?, session_id: r.get(5)? }),
+            ).unwrap()
+        }).await;
+        (au, cookie_of(&tok))
+    }
+
+    #[tokio::test]
+    async fn password_hashes_are_argon2() {
+        let h = hash_password("correct horse battery").unwrap();
+        assert!(h.starts_with("$argon2"));
+        assert!(verify_password(&h, "correct horse battery"));
+        assert!(!verify_password(&h, "wrong password"));
+    }
+
+    #[tokio::test]
+    async fn register_creates_member_and_ignores_client_role() {
+        let s = st();
+        let (auth, _) = register_ok(&s, "[email protected]", "Tester User", "a-reasonable-passphrase-1").await;
+        assert_eq!(auth.email, "[email protected]");
+        assert_eq!(auth.role, "member", "role must never come from client input");
+        assert!(!auth.ai_enabled, "ai stays off by default for members");
+        // No password hash ever appears in user-facing rows' JSON contract:
+        // the users table keeps the hash, API payloads never include it.
+        let u = fetch_user(&handle(&s), &auth.id).await.unwrap();
+        let payload = serde_json::to_string(&u).unwrap();
+        assert!(!payload.contains("password_hash"));
+        assert!(!payload.contains("argon2"));
+    }
+
+    #[tokio::test]
+    async fn invitation_is_single_use() {
+        let s = st();
+        let (inv, _) = make_admin_token(&handle(&s), "invitations", None, 24, None).await.unwrap();
+        let body = RegisterReq {
+            invite_token: inv,
+            name: "Once".into(),
+            email: "[email protected]".into(),
+            password: "a-reasonable-passphrase-1".into(),
+        };
+        register(handle(&s), Some(Json(body))).await.expect("first use ok");
+        let consumed: i64 = handle(&s).with_db(|db| db.query_row(
+            "SELECT COUNT(*) FROM invitations WHERE used_by IS NOT NULL", [], |r| r.get(0)).unwrap()).await;
+        assert_eq!(consumed, 1);
+        let token_rows: i64 = handle(&s).with_db(|db| db.query_row(
+            "SELECT COUNT(*) FROM invitations WHERE used_by IS NULL", [], |r| r.get(0)).unwrap()).await;
+        assert_eq!(token_rows, 0, "invitation must be one-time");
+    }
+
+    #[tokio::test]
+    async fn invitation_email_binding_enforced() {
+        let s = st();
+        let (inv, _) = make_admin_token(&handle(&s), "invitations", None, 24, Some("[email protected]")).await.unwrap();
+        let body = RegisterReq {
+            invite_token: inv,
+            name: "Bound".into(),
+            email: "[email protected]".into(),
+            password: "a-reasonable-passphrase-1".into(),
+        };
+        let err = register(handle(&s), Some(Json(body))).await.err().expect("must reject mismatched email");
+        assert_eq!(err.code, "invite_email_mismatch");
+        // The token was NOT consumed by the failed attempt.
+        let unused: i64 = handle(&s).with_db(|db| {
+            db.query_row("SELECT COUNT(*) FROM invitations WHERE used_by IS NULL", [], |r| r.get(0)).unwrap()
+        }).await;
+        assert_eq!(unused, 1);
+    }
+
+    #[tokio::test]
+    async fn reset_password_is_single_use_and_revokes_sessions() {
+        let s = st();
+        let (auth, _) = register_ok(&s, "[email protected]", "Reset User", "original-passphrase-1").await;
+        handle(&s).with_db(|db| {
+            db.execute("INSERT INTO sessions (id,user_id,created_at,expires_at) VALUES ('dead-session',?1,?2,?3)",
+                rusqlite::params![&auth.id, now_iso(), plus_hours(1)]).unwrap();
+        }).await;
+        let (token, _) = make_admin_token(&handle(&s), "password_resets", Some(auth.id.as_str()), 1, None).await.unwrap();
+        reset_password(handle(&s), Some(Json(ResetReq { token: token.clone(), new_password: "new-passphrase-1".into() }))).await.unwrap();
+        let sessions: i64 = handle(&s).with_db(|db| db.query_row(
+            "SELECT COUNT(*) FROM sessions WHERE user_id=?1", [&auth.id], |r| r.get(0)).unwrap()).await;
+        assert_eq!(sessions, 0, "reset must revoke all sessions");
+        let second = reset_password(handle(&s), Some(Json(ResetReq { token, new_password: "new-passphrase-2".into() }))).await;
+        assert!(second.is_err(), "reset token must be single-use");
+        let hash: String = handle(&s).with_db(|db| db.query_row(
+            "SELECT password_hash FROM users WHERE id=?1", [&auth.id], |r| r.get(0)).unwrap()).await;
+        assert!(verify_password(&hash, "new-passphrase-1"));
+        assert!(!verify_password(&hash, "new-passphrase-2"));
+    }
+
+    #[tokio::test]
+    async fn change_password_revokes_others_keeps_current() {
+        let s = st();
+        let (auth, headers) = register_ok(&s, "[email protected]", "Change User", "original-passphrase-1").await;
+        let extra = gen_token();
+        let extra_id = sha256_hex(extra.as_bytes());
+        handle(&s).with_db(|db| {
+            db.execute("INSERT INTO sessions (id,user_id,created_at,expires_at) VALUES (?1,?2,?3,?4)",
+                rusqlite::params![&extra_id, &auth.id, now_iso(), plus_hours(1)]).unwrap();
+        }).await;
+        let done = change_password(handle(&s), auth.clone(), headers.clone(), Some(Json(PasswordReq {
+            current_password: "original-passphrase-1".into(),
+            new_password: "brand-new-passphrase".into(),
+        }))).await;
+        done.unwrap_or_else(|e| panic!("change failed: {e}"));
+        let remaining: i64 = handle(&s).with_db(|db| {
+            let now = now_iso();
+            db.query_row("SELECT COUNT(*) FROM sessions WHERE user_id=?1 AND expires_at > ?2", [&auth.id, &now], |r| r.get(0)).unwrap()
+        }).await;
+        assert_eq!(remaining, 1, "only the current session survives");
+        let wrong = change_password(handle(&s), auth.clone(), headers, Some(Json(PasswordReq {
+            current_password: "wrong".into(),
+            new_password: "another-passphrase-1".into(),
+        }))).await;
+        assert!(wrong.is_err(), "wrong current password must be rejected");
+    }
+
+    #[tokio::test]
+    async fn audit_never_records_secrets() {
+        let s = st();
+        let long = format!("{}password=secret-value", "x".repeat(400));
+        audit(&handle(&s), None, "login_failed", &long, "fail").await;
+        let stored: String = handle(&s).with_db(|db| db.query_row(
+            "SELECT target FROM audit ORDER BY seq DESC LIMIT 1", [], |r| r.get(0)).unwrap()).await;
+        assert_eq!(stored, "redacted-oversized");
+        let dirty: i64 = handle(&s).with_db(|db| {
+            db.query_row("SELECT COUNT(*) FROM audit WHERE target LIKE '%password=%'", [], |r| r.get(0)).unwrap()
+        }).await;
+        assert_eq!(dirty, 0, "secrets must never be written into the audit trail");
+    }
+
+    #[tokio::test]
+    async fn cookie_sessions_are_hashed_never_plaintext() {
+        let s = st();
+        handle(&s).with_db(|db| {
+            db.execute("INSERT INTO users (id,email,password_hash,name,role,active,ai_enabled,daily_run_limit,created_at) VALUES ('some-user','[email protected]','h','n','member',1,0,10,?1)",
+                [now_iso()]).unwrap();
+        }).await;
+        let (cookie, _) = session_cookie(&handle(&s), "some-user").await.unwrap();
+        let tok = cookie.split(';').next().unwrap()
+            .strip_prefix(COOKIE_NAME).and_then(|c| c.strip_prefix('=')).unwrap();
+        let plaintext: i64 = handle(&s).with_db(|db| db.query_row(
+            "SELECT COUNT(*) FROM sessions WHERE id=?1", [&tok], |r| r.get(0)).unwrap()).await;
+        assert_eq!(plaintext, 0, "raw token must never be stored");
+        let hashed: i64 = handle(&s).with_db(|db| db.query_row(
+            "SELECT COUNT(*) FROM sessions WHERE id=?1", [&sha256_hex(tok.as_bytes())], |r| r.get(0)).unwrap()).await;
+        assert_eq!(hashed, 1);
+        assert!(cookie.contains("HttpOnly") && cookie.contains("SameSite=Strict"));
+    }
+
+    #[tokio::test]
+    async fn login_throttles_after_repeated_failures() {
+        let s = st();
+        for _ in 0..25 {
+            record_failure(&handle(&s), "[email protected]").await;
+        }
+        let res = login(handle(&s), Some(Json(LoginReq {
+            email: "[email protected]".into(),
+            password: "whatever-passphrase-1".into(),
+        }))).await;
+        match res {
+            Err(e) => assert_eq!(e.status, StatusCode::TOO_MANY_REQUESTS),
+            Ok(_) => panic!("login must be throttled after sustained failures"),
+        }
+    }
+
+    #[tokio::test]
+    async fn logout_invalidates_the_server_session() {
+        let s = st();
+        let (auth, headers) = register_ok(&s, "[email protected]", "Logout User", "a-reasonable-passphrase-1").await;
+        let tok = headers.get(header::COOKIE).unwrap().to_str().unwrap()
+            .split(';').next().unwrap()
+            .strip_prefix(COOKIE_NAME).and_then(|c| c.strip_prefix('=')).unwrap().to_string();
+        logout(handle(&s), auth, headers).await.unwrap();
+        let left: i64 = handle(&s).with_db(|db| db.query_row(
+            "SELECT COUNT(*) FROM sessions WHERE id=?1", [&sha256_hex(tok.as_bytes())], |r| r.get(0)).unwrap()).await;
+        assert_eq!(left, 0, "logout must delete the server-side session row");
+    }
+}
diff --git a/server/src/config.rs b/server/src/config.rs
new file mode 100644
index 0000000..410cb09
--- /dev/null
+++ b/server/src/config.rs
@@ -0,0 +1,65 @@
+use std::env;
+
+#[derive(Clone, Debug)]
+pub struct Config {
+    pub bind_addr: String,
+    pub canonical_origin: String,
+    pub secure_cookies: bool,
+    pub db_path: String,
+    pub data_dir: String,
+    pub frontend_dir: String,
+    pub worker_image: String,
+    pub fetch_timeout_secs: u64,
+    pub backtest_timeout_secs: u64,
+    pub run_concurrency: usize,
+    pub fetch_concurrency: usize,
+    pub session_hours: i64,
+    pub bootstrap_admin_email: Option<String>,
+    pub bootstrap_admin_password: Option<String>,
+    pub ai_base_url: String,
+    pub ai_model: String,
+    pub ai_daily_request_cap: i64,
+    pub ai_input_token_cap: i64,
+    pub ai_output_token_cap: i64,
+    pub ai_enabled_poc: bool,
+    pub default_run_limit_per_day: i64,
+    pub version: String,
+}
+
+fn envs(k: &str, d: &str) -> String {
+    env::var(k).ok().filter(|v| !v.is_empty()).unwrap_or_else(|| d.to_string())
+}
+
+impl Config {
+    pub fn from_env() -> Self {
+        let canonical_origin = envs("ORIGIN", "");
+        Config {
+            bind_addr: envs("BIND", "127.0.0.1:8787"),
+            canonical_origin: canonical_origin.clone(),
+            secure_cookies: canonical_origin.starts_with("https://"),
+            db_path: envs("DB_PATH", "server-data/strategy-lab.sqlite3"),
+            data_dir: envs("DATA_DIR", "server-data/data"),
+            frontend_dir: envs("FRONTEND_DIR", "frontend/dist"),
+            worker_image: envs("WORKER_IMAGE", "strategy-lab-worker:local"),
+            fetch_timeout_secs: envs("FETCH_TIMEOUT_SECS", "1800").parse().unwrap_or(1800),
+            backtest_timeout_secs: envs("BACKTEST_TIMEOUT_SECS", "1800").parse().unwrap_or(1800),
+            run_concurrency: envs("RUN_CONCURRENCY", "1").parse().unwrap_or(1),
+            fetch_concurrency: envs("FETCH_CONCURRENCY", "2").parse().unwrap_or(2),
+            session_hours: envs("SESSION_HOURS", "336").parse().unwrap_or(336),
+            bootstrap_admin_email: env::var("BOOTSTRAP_ADMIN_EMAIL").ok(),
+            bootstrap_admin_password: env::var("BOOTSTRAP_ADMIN_PASSWORD").ok(),
+            ai_base_url: envs("AI_BASE_URL", "https://opencode.ai/zen/go/v1"),
+            ai_model: envs("AI_MODEL", "glm-5.3-flash"),
+            ai_daily_request_cap: envs("AI_DAILY_REQUEST_CAP", "20").parse().unwrap_or(20),
+            ai_input_token_cap: envs("AI_INPUT_TOKEN_CAP", "8000").parse().unwrap_or(8000),
+            ai_output_token_cap: envs("AI_OUTPUT_TOKEN_CAP", "6000").parse().unwrap_or(6000),
+            ai_enabled_poc: envs("AI_ENABLED_POC", "true") == "true",
+            default_run_limit_per_day: envs("DEFAULT_RUN_LIMIT_PER_DAY", "10").parse().unwrap_or(10),
+            version: envs("APP_VERSION", env!("CARGO_PKG_VERSION")),
+        }
+    }
+}
+
+pub const MAX_CODE_LEN: usize = 256_000;
+pub const MAX_SYMBOLS: usize = 5;
+pub const MAX_RANGE_YEARS: i64 = 15;
diff --git a/server/src/datasets.rs b/server/src/datasets.rs
new file mode 100644
index 0000000..5b2428f
--- /dev/null
+++ b/server/src/datasets.rs
@@ -0,0 +1,244 @@
+use axum::{extract::{Path, State}, Json};
+use chrono::NaiveDate;
+use serde::{Deserialize, Serialize};
+use serde_json::{json, Value};
+
+use crate::auth::{audit, AuthUser};
+use crate::config::{MAX_RANGE_YEARS, MAX_SYMBOLS};
+use crate::error::{AppError, AppResult};
+use crate::state::S;
+use crate::util::{canonical_json, new_id, now_iso, sha256_hex};
+
+/// Matches the fixed backend contract `Cx`; wired by State extractor in main.rs.
+pub type Cx = State<S>;
+const FIELDS: [&str; 6] = ["open", "high", "low", "close", "volume", "adj_factor"];
+
+
+#[derive(Deserialize, Serialize, Clone)]
+pub struct InstrumentReq {
+    pub symbol: String,
+    pub market: String,
+    pub asset_type: String,
+    pub name: Option<String>,
+}
+
+#[derive(Deserialize)]
+pub struct DatasetRequest {
+    /// Optional: SPEC/UI permit an absent or blank name; the backend then
+    /// auto-generates a descriptive name and persists it.
+    /// Optional: SPEC/UI permit an absent or blank name; the backend then
+    /// auto-generates a descriptive name and persists it.
+    #[serde(default)]
+    pub name: Option<String>,
+    pub instruments: Vec<InstrumentReq>,
+    pub start_date: String,
+    pub end_date: String,
+    pub frequency: String,
+    pub adjustment: String,
+    pub fields: Vec<String>,
+}
+
+pub fn canonical_request_value(req: &DatasetRequest) -> Value {
+    json!({
+        "instruments": req.instruments.iter().map(|i| json!({
+            "symbol": i.symbol, "market": i.market, "asset_type": i.asset_type,
+            "name": i.name.clone(),
+        })).collect::<Vec<_>>(),
+        "start_date": req.start_date, "end_date": req.end_date,
+        "frequency": req.frequency, "adjustment": req.adjustment, "fields": req.fields,
+    })
+}
+
+pub fn validate_request(req: &DatasetRequest) -> AppResult<String> {
+    if req.instruments.is_empty() || req.instruments.len() > MAX_SYMBOLS {
+        return Err(AppError::bad("validation", format!("instruments must be 1-{} items", MAX_SYMBOLS)));
+    }
+    if req.frequency != "daily" { return Err(AppError::bad("validation", "only daily frequency is supported")); }
+    if !matches!(req.adjustment.as_str(), "none" | "qfq" | "hfq") {
+        return Err(AppError::bad("validation", "adjustment must be none|qfq|hfq"));
+    }
+    if req.fields.is_empty() { return Err(AppError::bad("validation", "fields must not be empty")); }
+    for f in &req.fields {
+        if !FIELDS.contains(&f.as_str()) { return Err(AppError::bad("validation", format!("unsupported field: {f}"))); }
+    }
+    let mut seen = std::collections::HashSet::new();
+    for i in &req.instruments {
+        if i.symbol.trim().is_empty() || i.market.trim().is_empty() {
+            return Err(AppError::bad("validation", "each instrument needs a symbol and a market"));
+        }
+        if !matches!(i.asset_type.as_str(), "stock" | "etf" | "index") {
+            return Err(AppError::bad("validation", "asset_type must be stock|etf|index"));
+        }
+        if i.asset_type == "index" && req.adjustment != "none" {
+            return Err(AppError::bad("validation", "index instruments support adjustment 'none' only (explicit restriction, no factors)"));
+        }
+        if !seen.insert(format!("{}|{}|{}", i.market, i.asset_type, i.symbol)) {
+            return Err(AppError::bad("validation", "duplicate instrument in request"));
+        }
+    }
+    let sd = NaiveDate::parse_from_str(&req.start_date, "%Y-%m-%d").map_err(|_| AppError::bad("validation", "start_date must be YYYY-MM-DD"))?;
+    let ed = NaiveDate::parse_from_str(&req.end_date, "%Y-%m-%d").map_err(|_| AppError::bad("validation", "end_date must be YYYY-MM-DD"))?;
+    if ed < sd { return Err(AppError::bad("validation", "end_date must not precede start_date")); }
+    if (ed - sd).num_days() > MAX_RANGE_YEARS * 366 {
+        return Err(AppError::bad("validation", "range exceeds maximum of 15 years"));
+    }
+    Ok(sha256_hex(canonical_json(&canonical_request_value(req)).as_bytes()))
+}
+
+/// Full stored manifest JSON -> client copy without host/internal paths.
+pub fn client_manifest(m: &Value) -> Value {
+    let mut o = m.clone();
+    o.as_object_mut().map(|m| m.remove("preview"));
+    if let Some(objs) = o.get_mut("objects").and_then(|v| v.as_array_mut()) {
+        for obj in objs.iter_mut() {
+            if let Some(map) = obj.as_object_mut() {
+                map.remove("path");
+            }
+        }
+    }
+    o
+}
+
+fn row_dataset(r: &rusqlite::Row) -> rusqlite::Result<Value> {
+    let manifest: Option<String> = r.get(6)?;
+    let manifest_v: Value = manifest.and_then(|m| serde_json::from_str::<Value>(&m).ok()).unwrap_or(Value::Null);
+    // The stored canonical request is persisted as a JSON string; clients get an object.
+    let request_s: String = r.get(2)?;
+    let request_v: Value = serde_json::from_str::<Value>(&request_s)
+        .map_err(|_| rusqlite::Error::InvalidColumnType(2, "dataset request".into(), rusqlite::types::Type::Text))?;
+    Ok(json!({
+        "id": r.get::<_, String>(0)?,
+        "name": r.get::<_, String>(1)?,
+        "request": request_v,
+        "status": r.get::<_, String>(3)?,
+        "error": r.get::<_, Option<String>>(4)?,
+        "cache_hit": r.get::<_, Option<i64>>(5)?.map(|v| v != 0),
+        "manifest": if manifest_v.is_null() { Value::Null } else { client_manifest(&manifest_v) },
+        "warnings": manifest_v.get("warnings").cloned().unwrap_or(json!([])),
+        "manifest_hash": manifest_v.get("hash").cloned().unwrap_or(Value::Null),
+        "created_at": r.get::<_, String>(7)?,
+        "updated_at": r.get::<_, String>(8)?,
+    }))
+}
+
+pub async fn assert_owned(cx: &Cx, user_id: &str, dataset_id: &str) -> AppResult<()> {
+    let found: Option<String> = cx.with_db(|db| {
+        db.query_row("SELECT user_id FROM datasets WHERE id=?1", [dataset_id], |r| r.get(0)).ok()
+    }).await;
+    match found {
+        Some(o) if o == user_id => Ok(()),
+        _ => Err(AppError::not_found("dataset not found")),
+    }
+}
+
+async fn query_dataset(cx: &Cx, sql: &str, dataset_id: &str) -> AppResult<Value> {
+    cx.with_db(move |db| {
+        db.query_row(sql, [dataset_id], row_dataset)
+            .map_err(|_| AppError::not_found("dataset not found"))
+    }).await
+}
+
+const LIST_SQL: &str = "SELECT id,name,request,status,error,cache_hit,manifest,created_at,updated_at FROM datasets WHERE user_id=?1 ORDER BY created_at DESC LIMIT 200";
+
+pub async fn list(cx: Cx, auth: AuthUser) -> AppResult<Json<Value>> {
+    let items: Vec<Value> = cx.with_db(|db| {
+        let mut st = db.prepare(LIST_SQL)?;
+        let mut rows = st.query([auth.id.clone()])?;
+        let mut out = Vec::new();
+        while let Some(r) = rows.next()? { out.push(row_dataset(r)?); }
+        Ok::<_, AppError>(out)
+    }).await?;
+    Ok(Json(json!({ "items": items })))
+}
+
+pub async fn get(cx: Cx, auth: AuthUser, Path(id): Path<String>) -> AppResult<Json<Value>> {
+    assert_owned(&cx, &auth.id, &id).await?;
+    let v = query_dataset(&cx, "SELECT id,name,request,status,error,cache_hit,manifest,created_at,updated_at FROM datasets WHERE id=?1", &id).await?;
+    Ok(Json(v))
+}
+
+/// Load the stored worker manifest (with internal paths) of a ready dataset.
+pub async fn load_manifest(cx: &Cx, dataset_id: &str) -> AppResult<Value> {
+    let m: String = cx.with_db(|db| {
+        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"))
+    }).await?;
+    serde_json::from_str(&m).map_err(|e| AppError::internal(format!("manifest corrupt: {e}")))
+}
+
+pub async fn create(cx: Cx, auth: AuthUser, body: Option<Json<DatasetRequest>>) -> AppResult<(axum::http::StatusCode, Json<Value>)> {
+    let Json(r) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?;
+    // Frontend sends name: optional/empty; default is an auto-generated descriptive name.
+    let name = if r.name.as_deref().map(|n| n.trim().is_empty()).unwrap_or(true) {
+        // Frontend allows an empty name to auto-generate: instruments + range.
+        let syms: Vec<String> = r.instruments.iter().map(|i| i.symbol.clone()).collect();
+        format!("{} · {} ~ {}", syms.join(","), r.start_date, r.end_date)
+    } else {
+        r.name.as_deref().unwrap_or_default().trim().to_string()
+    };
+    if name.len() > 200 { return Err(AppError::bad("validation", "name required (max 200)")); }
+    let _key = validate_request(&r)?;
+    let stored_request = canonical_request_value(&r);
+    let id = new_id();
+    let ts = now_iso();
+    let uid = auth.id.clone();
+    cx.with_db(|db| -> AppResult<()> {
+        db.execute("INSERT INTO datasets (id,user_id,name,request,status,cache_hit,created_at,updated_at) VALUES (?1,?2,?3,?4,'pending',0,?5,?5)",
+            rusqlite::params![&id, &uid, &name, stored_request.to_string(), &ts])?;
+        Ok(())
+    }).await?;
+    audit(&cx, Some(&auth.id), "dataset_create", &id, "ok").await;
+    let v = query_dataset(&cx, "SELECT id,name,request,status,error,cache_hit,manifest,created_at,updated_at FROM datasets WHERE id=?1", &id).await?;
+    Ok((axum::http::StatusCode::ACCEPTED, Json(v)))
+}
+
+pub async fn preview(cx: Cx, auth: AuthUser, Path(id): Path<String>) -> AppResult<Json<Value>> {
+    assert_owned(&cx, &auth.id, &id).await?;
+    let m = load_manifest(&cx, &id).await?;
+    let p = m.get("preview").cloned().unwrap_or(Value::Null);
+    if p.is_null() { return Err(AppError::not_found("preview not available yet")); }
+    Ok(Json(p))
+}
+
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn req() -> DatasetRequest {
+        serde_json::from_value(json!({
+            "name": "t", "instruments": [{"symbol": "600000", "market": "cn", "asset_type": "stock", "name": "浦发银行"}],
+            "start_date": "2024-01-01", "end_date": "2024-06-30",
+            "frequency": "daily", "adjustment": "none", "fields": ["open","high","low","close","volume"]
+        })).unwrap()
+    }
+
+    #[test]
+    fn validate_rejects_unsupported() {
+        assert_eq!(validate_request(&req()).unwrap().len(), 64);
+        let mut r = req(); r.frequency = "hourly".into();
+        assert_eq!(validate_request(&r).unwrap_err().code, "validation");
+        let mut r = req(); r.adjustment = "qfq".into(); r.instruments[0].asset_type = "index".into();
+        assert_eq!(validate_request(&r).unwrap_err().code, "validation", "index+adjustment must be explicit rejections");
+        let mut r = req(); r.instruments.push(r.instruments[0].clone());
+        assert_eq!(validate_request(&r).unwrap_err().code, "validation", "duplicate instruments rejected");
+    }
+
+    #[test]
+    fn cache_key_is_stable_and_shared_regardless_of_display_name() {
+        let other = { let mut o = req(); o.name = Some("别的名字".to_string()); o };
+        assert_eq!(canonical_request_value(&req()), canonical_request_value(&other));
+        assert_eq!(validate_request(&req()).unwrap(), validate_request(&other).unwrap());
+    }
+
+    #[test]
+    fn client_manifest_strips_internal_paths_and_preview() {
+        let m = json!({
+            "hash": "h", "warnings": [], "preview": {"rows": [1]},
+            "objects": [{"instrument": {"symbol": "SH#600000"}, "path": "objects/ab/ab12.csv"}]
+        });
+        let c = client_manifest(&m);
+        assert!(serde_json::to_string(&c).unwrap().find("objects/ab").is_none(), "host paths must not leak");
+        assert_eq!(c.get("hash"), Some(&json!("h")));
+    }
+}
diff --git a/server/src/db.rs b/server/src/db.rs
new file mode 100644
index 0000000..d8b09dd
--- /dev/null
+++ b/server/src/db.rs
@@ -0,0 +1,146 @@
+use rusqlite::Connection;
+
+pub fn init_db(conn: &Connection) -> rusqlite::Result<()> {
+    conn.execute_batch(
+        r#"
+        PRAGMA journal_mode=WAL;
+        PRAGMA foreign_keys=ON;
+        CREATE TABLE IF NOT EXISTS users (
+            id TEXT PRIMARY KEY,
+            email TEXT NOT NULL UNIQUE,
+            password_hash TEXT NOT NULL,
+            name TEXT NOT NULL,
+            role TEXT NOT NULL CHECK(role IN ('admin','member')),
+            active INTEGER NOT NULL DEFAULT 1,
+            ai_enabled INTEGER NOT NULL DEFAULT 0,
+            daily_run_limit INTEGER NOT NULL DEFAULT 10,
+            created_at TEXT NOT NULL
+        );
+        CREATE TABLE IF NOT EXISTS sessions (
+            id TEXT PRIMARY KEY,           -- sha256(token)
+            user_id TEXT NOT NULL REFERENCES users(id),
+            created_at TEXT NOT NULL,
+            expires_at TEXT NOT NULL,
+            origin_note TEXT
+        );
+        CREATE TABLE IF NOT EXISTS login_failures (
+            email TEXT NOT NULL,
+            failed_at TEXT NOT NULL
+        );
+        CREATE TABLE IF NOT EXISTS invitations (
+            id TEXT PRIMARY KEY,
+            email TEXT,
+            token_hash TEXT NOT NULL UNIQUE,
+            role TEXT NOT NULL DEFAULT 'member',
+            expires_at TEXT NOT NULL,
+            used_by TEXT,
+            created_at TEXT NOT NULL
+        );
+        CREATE TABLE IF NOT EXISTS password_resets (
+            id TEXT PRIMARY KEY,
+            user_id TEXT NOT NULL REFERENCES users(id),
+            token_hash TEXT NOT NULL UNIQUE,
+            expires_at TEXT NOT NULL,
+            used INTEGER NOT NULL DEFAULT 0
+        );
+        CREATE TABLE IF NOT EXISTS audit (
+            seq INTEGER PRIMARY KEY AUTOINCREMENT,
+            ts TEXT NOT NULL,
+            actor_id TEXT,
+            action TEXT NOT NULL,
+            target TEXT,
+            status TEXT NOT NULL,
+            details TEXT
+        );
+        CREATE TABLE IF NOT EXISTS projects (
+            id TEXT PRIMARY KEY,
+            user_id TEXT NOT NULL REFERENCES users(id),
+            name TEXT NOT NULL,
+            description TEXT NOT NULL DEFAULT '',
+            draft_code TEXT NOT NULL DEFAULT '',
+            draft_generation INTEGER NOT NULL DEFAULT 0,
+            created_at TEXT NOT NULL,
+            updated_at TEXT NOT NULL
+        );
+        CREATE TABLE IF NOT EXISTS project_versions (
+            id TEXT PRIMARY KEY,
+            project_id TEXT NOT NULL REFERENCES projects(id),
+            code TEXT NOT NULL,
+            hash TEXT NOT NULL,
+            message TEXT NOT NULL,
+            source TEXT NOT NULL CHECK(source IN ('manual','run','ai','restore')),
+            created_at TEXT NOT NULL
+        );
+        CREATE TABLE IF NOT EXISTS datasets (
+            id TEXT PRIMARY KEY,
+            user_id TEXT NOT NULL REFERENCES users(id),
+            name TEXT NOT NULL,
+            request TEXT NOT NULL,
+            status TEXT NOT NULL CHECK(status IN ('pending','running','ready','failed')),
+            error TEXT,
+            manifest_hash TEXT,
+            manifest TEXT,
+            cache_hit INTEGER,
+            created_at TEXT NOT NULL,
+            updated_at TEXT NOT NULL
+        );
+        CREATE TABLE IF NOT EXISTS data_objects (
+            hash TEXT PRIMARY KEY,
+            path TEXT NOT NULL UNIQUE,
+            size INTEGER NOT NULL,
+            fetched_at TEXT NOT NULL
+        );
+        CREATE TABLE IF NOT EXISTS runs (
+            id TEXT PRIMARY KEY,
+            user_id TEXT NOT NULL REFERENCES users(id),
+            project_id TEXT NOT NULL REFERENCES projects(id),
+            version_id TEXT NOT NULL REFERENCES project_versions(id),
+            dataset_id TEXT NOT NULL REFERENCES datasets(id),
+            status TEXT NOT NULL CHECK(status IN ('queued','running','succeeded','failed','cancelled')),
+            config TEXT NOT NULL,
+            manifest_hash TEXT,
+            container_id TEXT,
+            result TEXT,
+            error TEXT,
+            created_at TEXT NOT NULL,
+            started_at TEXT,
+            finished_at TEXT
+        );
+        CREATE INDEX IF NOT EXISTS runs_project ON runs(project_id, created_at DESC);
+        CREATE INDEX IF NOT EXISTS runs_user_daily ON runs(user_id, created_at DESC);
+        CREATE TABLE IF NOT EXISTS ai_requests (
+            id TEXT PRIMARY KEY,
+            user_id TEXT NOT NULL REFERENCES users(id),
+            project_id TEXT NOT NULL REFERENCES projects(id),
+            instruction TEXT NOT NULL,
+            status TEXT NOT NULL CHECK(status IN ('pending','succeeded','failed')),
+            model TEXT,
+            explanation TEXT,
+            proposed_code TEXT,
+            diff TEXT,
+            base_generation INTEGER,
+            base_code_hash TEXT,
+            version_id TEXT,
+            usage TEXT,
+            error TEXT,
+            created_at TEXT NOT NULL
+        );
+        CREATE TABLE IF NOT EXISTS fetch_cache (
+            key TEXT PRIMARY KEY,
+            manifest_hash TEXT NOT NULL,
+            manifest TEXT NOT NULL,
+            object_count INTEGER NOT NULL,
+            fetched_at TEXT NOT NULL
+        );
+        CREATE TABLE IF NOT EXISTS ai_usage (
+            id TEXT PRIMARY KEY,
+            user_id TEXT NOT NULL,
+            request_id TEXT,
+            ts TEXT NOT NULL,
+            kind TEXT NOT NULL,
+            input_tokens INTEGER DEFAULT 0,
+            output_tokens INTEGER DEFAULT 0
+        );
+        "#,
+    )
+}
diff --git a/server/src/error.rs b/server/src/error.rs
new file mode 100644
index 0000000..0342afd
--- /dev/null
+++ b/server/src/error.rs
@@ -0,0 +1,145 @@
+use axum::{
+    http::StatusCode,
+    response::{IntoResponse, Response},
+    Json,
+};
+use serde_json::{json, Value};
+
+/// Coherent error type for all handlers and helpers. Exactly one set of
+/// constructors, one IntoResponse, and the three owned From conversions
+/// (rusqlite, std::io, serde_json) live here and nowhere else.
+#[derive(Debug)]
+pub struct AppError {
+    pub status: StatusCode,
+    pub code: &'static str,
+    pub message: String,
+    pub details: Option<Value>,
+}
+
+impl std::fmt::Display for AppError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}: {}", self.code, self.message)
+    }
+}
+
+impl std::error::Error for AppError {}
+
+impl AppError {
+    pub fn new(status: StatusCode, code: &'static str, message: impl Into<String>) -> Self {
+        AppError { status, code, message: message.into(), details: None }
+    }
+    pub fn bad(code: &'static str, msg: impl Into<String>) -> Self {
+        Self::new(StatusCode::BAD_REQUEST, code, msg)
+    }
+    pub fn unauthorized(msg: impl Into<String>) -> Self {
+        Self::new(StatusCode::UNAUTHORIZED, "unauthorized", msg)
+    }
+    pub fn forbidden(msg: impl Into<String>) -> Self {
+        Self::new(StatusCode::FORBIDDEN, "forbidden", msg)
+    }
+    pub fn not_found(msg: impl Into<String>) -> Self {
+        Self::new(StatusCode::NOT_FOUND, "not_found", msg)
+    }
+    pub fn conflict(code: &'static str, msg: impl Into<String>) -> Self {
+        Self::new(StatusCode::CONFLICT, code, msg)
+    }
+    pub fn internal(msg: impl Into<String>) -> Self {
+        Self::new(StatusCode::INTERNAL_SERVER_ERROR, "internal", msg)
+    }
+    pub fn with_details(mut self, d: Value) -> Self {
+        self.details = Some(d);
+        self
+    }
+    pub fn with_code(mut self, code: &'static str) -> Self {
+        self.code = code;
+        self
+    }
+}
+
+impl IntoResponse for AppError {
+    fn into_response(self) -> Response {
+        let mut e = json!({"code": self.code, "message": self.message});
+        if let Some(details) = self.details {
+            e["details"] = details;
+        }
+        (self.status, Json(json!({"error": e}))).into_response()
+    }
+}
+
+impl From<rusqlite::Error> for AppError {
+    fn from(e: rusqlite::Error) -> Self {
+        match &e {
+            rusqlite::Error::QueryReturnedNoRows => AppError::not_found("resource not found"),
+            _ => {
+                tracing::error!("db error: {e}");
+                AppError::internal("internal storage error")
+            }
+        }
+    }
+}
+
+impl From<std::io::Error> for AppError {
+    fn from(e: std::io::Error) -> Self {
+        tracing::error!("io error: {e}");
+        AppError::internal("internal filesystem error")
+    }
+}
+
+impl From<serde_json::Error> for AppError {
+    fn from(e: serde_json::Error) -> Self {
+        tracing::error!("json error: {e}");
+        AppError::internal("internal json error")
+    }
+}
+
+pub type AppResult<T> = Result<T, AppError>;
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn constructors_map_to_expected_status_and_code() {
+        let e = AppError::bad("validation", "x");
+        assert_eq!(e.status, StatusCode::BAD_REQUEST);
+        assert_eq!(e.code, "validation");
+        assert_eq!(AppError::unauthorized("y").status, StatusCode::UNAUTHORIZED);
+        assert_eq!(AppError::forbidden("y").status, StatusCode::FORBIDDEN);
+        assert_eq!(AppError::not_found("y").status, StatusCode::NOT_FOUND);
+        assert_eq!(AppError::conflict("stale", "y").status, StatusCode::CONFLICT);
+        assert_eq!(AppError::internal("y").status, StatusCode::INTERNAL_SERVER_ERROR);
+        let v = AppError::conflict("stale", "y")
+            .with_code("dataset_not_ready")
+            .with_details(json!({"warnings": [1]}));
+        assert_eq!(v.code, "dataset_not_ready");
+        assert_eq!(v.details.expect("details")["warnings"], json!([1]));
+    }
+
+    #[tokio::test]
+    async fn body_shape_is_error_code_message_details() {
+        let mut resp = AppError::conflict("stale", "expected_generation out of date")
+            .with_details(json!({"expected": 3}))
+            .into_response();
+        let bytes = axum::body::to_bytes(std::mem::take(resp.body_mut()), 64 * 1024).await.unwrap();
+        let parsed: Value = serde_json::from_slice(&bytes).unwrap();
+        assert_eq!(parsed["error"]["code"], "stale");
+        assert_eq!(parsed["error"]["message"], "expected_generation out of date");
+        assert_eq!(parsed["error"]["details"]["expected"], 3);
+
+        let mut plain = AppError::internal("boom").into_response();
+        let plain: Value = serde_json::from_slice(
+            &axum::body::to_bytes(std::mem::take(plain.body_mut()), 64 * 1024).await.unwrap(),
+        ).unwrap();
+        assert!(plain["error"].get("details").is_none());
+    }
+
+    #[test]
+    fn from_impls_convert_without_duplicates() {
+        let e: AppError = rusqlite::Error::QueryReturnedNoRows.into();
+        assert_eq!(e.status, StatusCode::NOT_FOUND);
+        let e: AppError = std::io::Error::new(std::io::ErrorKind::NotFound, "nope").into();
+        assert_eq!(e.status, StatusCode::INTERNAL_SERVER_ERROR);
+        let e: AppError = serde_json::from_str::<Value>("{").unwrap_err().into();
+        assert_eq!(e.status, StatusCode::INTERNAL_SERVER_ERROR);
+    }
+}
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");
+    }
+}
diff --git a/server/src/main.rs b/server/src/main.rs
new file mode 100644
index 0000000..87f5467
--- /dev/null
+++ b/server/src/main.rs
@@ -0,0 +1,667 @@
+mod admin;
+mod ai;
+mod auth;
+mod config;
+mod db;
+mod datasets;
+mod error;
+mod jobs;
+mod projects;
+mod runs;
+mod state;
+mod store;
+mod util;
+mod worker;
+
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::sync::Arc;
+
+use axum::{
+    body::Body,
+    extract::Request,
+    http::{header, HeaderValue, Method, StatusCode},
+    middleware::{self, Next},
+    response::{IntoResponse, Response},
+    routing::{delete, get, patch, post, put},
+    Json, Router,
+};
+use tokio::sync::Mutex as AsyncMutex;
+
+use crate::state::{AppState, Cx};
+
+const MIME_FALLBACK: &str = "application/octet-stream";
+
+/// CSRF / content protections for writes.
+async fn csrf_middleware(req: Request<Body>, next: Next) -> Response {
+    if !is_write(req.method()) {
+        return next.run(req).await;
+    }
+    let headers = req.headers().clone();
+    if let Some(origin) = headers.get(header::ORIGIN).and_then(|v| v.to_str().ok()) {
+        let host = headers.get(header::HOST).and_then(|v| v.to_str().ok()).unwrap_or("");
+        // No trust of X-Forwarded-* hosts; only exact canonical/host origin matches.
+        if !origin_allowed(origin, host, canonical_origin()) {
+            return AppErr::forbidden("cross-origin write rejected").into_response();
+        }
+    }
+    if let Some(site) = headers.get("sec-fetch-site").and_then(|v| v.to_str().ok()) {
+        if site == "cross-site" {
+            return AppErr::forbidden("cross-site request rejected").into_response();
+        }
+    }
+    // JSON writes only. DELETE carries no body: allow an absent content type.
+    let ctype = headers
+        .get(header::CONTENT_TYPE)
+        .and_then(|v| v.to_str().ok())
+        .unwrap_or("")
+        .to_string();
+    let missing_ok = req.method() == Method::DELETE && ctype.is_empty();
+    if !(missing_ok || ctype.starts_with("application/json")) {
+        return AppErr::forbidden("JSON Content-Type required").into_response();
+    }
+    next.run(req).await
+}
+
+static CANONICAL_ORIGIN: std::sync::OnceLock<String> = std::sync::OnceLock::new();
+
+fn canonical_origin() -> &'static str {
+    CANONICAL_ORIGIN.get().map(|s| s.as_str()).unwrap_or("")
+}
+
+/// Exact-origin check. Substring matching is exploitable:
+/// an attacker-supplied Origin `https://allowed.example.evil.invalid` must fail.
+/// Empty canonical origin: exact local (Host-based) origin only.
+pub fn origin_allowed(origin: &str, host: &str, canonical: &str) -> bool {
+    let origin = origin.trim();
+    if !(origin.starts_with("http://") || origin.starts_with("https://")) {
+        return false;
+    }
+    if !canonical.is_empty() {
+        // Trust only the configured canonical https origin, compared exactly.
+        return origin.eq_ignore_ascii_case(canonical.trim());
+    }
+    if host.is_empty() {
+        // No Host and no canonical: cannot establish trust; reject.
+        return false;
+    }
+    origin == format!("http://{host}") || origin == format!("https://{host}")
+}
+
+fn set_canonical(origin: &str) {
+    let _ = CANONICAL_ORIGIN.set(origin.trim().to_string());
+}
+
+fn is_write(m: &Method) -> bool {
+    matches!(*m, Method::POST | Method::PUT | Method::PATCH | Method::DELETE)
+}
+
+type AppErr = error::AppError;
+
+async fn health(cx: Cx) -> Json<serde_json::Value> {
+    let docker = worker::docker_available().await;
+    Json(serde_json::json!({
+        "status": "ok",
+        "version": cx.cfg.version,
+        "worker_available": docker,
+        "ai_configured": std::env::var("OPENCODE_GO_API_KEY").map(|_| true).unwrap_or(false),
+    }))
+}
+
+async fn capabilities() -> Json<serde_json::Value> {
+    Json(serde_json::json!({
+        "frequencies": ["daily"],
+        "asset_types": ["stock", "etf", "index"],
+        "adjustments": [
+            {"code": "none", "label": "不复权"},
+            {"code": "qfq", "label": "前复权"},
+            {"code": "hfq", "label": "后复权"}
+        ],
+        "fields": [
+            {"code": "open", "label": "开盘", "raw": false},
+            {"code": "high", "label": "最高", "raw": false},
+            {"code": "low", "label": "最低", "raw": false},
+            {"code": "close", "label": "收盘", "raw": false},
+            {"code": "volume", "label": "成交量", "raw": false},
+            {"code": "adj_factor", "label": "复权因子", "raw": true}
+        ],
+        "limits": {"max_symbols": 5, "max_years": 15, "internal_only": true},
+    }))
+}
+
+/// Instrument search through the real worker container catalog command.
+/// Failures are surfaced honestly in `status`; no fake empty success.
+/// NOTE: Query deserialization: a plain `HashMap` accepts both absent and
+/// present query params. `Query<Option<...>>` rejects any non-empty query with
+/// `invalid type: map, expected option` (HTTP400 observed in live browser QA).
+async fn instruments(
+    cx: Cx,
+    q: axum::extract::Query<HashMap<String, String>>,
+) -> Json<serde_json::Value> {
+    let qm = q.0;
+    let query = qm.get("q").cloned().unwrap_or_default();
+    let limit = qm
+        .get("limit")
+        .and_then(|v| v.parse::<i64>().ok())
+        .unwrap_or(50);
+    match worker::search_instruments(&cx, &query, limit).await {
+    Ok(items) => {
+        let source = if items.is_empty() { "none" } else { "provider_suggest" };
+        Json(serde_json::json!({"items": items, "source": source, "status": "ok"}))
+    }
+        Err(e) => Json(serde_json::json!({
+            "items": [],
+            "source": "none",
+            "status": format!("unavailable: {}", e.message)
+        })),
+    }
+}
+
+async fn api_fallback() -> Response {
+    (
+        StatusCode::NOT_FOUND,
+        Json(serde_json::json!({"error": {"code": "not_found", "message": "unknown API route"}})),
+    )
+        .into_response()
+}
+
+fn mime_of(path: &std::path::Path) -> &'static str {
+    match path.extension().and_then(|e| e.to_str()).unwrap_or("") {
+        "html" => "text/html; charset=utf-8",
+        "css" => "text/css; charset=utf-8",
+        "js" | "mjs" => "text/javascript; charset=utf-8",
+        "json" => "application/json",
+        "svg" => "image/svg+xml",
+        "png" => "image/png",
+        "webp" => "image/webp",
+        "woff2" => "font/woff2",
+        "woff" => "font/woff",
+        "ico" => "image/x-icon",
+        "map" => "application/json",
+        "txt" => "text/plain; charset=utf-8",
+        _ => MIME_FALLBACK,
+    }
+}
+
+/// Serve the built SPA from frontend/dist. Exact files when they exist,
+/// otherwise /index.html so client routes work; unknown /api is handled by the
+/// inner fallback above and never falls back to the SPA.
+
+/// Static file resolution for the SPA (GET/HEAD only).
+async fn serve_static(root: PathBuf, path: &str) -> Response {
+    let rel = path.trim_start_matches('/');
+    if rel.contains("..") || rel.contains('\\') {
+        return StatusCode::NOT_FOUND.into_response();
+    }
+    let base = root.join("index.html");
+    let target = if rel.is_empty() {
+        base
+    } else {
+        let p = root.join(rel);
+        if p.is_file() {
+            p
+        } else if p.is_dir() || !p.exists() {
+            base
+        } else {
+            return StatusCode::NOT_FOUND.into_response();
+        }
+    };
+    match tokio::fs::read(&target).await {
+        Ok(bytes) => {
+            let mut resp = (
+                StatusCode::OK,
+                [(header::CONTENT_TYPE, mime_of(&target))],
+                bytes,
+            ).into_response();
+            if mime_of(&target) != "text/html" {
+                resp.headers_mut().insert(
+                    header::CACHE_CONTROL,
+                    HeaderValue::from_static("no-cache"),
+                );
+            }
+            resp
+        }
+        Err(_) => StatusCode::NOT_FOUND.into_response(),
+    }
+}
+
+pub fn build_app(cx: Arc<AppState>, frontend_dir: String) -> Router {
+    let api = Router::new()
+        .route("/health", get(health))
+        .route("/capabilities", get(capabilities))
+        .route("/instruments", get(instruments))
+        .route("/auth/login", post(auth::login))
+        .route("/auth/register", post(auth::register))
+        .route("/auth/logout", post(auth::logout))
+        .route("/auth/me", get(auth::me))
+        .route("/auth/profile", patch(auth::patch_profile))
+        .route("/auth/password", post(auth::change_password))
+        .route("/auth/sessions", get(auth::list_sessions))
+        .route("/auth/sessions/{id}", delete(auth::delete_session))
+        .route("/auth/reset-password", post(auth::reset_password))
+        .route("/projects", get(projects::list).post(projects::create))
+        .route("/projects/{id}", get(projects::get).patch(projects::patch))
+        .route("/projects/{id}/draft", put(projects::put_draft))
+        .route("/projects/{id}/versions", get(projects::list_versions).post(projects::create_version))
+        .route("/projects/{id}/versions/{vid}", get(projects::get_version))
+        .route("/projects/{id}/versions/{vid}/diff", post(projects::diff_versions))
+        .route("/projects/{id}/restore", post(projects::restore))
+        .route("/datasets", get(datasets::list).post(datasets::create))
+        .route("/datasets/{id}", get(datasets::get))
+        .route("/datasets/{id}/preview", get(datasets::preview))
+        .route("/runs", get(runs::list).post(runs::enqueue))
+        .route("/runs/{id}", get(runs::get))
+        .route("/runs/{id}/cancel", post(runs::cancel))
+        .route("/runs/{id}/rerun", post(runs::rerun))
+        .route("/ai/assist", post(ai::assist))
+        .route("/ai/usage", get(ai::list_ai_usage))
+        .route("/ai/{id}/accept", post(ai::accept))
+        .route("/admin/users", get(admin::users))
+        .route("/admin/users/{id}", patch(admin::patch_user))
+        .route("/admin/invitations", get(admin::list_invitations).post(admin::create_invitation))
+        .route("/admin/invitations/{id}", delete(admin::delete_invitation))
+        .route("/admin/users/{id}/reset-password", post(admin::create_reset))
+        .route("/admin/audit", get(admin::audit_list))
+        .fallback(api_fallback)
+        .layer(middleware::from_fn(csrf_middleware))
+        .with_state(cx.clone());
+
+    let spa = move |req: Request<Body>| {
+        let root = frontend_dir.clone();
+        async move {
+            let method = req.method().clone();
+            let path = req.uri().path().to_string();
+            if !matches!(method, Method::GET | Method::HEAD) {
+                return StatusCode::METHOD_NOT_ALLOWED.into_response();
+            }
+            serve_static(PathBuf::from(root), &path).await
+        }
+    };
+
+    Router::new().nest("/api", api).fallback(spa)
+}
+
+#[cfg(test)]
+mod probe {
+    use super::*;
+
+    fn probe_static(root: &str, path: &str) -> Response {
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        rt.block_on(serve_static(PathBuf::from(root), path))
+    }
+
+    #[axum::debug_handler(state = Arc<AppState>)]
+    async fn probe_auth(
+        _cx: Cx,
+        _auth: auth::AuthUser,
+        _path: axum::extract::Path<String>,
+    ) -> Result<Json<serde_json::Value>, error::AppError> {
+        Ok(Json(serde_json::json!({})))
+    }
+
+    #[axum::debug_handler(state = Arc<AppState>)]
+    async fn probe_query(
+        _cx: Cx,
+        _q: axum::extract::Query<HashMap<String, String>>,
+    ) -> Json<serde_json::Value> {
+        Json(serde_json::json!({}))
+    }
+
+    #[test]
+    fn probe_router() {
+        let state = Arc::new(AppState {
+            cfg: config::Config::from_env(),
+            db: AsyncMutex::new(rusqlite::Connection::open_in_memory().unwrap()),
+            run_sem: Arc::new(tokio::sync::Semaphore::new(1)),
+            fetch_sem: Arc::new(tokio::sync::Semaphore::new(1)),
+        });
+        let _r: Router = Router::new()
+            .route("/x", get(probe_auth))
+            .route("/y", get(probe_query))
+            .with_state(state);
+    }
+
+    /// Exact-origin CSRF: substring attacks must be rejected.
+    #[test]
+    fn csrf_rejects_malicious_substring_origin() {
+        const CANON: &str = "https://fin.somhairle.bid";
+        assert!(origin_allowed(CANON, "fin.somhairle.bid", CANON));
+        // attacker-controlled suffix
+        assert!(!origin_allowed("https://fin.somhairle.bid.evil.invalid", "fin.somhairle.bid", CANON));
+        // attacker-controlled prefix host
+        assert!(!origin_allowed("https://evil.fin.somhairle.bid", "fin.somhairle.bid", CANON));
+        // different scheme
+        assert!(!origin_allowed("http://fin.somhairle.bid", "fin.somhairle.bid", CANON));
+        // different port
+        assert!(!origin_allowed("https://fin.somhairle.bid:8443", "fin.somhairle.bid", CANON));
+        // no canonical: exact local origin only
+        assert!(origin_allowed("http://127.0.0.1:8787", "127.0.0.1:8787", ""));
+        assert!(!origin_allowed("http://127.0.0.1:8787.evil.invalid", "127.0.0.1:8787", ""));
+        assert!(!origin_allowed("http://127.0.0.1:8787x", "127.0.0.1:8787", ""));
+        // no host, no canonical: reject
+        assert!(!origin_allowed("http://whatever", "", ""));
+        assert!(!origin_allowed("javascript:alert(1)", "127.0.0.1:8787", ""));
+        assert!(!origin_allowed("", "127.0.0.1:8787", ""));
+    }
+
+    #[test]
+    fn spa_serves_index_for_unknown_paths_and_sanitizes_traversal() {
+        let td = tempfile::tempdir_in("/tmp/opencode").unwrap();
+        std::fs::write(td.path().join("index.html"), b"<html>ok</html>").unwrap();
+        std::fs::write(td.path().join("assets.js"), b"console.log(1)").unwrap();
+        let resp = probe_static(td.path().to_str().unwrap(), "/");
+        assert_eq!(resp.status(), 200);
+        assert!(resp.headers().get(header::CONTENT_TYPE).unwrap().to_str().unwrap().starts_with("text/html"));
+        let resp = probe_static(td.path().to_str().unwrap(), "/assets.js");
+        assert_eq!(resp.status(), 200);
+        let resp = probe_static(td.path().to_str().unwrap(), "/unknown/route");
+        // SPA fallback serves index.html for client routes
+        assert_eq!(resp.status(), 200);
+        let resp = probe_static(td.path().to_str().unwrap(), "/../../etc/passwd");
+        assert_ne!(resp.status(), 200);
+    }
+
+    /// Executable-level regression: boot the real binary on an isolated
+    /// DB/port, assert it stays alive (>10s) WITHOUT any shutdown signal, then
+    /// SIGTERM must produce a bounded timely exit. This formerly caught a bug
+    /// where the drain deadline incorrectly started at startup and killed the
+    /// server at ~10s of healthy uptime (premature-exit regression).
+    #[tokio::test]
+    async fn server_survives_past_10s_then_bounds_sigterm_exit() {
+        let td = tempfile::tempdir_in("/tmp/opencode").unwrap();
+        let db_path = td.path().join("db.sqlite3");
+        let data_dir = td.path().join("data");
+        std::fs::create_dir_all(&data_dir).unwrap();
+        // reserve a port using the OS
+        let probe_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let port = probe_listener.local_addr().unwrap().port();
+        std::mem::drop(probe_listener);
+
+        // a known-empty frontend dir keeps static serving honest for this probe
+        let fe = td.path().join("fe");
+        std::fs::create_dir_all(&fe).unwrap();
+
+        let bin = std::env::var("CARGO_BIN_EXE_strategy-lab-server")
+            .unwrap_or_else(|_| "target/debug/strategy-lab-server".to_string());
+        let bin_path = std::path::PathBuf::from(bin);
+        assert!(bin_path.is_file(), "isolated server binary not built at {}", bin_path.display());
+        let mut child = tokio::process::Command::new(bin_path)
+            .env("BIND", format!("127.0.0.1:{port}"))
+            .env("DB_PATH", db_path.display().to_string())
+            .env("DATA_DIR", data_dir.display().to_string())
+            .env("FRONTEND_DIR", fe.display().to_string())
+            .env("WORKER_IMAGE", "strategy-lab-worker-unset")
+            .stdout(std::process::Stdio::null())
+            .stderr(std::process::Stdio::null())
+            .spawn()
+            .unwrap();
+
+        async fn health_ok(port: u16) -> bool {
+            let url = format!("http://127.0.0.1:{port}/api/health");
+            match reqwest::get(&url).await {
+                Ok(r) => r.status().is_success(),
+                Err(_) => false,
+            }
+        }
+        // server must come up
+        let mut up = false;
+        for _ in 0..100 {
+            if health_ok(port).await { up = true; break; }
+            tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+        }
+        assert!(up, "server did not come up in 20s");
+        // CRITICAL: still alive WELL PAST 10 seconds with NO shutdown signal
+        tokio::time::sleep(std::time::Duration::from_secs(12)).await;
+        assert!(health_ok(port).await, "premature-exit regression: server died ~10s after startup without any signal");
+        // now signal and require bounded exit
+        let pid = child.id().unwrap();
+        nix_pid_kill_term(pid);
+        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(20);
+        let mut exited = false;
+        while std::time::Instant::now() < deadline {
+            if child.try_wait().unwrap().is_some() {
+                exited = true;
+                break;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+        }
+        assert!(exited, "server did not exit in bounded time after SIGTERM");
+        // give the runtime a moment to reap
+        let _ = child.wait().await;
+        let pid_gone = !std::path::Path::new(format!("/proc/{pid}").as_str()).exists();
+        assert!(pid_gone, "server process still alive after bounded exit check");
+    }
+
+    fn nix_pid_kill_term(pid: u32) {
+        // POSIX kill of the exact PID only; no process-name scans, no group ops
+        std::process::Command::new("kill").args(["-TERM", &pid.to_string()])
+            .stdout(std::process::Stdio::null()).stderr(std::process::Stdio::null())
+            .status().ok();
+    }
+
+    /// Route-level regression: `GET /api/instruments?q=600000` formerly failed
+    /// with HTTP400 "invalid type: map, expected option" because the query
+    /// extractor was `Query<Option<...>>`. Must be JSON 200 with `items`,
+    /// `source`, `status` (parent browser workflow live failure).
+    #[tokio::test]
+    async fn instruments_query_with_q_is_json_200() {
+        let conn = rusqlite::Connection::open_in_memory().unwrap();
+        crate::db::init_db(&conn).unwrap();
+        let mut cfg = crate::config::Config {
+            db_path: format!("{}/nonexistent.sqlite3", std::env::temp_dir().display()),
+            data_dir: std::env::temp_dir().display().to_string(),
+            worker_image: "strategy-lab-worker-unset".into(),
+            ..config::Config::from_env()
+        };
+        cfg.bind_addr = "127.0.0.1:0".into();
+        let st = std::sync::Arc::new(AppState {
+            cfg,
+            db: AsyncMutex::new(conn),
+            run_sem: Arc::new(tokio::sync::Semaphore::new(1)),
+            fetch_sem: Arc::new(tokio::sync::Semaphore::new(1)),
+        });
+        let app = build_app(st, "frontend-dist-missing-for-probe".to_string());
+        let req = axum::http::Request::builder()
+            .method("GET")
+            .uri("/api/instruments?q=600000")
+            .body(axum::body::Body::empty())
+            .unwrap();
+        let resp = tower::ServiceExt::oneshot(app, req).await.unwrap();
+        assert_eq!(resp.status(), StatusCode::OK,
+            "q= query must deserialize (was HTTP400 invalid type: map, expected option)");
+        let body = axum::body::to_bytes(resp.into_body(), 64_000).await.unwrap();
+        let v: serde_json::Value = serde_json::from_slice(&body).unwrap();
+        assert!(v.is_object(), "structured JSON error/shape, never plain text: {v}");
+        assert!(v.get("items").and_then(|i| i.as_array()).is_some(), "items array present");
+        assert!(v.get("source").and_then(|s| s.as_str()).is_some(), "source present");
+        // absent q must not 400 either
+        let app2 = {
+            let conn = rusqlite::Connection::open_in_memory().unwrap();
+            crate::db::init_db(&conn).unwrap();
+            let mut cfg = crate::config::Config::from_env();
+            cfg.db_path = format!("{}/x.sqlite3", std::env::temp_dir().display());
+            cfg.data_dir = std::env::temp_dir().display().to_string();
+            cfg.bind_addr = "127.0.0.1:0".into();
+            std::sync::Arc::new(AppState {
+                cfg,
+                db: AsyncMutex::new(conn),
+                run_sem: Arc::new(tokio::sync::Semaphore::new(1)),
+                fetch_sem: Arc::new(tokio::sync::Semaphore::new(1)),
+            })
+        };
+        let app2 = build_app(app2, "missing-frontend-probe".to_string());
+        let req = axum::http::Request::builder().method("GET").uri("/api/instruments?")
+            .body(axum::body::Body::empty()).unwrap();
+        let resp = tower::ServiceExt::oneshot(app2, req).await.unwrap();
+        assert_eq!(resp.status(), StatusCode::OK);
+    }
+
+    /// Dataset creation shape: absent `name` and blank `name` must both
+    /// deserialize (SPEC/UI permit; backend auto-generates and persists).
+    #[test]
+    fn dataset_request_allows_missing_or_blank_name() {
+        let base = serde_json::json!({
+            "instruments": [{"symbol": "600000", "market": "SH", "asset_type": "stock"}],
+            "start_date": "2024-01-01", "end_date": "2024-06-30",
+            "frequency": "daily", "adjustment": "none",
+            "fields": ["open", "high", "low", "close", "volume"]
+        });
+        // missing `name` entirely (previously HTTP422 missing field `name`)
+        let r: datasets::DatasetRequest = serde_json::from_value(base.clone()).unwrap();
+        assert!(r.name.is_none(), "absent name must deserialize as None");
+        // blank name
+        let mut with_blank = base.clone();
+        with_blank["name"] = serde_json::json!("   ");
+        let r = serde_json::from_value::<datasets::DatasetRequest>(with_blank).unwrap();
+        assert!(r.name.unwrap().trim().is_empty());
+        // normal name still works
+        let mut with_name = base;
+        with_name["name"] = serde_json::json!("真实行情验收：浦发银行");
+        let r = serde_json::from_value::<datasets::DatasetRequest>(with_name).unwrap();
+        assert_eq!(r.name.as_deref(), Some("真实行情验收：浦发银行"));
+    }
+}
+
+fn make_state(cfg: config::Config) -> Arc<AppState> {
+    if let Some(parent) = std::path::Path::new(&cfg.db_path).parent() {
+        std::fs::create_dir_all(parent).expect("create db directory");
+    }
+    let conn = rusqlite::Connection::open(&cfg.db_path).expect("open db");
+    db::init_db(&conn).expect("init schema");
+    Arc::new(AppState {
+        run_sem: Arc::new(tokio::sync::Semaphore::new(cfg.run_concurrency)),
+        fetch_sem: Arc::new(tokio::sync::Semaphore::new(cfg.fetch_concurrency)),
+        cfg: cfg.clone(),
+        db: AsyncMutex::new(conn),
+    })
+}
+
+/// Create required runtime directories (databases, object store, job dirs).
+fn ensure_directories(cfg: &config::Config) {
+    let data = std::path::Path::new(&cfg.data_dir);
+    for d in [data.join("objects"), data.join("jobs")] {
+        if let Err(e) = std::fs::create_dir_all(&d) {
+            tracing::error!("cannot create {d:?}: {e}");
+            std::process::exit(1);
+        }
+    }
+}
+
+/// Remove leftover containers from any earlier (crashed) server instance.
+async fn cleanup_orphan_containers(cx: &Arc<AppState>) {
+    let names: Vec<String> = cx
+        .with_db(|db| {
+            let mut st = db.prepare("SELECT container_id FROM runs WHERE container_id IS NOT NULL")?;
+            let rows = st.query_map([], |r| r.get::<_, Option<String>>(0))?;
+            let t: Vec<Option<String>> =
+                rows.collect::<Result<Vec<Option<String>>, rusqlite::Error>>()?;
+            let out: Vec<String> = t
+                .into_iter()
+                .flatten()
+                .filter(|s| !s.is_empty())
+                .collect();
+            Ok::<Vec<String>, rusqlite::Error>(out)
+        })
+        .await
+        .unwrap_or_default();
+    let mut removed = 0usize;
+    for name in names {
+        if worker::cancel_container(&name).await {
+            removed += 1;
+        }
+    }
+    if removed > 0 {
+        tracing::info!("cleaned {removed} leftover worker container(s)");
+    }
+}
+
+#[tokio::main]
+async fn main() {
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::try_from_default_env()
+                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
+        )
+        .init();
+    let cfg = config::Config::from_env();
+    set_canonical(&cfg.canonical_origin);
+    let mode = std::env::args().nth(1).unwrap_or_else(|| "serve".into());
+
+    if mode == "bootstrap-admin" {
+        let state = make_state(cfg.clone());
+        ensure_directories(&cfg);
+        let cx = axum::extract::State(state.clone());
+        let _ = admin::bootstrap_admin(&cx)
+            .await
+            .map_err(|e| tracing::error!("bootstrap-admin failed: {e}"));
+        println!("bootstrap-admin done (if ADMIN_BOOTSTRAP env configured)");
+        return;
+    }
+
+    ensure_directories(&cfg);
+    let state = make_state(cfg.clone());
+    let cx = axum::extract::State(state.clone());
+    match admin::bootstrap_admin(&cx).await {
+        Ok(_) => {}
+        Err(e) => tracing::error!("bootstrap_admin: {e}"),
+    }
+
+    // Restart safety: runs stuck as running become failed; stale containers are
+    // removed by name (never a global docker prune).
+    runs::cleanup_interrupted(&cx).await;
+    cleanup_orphan_containers(&state).await;
+
+    let app = build_app(state.clone(), cfg.frontend_dir.clone());
+
+    tokio::spawn(async move {
+        jobs::main_loop(state, jobs::Signals::new()).await;
+    });
+
+    let addr = cfg.bind_addr.clone();
+    let listener = tokio::net::TcpListener::bind(&addr).await.expect("bind");
+    tracing::info!("listening on {addr}");
+    // NO startup deadline. The drain budget starts only after the shutdown
+    // signal: select between
+    //   (a) the serve future completing normally / after graceful shutdown, and
+    //   (b) signal-received AFTER which a 10s post-signal ceiling passes —
+    // the (b) arm cannot fire before the signal is delivered, so a healthy
+    // server with no signal stays up indefinitely.
+    tokio::select! {
+        drained = axum::serve(listener, app.into_make_service())
+            .with_graceful_shutdown(wait_shutdown_signal()) =>
+        {
+            match drained {
+                Ok(()) => tracing::info!("http drained, exiting"),
+                Err(e) => tracing::error!("http serve failed: {e}"),
+            }
+        }
+        _ = async {
+            wait_shutdown_signal().await;
+            tracing::info!("shutdown signal received; http draining (<=10s)");
+            tokio::time::sleep(std::time::Duration::from_secs(10)).await;
+        } => {
+            tracing::warn!("post-signal drain budget (10s) exceeded; exiting now");
+        }
+    }
+}
+
+/// Waits for SIGTERM or SIGINT (whichever arrives first).
+async fn wait_shutdown_signal() {
+    use tokio::signal::unix::{signal, SignalKind};
+    let term_fut = async {
+        match signal(SignalKind::terminate()) {
+            Ok(mut s) => { s.recv().await; }
+            Err(_) => std::future::pending::<()>().await,
+        }
+    };
+    let int_fut = async {
+        match signal(SignalKind::interrupt()) {
+            Ok(mut s) => { s.recv().await; }
+            Err(_) => std::future::pending::<()>().await,
+        }
+    };
+    tokio::select! { _ = term_fut => {}, _ = int_fut => {} }
+}
diff --git a/server/src/projects.rs b/server/src/projects.rs
new file mode 100644
index 0000000..f5305da
--- /dev/null
+++ b/server/src/projects.rs
@@ -0,0 +1,219 @@
+use axum::extract::Path;
+use axum::Json;
+use serde::{Deserialize, Serialize};
+use serde_json::{json, Value};
+
+use crate::auth::{audit, AuthUser};
+use crate::error::{AppError, AppResult};
+use crate::util::{new_id, now_iso, sha256_hex, unified_diff};
+
+pub type Cx = crate::state::Cx;
+
+#[derive(Serialize)]
+pub struct Project {
+    pub id: String,
+    pub name: String,
+    pub description: String,
+    pub draft_code: String,
+    pub draft_generation: i64,
+    pub created_at: String,
+    pub updated_at: String,
+}
+
+pub async fn load_owned(cx: &Cx, user_id: &str, project_id: &str) -> AppResult<Project> {
+    cx.with_db(|db| {
+        db.query_row("SELECT id,name,description,draft_code,draft_generation,created_at,updated_at FROM projects WHERE id=?1 AND user_id=?2",
+            rusqlite::params![project_id, user_id], row_project)
+            .map_err(|_| AppError::not_found("project not found"))
+    }).await
+}
+
+fn row_project(r: &rusqlite::Row) -> rusqlite::Result<Project> {
+    Ok(Project {
+        id: r.get(0)?, name: r.get(1)?, description: r.get(2)?,
+        draft_code: r.get(3)?, draft_generation: r.get(4)?,
+        created_at: r.get(5)?, updated_at: r.get(6)?,
+    })
+}
+
+#[derive(Deserialize)]
+pub struct CreateProject { pub name: String, pub description: Option<String> }
+
+pub async fn create(cx: Cx, auth: AuthUser, body: Option<Json<CreateProject>>) -> AppResult<(axum::http::StatusCode, Json<Value>)> {
+    let Json(r) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?;
+    let name = r.name.trim().to_string();
+    if name.is_empty() || name.len() > 200 { return Err(AppError::bad("validation", "name required (max 200 chars)")); }
+    let id = new_id();
+    let desc = r.description.unwrap_or_default();
+    let ts = now_iso();
+    let uid = auth.id.clone();
+    cx.with_db(|db| -> AppResult<()> {
+        db.execute("INSERT INTO projects (id,user_id,name,description,draft_code,draft_generation,created_at,updated_at) VALUES (?1,?2,?3,?4,'',0,?5,?5)",
+            rusqlite::params![&id, &uid, &name, &desc, &ts])?;
+        Ok(())
+    }).await?;
+    audit(&cx, Some(&auth.id), "project_create", &id, "ok").await;
+    let p = load_owned(&cx, &auth.id, &id).await?;
+    Ok((axum::http::StatusCode::CREATED, Json(json!(p))))
+}
+
+pub async fn list(cx: Cx, auth: AuthUser) -> AppResult<Json<Value>> {
+    let items: Vec<Project> = cx.with_db(|db| {
+        let mut st = db.prepare("SELECT id,name,description,draft_code,draft_generation,created_at,updated_at FROM projects WHERE user_id=?1 ORDER BY updated_at DESC")?;
+        let mut rows = st.query([auth.id.clone()])?;
+        let mut out = Vec::new();
+        while let Some(r) = rows.next()? { out.push(row_project(r)?); }
+        Ok::<_, AppError>(out)
+    }).await?;
+    Ok(Json(json!({ "items": items })))
+}
+
+pub async fn get(cx: Cx, auth: AuthUser, Path(id): Path<String>) -> AppResult<Json<Value>> {
+    let p = load_owned(&cx, &auth.id, &id).await?;
+    Ok(Json(json!(p)))
+}
+
+#[derive(Deserialize)]
+pub struct PatchProject { pub name: Option<String>, pub description: Option<String> }
+
+pub async fn patch(cx: Cx, auth: AuthUser, Path(id): Path<String>, body: Option<Json<PatchProject>>) -> AppResult<Json<Value>> {
+    let Json(r) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?;
+    if let Some(n) = &r.name {
+        if n.trim().is_empty() { return Err(AppError::bad("validation", "name must not be empty")); }
+    }
+    load_owned(&cx, &auth.id, &id).await?;
+    cx.with_db(|db| -> AppResult<()> {
+        db.execute("UPDATE projects SET name=COALESCE(?1,name), description=COALESCE(?2,description), updated_at=?3 WHERE id=?4",
+            rusqlite::params![r.name, r.description, now_iso(), &id])?;
+        Ok(())
+    }).await?;
+    let p = load_owned(&cx, &auth.id, &id).await?;
+    Ok(Json(json!(p)))
+}
+
+#[derive(Deserialize)]
+pub struct PutDraft { pub code: String, pub expected_generation: i64 }
+
+pub async fn put_draft(cx: Cx, auth: AuthUser, Path(id): Path<String>, body: Option<Json<PutDraft>>) -> AppResult<Json<Value>> {
+    let Json(r) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?;
+    if r.code.len() > crate::config::MAX_CODE_LEN { return Err(AppError::bad("validation", "code too long")); }
+    let p: Project = cx.with_db(|db| -> AppResult<Project> {
+        let cur: i64 = db.query_row("SELECT draft_generation FROM projects WHERE id=?1 AND user_id=?2",
+            rusqlite::params![&id, &auth.id], |row| row.get(0))
+            .map_err(|_| AppError::not_found("project not found"))?;
+        if cur != r.expected_generation {
+            return Err(AppError::conflict("stale_generation", "draft changed; reload first")
+                .with_details(json!({ "current_generation": cur })));
+        }
+        db.execute("UPDATE projects SET draft_code=?1, draft_generation=?2, updated_at=?3 WHERE id=?4",
+            rusqlite::params![&r.code, cur + 1, now_iso(), &id])?;
+        db.query_row("SELECT id,name,description,draft_code,draft_generation,created_at,updated_at FROM projects WHERE id=?1", [&id], row_project)
+            .map_err(AppError::from)
+    }).await?;
+    Ok(Json(json!(p)))
+}
+
+// ---- versions ----
+
+#[derive(Serialize)]
+pub struct Version {
+    pub id: String,
+    pub project_id: String,
+    pub code: String,
+    pub hash: String,
+    pub message: String,
+    pub created_at: String,
+    pub source: String,
+}
+
+fn row_version(r: &rusqlite::Row) -> rusqlite::Result<Version> {
+    Ok(Version { id: r.get(0)?, project_id: r.get(1)?, code: r.get(2)?, hash: r.get(3)?, message: r.get(4)?, created_at: r.get(5)?, source: r.get(6)? })
+}
+
+fn select_version(db: &mut rusqlite::Connection, vid: &str, pid: &str) -> rusqlite::Result<Version> {
+    let mut st = db.prepare("SELECT id,project_id,code,hash,message,created_at,source FROM project_versions WHERE id=?1 AND project_id=?2")?;
+    st.query_row(rusqlite::params![vid, pid], row_version)
+}
+
+pub async fn list_versions(cx: Cx, auth: AuthUser, Path(id): Path<String>) -> AppResult<Json<Value>> {
+    load_owned(&cx, &auth.id, &id).await?;
+    let mut items: Vec<Version> = cx.with_db(|db| {
+        let mut st = db.prepare("SELECT id,project_id,code,hash,message,created_at,source FROM project_versions WHERE project_id=?1 ORDER BY created_at ASC")?;
+        let mut rows = st.query([id.clone()])?;
+        let mut out = Vec::new();
+        while let Some(r) = rows.next()? { out.push(row_version(r)?); }
+        Ok::<_, AppError>(out)
+    }).await?;
+    items.reverse();
+    Ok(Json(json!({ "items": items })))
+}
+
+pub async fn get_version(cx: Cx, auth: AuthUser, Path((pid, vid)): Path<(String, String)>) -> AppResult<Json<Value>> {
+    load_owned(&cx, &auth.id, &pid).await?;
+    let v: Version = cx.with_db(|db| select_version(db, &vid, &pid))
+        .await
+        .map_err(|_| AppError::not_found("version not found"))?;
+    Ok(Json(json!(v)))
+}
+
+pub async fn diff_versions(cx: Cx, auth: AuthUser, Path((pid, vid)): Path<(String, String)>, body: Option<Json<Value>>) -> AppResult<Json<Value>> {
+    let base: Value = body.map(|b| b.0).unwrap_or(json!({}));
+    load_owned(&cx, &auth.id, &pid).await?;
+    let v: Version = cx.with_db(|db| select_version(db, &vid, &pid))
+        .await
+        .map_err(|_| AppError::not_found("version not found"))?;
+    let base_code: String = match base.get("base_version").and_then(|b| b.as_str()) {
+        Some(bv) => cx.with_db(|db| {
+            db.query_row("SELECT code FROM project_versions WHERE id=?1 AND project_id=?2", rusqlite::params![bv, &pid], |r| r.get(0))
+                .map_err(|_| AppError::bad("validation", "base_version not found"))
+        }).await?,
+        None => String::new(),
+    };
+    Ok(Json(json!({ "diff": unified_diff(&base_code, &v.code), "version": { "id": v.id, "hash": v.hash, "message": v.message } })))
+}
+
+pub async fn create_version(cx: Cx, auth: AuthUser, Path(id): Path<String>, body: Option<Json<Value>>) -> AppResult<(axum::http::StatusCode, Json<Value>)> {
+    let b = body.map(|b| b.0).unwrap_or(json!({}));
+    let message = b.get("message").and_then(|m| m.as_str()).unwrap_or("").to_string();
+    if message.trim().is_empty() { return Err(AppError::bad("validation", "message required")); }
+    let v = create_version_from_draft(&cx, &auth.id, &id, &message, "manual").await?;
+    Ok((axum::http::StatusCode::CREATED, Json(json!(v))))
+}
+
+/// Snapshot the current draft as an immutable version.
+pub async fn create_version_from_draft(cx: &Cx, user_id: &str, project_id: &str, message: &str, source: &str) -> AppResult<Version> {
+    let p = load_owned(cx, user_id, project_id).await?;
+    if p.draft_code.trim().is_empty() { return Err(AppError::bad("validation", "draft is empty; nothing to version")); }
+    let hash = sha256_hex(p.draft_code.as_bytes());
+    let draft = p.draft_code.clone();
+    let v: Version = cx.with_db(|db| -> AppResult<Version> {
+        let id = new_id();
+        let ts = now_iso();
+        db.execute("INSERT INTO project_versions (id,project_id,code,hash,message,source,created_at) VALUES (?1,?2,?3,?4,?5,?6,?7)",
+            rusqlite::params![&id, project_id, &draft, &hash, message, source, &ts])?;
+        let mut st = db.prepare("SELECT id,project_id,code,hash,message,created_at,source FROM project_versions WHERE id=?1")?;
+        st.query_row([&id], row_version).map_err(AppError::from)
+    }).await?;
+    audit(cx, Some(user_id), "version_create", project_id, "ok").await;
+    Ok(v)
+}
+
+#[derive(Deserialize)]
+pub struct RestoreReq { pub version_id: String, pub expected_generation: i64 }
+
+pub async fn restore(cx: Cx, auth: AuthUser, Path(id): Path<String>, body: Option<Json<RestoreReq>>) -> AppResult<Json<Value>> {
+    let Json(r) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?;
+    cx.with_db(|db| -> AppResult<()> {
+        let cur: i64 = db.query_row("SELECT draft_generation FROM projects WHERE id=?1 AND user_id=?2", rusqlite::params![&id, &auth.id], |row| row.get(0))
+            .map_err(|_| AppError::not_found("project not found"))?;
+        if cur != r.expected_generation { return Err(AppError::conflict("stale_generation", "draft changed; reload first")); }
+        let code: String = db.query_row("SELECT code FROM project_versions WHERE id=?1 AND project_id=?2", rusqlite::params![&r.version_id, &id], |row| row.get(0))
+            .map_err(|_| AppError::not_found("version not found"))?;
+        db.execute("UPDATE projects SET draft_code=?1, draft_generation=?2, updated_at=?3 WHERE id=?4",
+            rusqlite::params![&code, cur + 1, now_iso(), &id])?;
+        Ok(())
+    }).await?;
+    audit(&cx, Some(&auth.id), "version_restore", &id, "ok").await;
+    let p = load_owned(&cx, &auth.id, &id).await?;
+    Ok(Json(json!(p)))
+}
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);
+    }
+}
diff --git a/server/src/state.rs b/server/src/state.rs
new file mode 100644
index 0000000..512bcb3
--- /dev/null
+++ b/server/src/state.rs
@@ -0,0 +1,55 @@
+use std::sync::Arc;
+use tokio::sync::{Mutex as AsyncMutex, Semaphore};
+
+use crate::config::Config;
+
+/// Shared application state. The single SQLite connection is serialized behind
+/// an async mutex; concurrent fetch/backtest work is bounded by semaphores.
+pub struct AppState {
+    pub cfg: Config,
+    pub db: AsyncMutex<rusqlite::Connection>,
+    pub run_sem: Arc<Semaphore>,
+    pub fetch_sem: Arc<Semaphore>,
+}
+
+/// Fixed shared contract alias for handler arguments: the axum State extractor
+/// over `Arc<AppState>`. Handlers take `cx: Cx`, helpers take `&Cx`.
+pub type Cx = axum::extract::State<std::sync::Arc<AppState>>;
+
+/// Legacy alias kept for modules (auth, projects, datasets) whose handlers take
+/// the bare `Arc<AppState>`; both styles are valid extractors for this state.
+pub type S = Arc<AppState>;
+
+impl AppState {
+    /// Lock the connection, run the closure, return its result exactly.
+    /// Never `.await` inside the closure; keep transactions in ONE closure.
+    pub async fn with_db<R>(&self, f: impl FnOnce(&mut rusqlite::Connection) -> R) -> R {
+        let mut db = self.db.lock().await;
+        f(&mut db)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::config::Config;
+
+    #[tokio::test]
+    async fn with_db_returns_closure_result_and_composes() {
+        let cfg = Config::from_env();
+        let conn = rusqlite::Connection::open_in_memory().unwrap();
+        let st = AppState {
+            cfg: cfg.clone(),
+            db: AsyncMutex::new(conn),
+            run_sem: Arc::new(Semaphore::new(1)),
+            fetch_sem: Arc::new(Semaphore::new(1)),
+        };
+        let n: i64 = st.with_db(|db| {
+            db.execute("CREATE TABLE t(x INTEGER)", []).unwrap();
+            db.execute("INSERT INTO t VALUES (42)", []).unwrap();
+            db.query_row("SELECT SUM(x) FROM t", [], |r| r.get(0)).unwrap()
+        }).await;
+        assert_eq!(n, 42);
+        st.with_db(|_db| ()).await;
+    }
+}
diff --git a/server/src/store.rs b/server/src/store.rs
new file mode 100644
index 0000000..f371d10
--- /dev/null
+++ b/server/src/store.rs
@@ -0,0 +1,206 @@
+use std::path::{Path, PathBuf};
+
+use sha2::{Digest, Sha256};
+
+/// Immutable content-addressed object storage under {data_dir}/objects/{hash[:2]}/{hash}.{ext}.
+/// Content is deduplicated: storing identical bytes twice keeps the first object.
+pub struct ObjectStore {
+    pub root: PathBuf,
+}
+
+/// One ingested artifact (file content hashed and copied into the object store).
+#[derive(Debug, Clone)]
+pub struct StoredObject {
+    /// sha256 of content
+    pub hash: String,
+    /// path relative to the object root (immutable stored path)
+    pub stored_path: String,
+    /// file name as produced by the worker, safe for /data mounts
+    pub mount_name: String,
+    pub size: u64,
+}
+
+impl ObjectStore {
+    pub fn new(data_dir: &str) -> Self {
+        ObjectStore { root: Path::new(data_dir).join("objects") }
+    }
+
+    /// Store raw bytes by content hash. Returns (hash, stored relative path).
+    pub fn store(&self, bytes: &[u8], filename: &str) -> std::io::Result<(String, String)> {
+        let hash = format!("{:x}", Sha256::digest(bytes));
+        let dir = self.root.join(&hash[..2]);
+        std::fs::create_dir_all(&dir)?;
+        let ext = safe_ext(filename);
+        let path = dir.join(format!("{hash}.{ext}"));
+        if !path.is_file() {
+            // Atomic write in the final directory; suffix append (not with_extension)
+            // so different source extensions cannot collide on one tmp name.
+            let tmp = dir.join(format!("{hash}.{ext}.tmp"));
+            std::fs::write(&tmp, bytes)?;
+            std::fs::rename(&tmp, &path)?;
+        }
+        let rel = path
+            .strip_prefix(&self.root)
+            .map(|p| p.to_string_lossy().into_owned())
+            .unwrap_or_else(|_| path.to_string_lossy().into_owned());
+        Ok((hash, rel))
+    }
+
+    /// Map a stored relative path (server controlled) back to an absolute path.
+    pub fn absolute(&self, rel: &str) -> PathBuf {
+        self.root.join(rel)
+    }
+
+    /// Hash and ingest every regular file under an output directory (worker
+    /// artifacts). Returns one entry per file, sorted for determinism.
+    /// Rejects symlinked entries.
+    pub fn ingest_directory(&self, dir: &Path) -> std::io::Result<Vec<StoredObject>> {
+        let mut files: Vec<PathBuf> = Vec::new();
+        collect_files(dir, dir, &mut files)?;
+        files.sort();
+        let mut out = Vec::with_capacity(files.len());
+        for f in files {
+            let is_symlink = f.symlink_metadata()?.file_type().is_symlink()
+                || std::fs::symlink_metadata(&f)?.file_type().is_symlink();
+            if is_symlink {
+                return Err(std::io::Error::new(
+                    std::io::ErrorKind::InvalidInput,
+                    "symlinked artifact rejected",
+                ));
+            }
+            let bytes = std::fs::read(&f)?;
+            let mount_name = f
+                .file_name()
+                .and_then(|n| n.to_str())
+                .unwrap_or("artifact.bin")
+                .to_string();
+            let rel_path = f.strip_prefix(dir).expect("strip_prefix");
+            let (hash, stored_path) =
+                self.store(&bytes, &mount_name)?;
+            out.push(StoredObject {
+                hash,
+                stored_path,
+                mount_name: rel_path.display().to_string(),
+                size: bytes.len() as u64,
+            });
+        }
+        Ok(out)
+    }
+}
+
+fn collect_files(_root: &Path, dir: &Path, out: &mut Vec<PathBuf>) -> std::io::Result<()> {
+    for entry in std::fs::read_dir(dir)? {
+        let p = entry?.path();
+        let ty = p.symlink_metadata()?.file_type();
+        if ty.is_symlink() {
+            // path traversal defense: no symlinked artifacts, ever
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidInput,
+                "symlink in artifact tree rejected",
+            ));
+        }
+        if ty.is_dir() {
+            collect_files(_root, &p, out)?;
+        } else {
+            out.push(p);
+        }
+    }
+    Ok(())
+}
+
+/// Sanitize a filename for use as an object extension: only the final
+/// extension survives, restricted to alphanumeric chars.
+fn safe_ext(filename: &str) -> String {
+    let base = filename.rsplit('/').next().unwrap_or("data");
+    let e = base.rsplit('.').next().unwrap_or("bin").to_string();
+    let v: String = e.chars().filter(|c| c.is_ascii_alphanumeric()).collect();
+    if v.is_empty() || v.parse::<usize>().is_ok() {
+        "bin".into()
+    } else {
+        v.to_lowercase()
+    }
+}
+
+/// Public wrapper used when a filename has no usable extension.
+pub fn filename_or_bin(name: &str, fallback: &str) -> String {
+    let base = name.rsplit('/').next().unwrap_or(fallback);
+    let v: String = base
+        .chars()
+        .filter(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '-' | '_'))
+        .collect();
+    if v.is_empty() || v == "." {
+        fallback.into()
+    } else {
+        v
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn tmp_store(tag: &str) -> (tempfile::TempDir, ObjectStore) {
+        let t = tempfile::tempdir_in("/tmp/opencode").unwrap();
+        let store = ObjectStore::new(t.path().join(tag).to_str().unwrap());
+        (t, store)
+    }
+
+    #[test]
+    fn store_is_content_addressed_and_deduplicated() {
+        let (_t, s) = tmp_store("objs1");
+        let (h1, p1) = s.store(b"hello world", "a.csv").unwrap();
+        let (h2, p2) = s.store(b"hello world", "b.csv").unwrap();
+        assert_eq!(h1, h2);
+        assert_eq!(p1, p2);
+        assert!(p1.starts_with(&h1[..2]), "{p1}");
+        let abs = s.absolute(&p1);
+        assert_eq!(std::fs::read(&abs).unwrap(), b"hello world".to_vec());
+        assert!(s.root.join(&p1) == abs, "stored path resolves under root");
+    }
+
+    #[test]
+    fn same_stem_different_extension_no_collision() {
+        let (_t, s) = tmp_store("objs2");
+        let (_, a) = s.store(b"csv-bytes", "obj.csv").unwrap();
+        let (_, b) = s.store(b"json-bytes", "obj.json").unwrap();
+        assert_ne!(a, b);
+        assert!(!a.ends_with(".tmp"));
+        assert!(std::fs::read(s.absolute(&a)).unwrap().starts_with(b"csv"));
+    }
+
+    #[test]
+    fn ingest_directory_walks_and_rejects_symlinks() {
+        let td = tempfile::tempdir_in("/tmp/opencode").unwrap();
+        let out = td.path().join("out");
+        let objd = out.join("objects");
+        std::fs::create_dir_all(&objd).unwrap();
+        std::fs::write(objd.join("data.csv"), b"date,close\n2024-01-02,10\n").unwrap();
+        std::fs::write(out.join("result.json"), b"{\"status\":\"ready\"}").unwrap();
+        let (_t, s) = tmp_store("objs3");
+        let stored = s.ingest_directory(&out).unwrap();
+        assert_eq!(stored.len(), 2);
+        let names: Vec<&str> = stored.iter().map(|o| o.mount_name.as_str()).collect();
+        assert!(names.contains_all(&["objects/data.csv", "result.json"]), "{names:?}");
+        // same content re-ingested maps to the same stored object
+        let again = s.ingest_directory(&out).unwrap();
+        for o in &stored {
+            assert!(again.iter().any(|n| n.hash == o.hash));
+        }
+        // symlink rejection
+        std::os::unix::fs::symlink(
+            objd.join("data.csv"),
+            objd.join("data_link.csv"),
+        )
+        .unwrap();
+        assert!(s.ingest_directory(&out).is_err());
+    }
+}
+
+trait ContainsAll {
+    fn contains_all(&self, needles: &[&str]) -> bool;
+}
+impl ContainsAll for Vec<&str> {
+    fn contains_all(&self, needles: &[&str]) -> bool {
+        needles.iter().all(|n| self.iter().any(|m| m.contains(n)))
+    }
+}
diff --git a/server/src/util.rs b/server/src/util.rs
new file mode 100644
index 0000000..5a82806
--- /dev/null
+++ b/server/src/util.rs
@@ -0,0 +1,138 @@
+use sha2::{Digest, Sha256};
+
+pub fn sha256_hex(data: &[u8]) -> String {
+    let mut h = Sha256::new();
+    h.update(data);
+    hex::encode(h.finalize())
+}
+
+/// Synchronous UTC timestamp helper, ISO 8601 with millisecond precision.
+pub fn now_iso() -> String {
+    chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
+}
+
+pub fn plus_hours(hours: i64) -> String {
+    (chrono::Utc::now() + chrono::Duration::hours(hours))
+        .to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
+}
+
+pub fn new_id() -> String {
+    uuid::Uuid::new_v4().to_string()
+}
+
+pub fn gen_token() -> String {
+    use rand::RngCore;
+    let mut buf = [0u8; 32];
+    rand::rngs::OsRng.fill_bytes(&mut buf);
+    hex::encode(buf)
+}
+
+/// Line based unified diff (full document, no hunks) used by the AI draft diff.
+pub fn unified_diff(a: &str, b: &str) -> String {
+    let x: Vec<&str> = a.lines().collect();
+    let y: Vec<&str> = b.lines().collect();
+    let n = x.len();
+    let m = y.len();
+    // LCS table
+    let mut dp = vec![vec![0usize; m + 1]; n + 1];
+    for i in (0..n).rev() {
+        for j in (0..m).rev() {
+            dp[i][j] = if x[i] == y[j] {
+                dp[i + 1][j + 1] + 1
+            } else {
+                dp[i + 1][j].max(dp[i][j + 1])
+            };
+        }
+    }
+    let mut out = String::new();
+    let mut ctx = std::collections::VecDeque::<usize>::new();
+    let (mut i, mut j) = (0usize, 0usize);
+    while i < n || j < m {
+        if i < n && j < m && x[i] == y[j] {
+            ctx.push_back(i);
+            if ctx.len() > 2 {
+                let keep = ctx.pop_front().unwrap();
+                out.push_str(&format!(" {}\n", x[keep]));
+            }
+            i += 1;
+            j += 1;
+        } else {
+            // entering a change: emit queued context lines first
+            for k in ctx.drain(..) {
+                out.push_str(&format!(" {}\n", x[k]));
+            }
+            if i < n && (j >= m || dp[i + 1][j] >= dp[i][j + 1]) {
+                out.push_str(&format!("-{}\n", x[i]));
+                i += 1;
+            } else {
+                out.push_str(&format!("+{}\n", y[j]));
+                j += 1;
+            }
+        }
+    }
+    out
+}
+
+/// Deterministic canonical JSON string (recursively sorted keys) for hashing.
+pub fn canonical_json(v: &serde_json::Value) -> String {
+    fn sort(v: &serde_json::Value) -> serde_json::Value {
+        match v {
+            serde_json::Value::Array(a) => serde_json::Value::Array(a.iter().map(sort).collect()),
+            serde_json::Value::Object(o) => {
+                let mut keys: Vec<(String, serde_json::Value)> =
+                    o.iter().map(|(k, v)| (k.clone(), sort(v))).collect();
+                keys.sort_by(|a, b| a.0.cmp(&b.0));
+                serde_json::Value::Object(keys.into_iter().collect())
+            }
+            other => other.clone(),
+        }
+    }
+    sort(v).to_string()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn now_iso_is_synchronous_iso8601() {
+        let t = now_iso();
+        let parsed = chrono::DateTime::parse_from_rfc3339(&t).expect("ISO timestamp");
+        assert!(parsed.timestamp() > 1_700_000_000);
+        assert!(t.ends_with('Z'));
+    }
+
+    #[test]
+    fn plus_hours_and_tokens() {
+        let e = chrono::DateTime::parse_from_rfc3339(&plus_hours(1)).unwrap();
+        let n = chrono::DateTime::parse_from_rfc3339(&now_iso()).unwrap();
+        assert!((e - n).num_minutes() >= 59);
+        let a = gen_token();
+        assert_eq!(a.len(), 64);
+        assert_ne!(a, gen_token());
+        assert_eq!(new_id().len(), 36);
+    }
+
+    #[test]
+    fn sha256_is_stable() {
+        assert_eq!(sha256_hex(b"abc"), "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad");
+    }
+
+    #[test]
+    fn unified_diff_marks_changes_with_context() {
+        let a = "x\nkeep\ny";
+        let b = "x\nkeep\nz";
+        let d = unified_diff(a, b);
+        assert!(d.contains("-y\n+z\n"), "diff: {d}");
+        assert!(d.contains(" keep"));
+        let same = unified_diff(a, "x\nkeep\ny");
+        assert!(!same.contains('+') && !same.contains('-'), "no diff: {same}");
+    }
+
+    #[test]
+    fn canonical_json_sorts_recursively() {
+        let v: serde_json::Value =
+            serde_json::from_str(r#"{"b":1,"a":{"y":[3,2],"x":true}}"#).unwrap();
+        assert_eq!(canonical_json(&v), r#"{"a":{"x":true,"y":[3,2]},"b":1}"#);
+    }
+}
diff --git a/server/src/worker.rs b/server/src/worker.rs
new file mode 100644
index 0000000..8e28de5
--- /dev/null
+++ b/server/src/worker.rs
@@ -0,0 +1,482 @@
+use std::sync::Arc;
+
+use std::os::unix::fs::PermissionsExt;
+
+use tokio::process::Command;
+
+use crate::error::{AppError, AppResult};
+use crate::state::AppState;
+
+const NONROOT_UID: u64 = 65534;
+const MAX_STDOUT: usize = 64_000;
+const MAX_STDERR: usize = 16_000;
+
+#[derive(Debug)]
+pub struct ContainerResult {
+    pub status: Option<i32>,
+    pub stdout: String,
+    pub stderr: String,
+}
+
+impl ContainerResult {
+    pub fn ok(&self) -> bool {
+        self.status == Some(0)
+    }
+}
+
+/// Docker run arguments (after `docker`). Isolation contract:
+/// no caps, no privilege escalation, non-root uid 65534, read-only root
+/// filesystem, small tmpfs, pids/memory/cpu limits, network only when the
+/// task requires the data provider; backtest always runs with --network=none.
+pub fn docker_args(
+    network: bool,
+    mounts: &[(String, String, bool)],
+    image: &str,
+    name: &str,
+    args: &[String],
+) -> Vec<String> {
+    let mut a: Vec<String> = [
+        "run",
+        "--rm",
+        // Signals reaching the runner CLI must never be proxied into the
+        // container: on service stop the container is expected to die via the
+        // app's exact-name cleanup (cancel/timeout/startup), not via a CLI
+        // relay; a proxying CLI was observed lingering under systemd
+        // final-sigterm (TimeoutStopSec exhaustion).
+        "--sig-proxy=false",
+        "--name",
+        name,
+        "--cap-drop=ALL",
+        "--security-opt=no-new-privileges",
+        CONCAT_USER,
+        "--read-only",
+        "--tmpfs=/tmp:rw,size=256m,mode=1777",
+        "--pids-limit=128",
+        "--memory=2g",
+        "--cpus=2",
+    ]
+    .iter()
+    .map(|s| s.to_string())
+    .collect();
+    // explicit network choice; only data fetch/search uses the bridge
+    a.push(if network { "--network=bridge".into() } else { "--network=none".into() });
+    for (src, dst, ro) in mounts {
+        a.push("-v".into());
+        a.push(format!("{}:{}{}", src, dst, if *ro { ":ro" } else { "" }));
+    }
+    a.push(image.to_string());
+    a.extend_from_slice(args);
+    a
+}
+
+async fn drain_bounded<R>(rd: R, max: usize) -> String
+where
+    R: tokio::io::AsyncRead + Unpin,
+{
+    use tokio::io::AsyncReadExt;
+    let mut buf = Vec::with_capacity(1024);
+    let mut chunk = [0u8; 8192];
+    let mut reader = rd;
+    loop {
+        match reader.read(&mut chunk).await {
+            Ok(0) => break,
+            Ok(n) => {
+                // drain everything, but keep only the tail-relevant bounded prefix
+                if buf.len() < max {
+                    let take = n.min(max - buf.len());
+                    buf.extend_from_slice(&chunk[..take]);
+                }
+                if buf.len() >= max {
+                    // continue draining the pipe without buffering the rest
+                    let mut sink = [0u8; 8192];
+                    loop {
+                        match reader.read(&mut sink).await {
+                            Ok(0) | Err(_) => break,
+                            Ok(_) => {}
+                        }
+                    }
+                    break;
+                }
+            }
+            Err(_) => break,
+        }
+    }
+    truncate(&String::from_utf8_lossy(&buf), max)
+}
+
+async fn execute_docker(full: &[String], name: &str, timeout_secs: u64) -> AppResult<ContainerResult> {
+    execute_docker_named("docker", full, name, timeout_secs).await
+}
+
+async fn execute_docker_named(
+    docker_bin: &str,
+    full: &[String],
+    name: &str,
+    timeout_secs: u64,
+) -> AppResult<ContainerResult> {
+    let mut child = Command::new(docker_bin)
+        .args(full)
+        // Kill the runner process when the future that owns it is dropped. A
+        // graceful server stop drops the jobs task; without this the docker
+        // CLI child would linger inside the systemd cgroup and block the
+        // unit stop until TimeoutStopSec forced SIGKILL. Bounded lifetime.
+        .kill_on_drop(true)
+        .stdout(std::process::Stdio::piped())
+        .stderr(std::process::Stdio::piped())
+        .spawn()
+        .map_err(|e| AppError::internal(format!("failed to spawn worker container: {e}")))?;
+    let stdout = child.stdout.take().expect("stdout piped");
+    let stderr = child.stderr.take().expect("stderr piped");
+    let stdout_task = tokio::spawn(drain_bounded(stdout, MAX_STDOUT));
+    let stderr_task = tokio::spawn(drain_bounded(stderr, MAX_STDERR));
+
+    let wait_res = tokio::time::timeout(
+        std::time::Duration::from_secs(timeout_secs),
+        child.wait(),
+    )
+    .await;
+
+    let status = match wait_res {
+        Ok(Ok(st)) => st.code(),
+        Ok(Err(e)) => {
+            return Err(AppError::internal(format!("worker process error: {e}")).with_code("runner_failed"))
+        }
+        Err(_) => {
+            // Timeout: kill the specific container by name so user code cannot
+            // linger; then reap the docker client process.
+            kill_container(name).await;
+            let _ = child.wait().await;
+            return Err(AppError::internal(format!(
+                "worker container timed out after {timeout_secs}s and was killed: {name}"
+            ))
+            .with_code("runner_timeout"));
+        }
+    };
+
+    Ok(ContainerResult {
+        status,
+        stdout: stdout_task.await.unwrap_or_default(),
+        stderr: stderr_task.await.unwrap_or_default(),
+    })
+}
+
+/// Kill and remove the named container. Returns true when docker succeeded.
+pub async fn cancel_container(name: &str) -> bool {
+    kill_container(name).await
+}
+
+async fn kill_container(name: &str) -> bool {
+    Command::new("docker")
+        .args(["kill", name])
+        .stdout(std::process::Stdio::null())
+        .stderr(std::process::Stdio::null())
+        .output()
+        .await
+        .ok();
+    Command::new("docker")
+        .args(["rm", "-f", name])
+        .stdout(std::process::Stdio::null())
+        .stderr(std::process::Stdio::null())
+        .output()
+        .await
+        .map(|o| o.status.success())
+        .unwrap_or(false)
+}
+
+pub async fn docker_available() -> bool {
+    tokio::process::Command::new("docker")
+        .args(["version", "--format", "ok"])
+        .stdout(std::process::Stdio::null())
+        .stderr(std::process::Stdio::null())
+        .output()
+        .await
+        .map(|o| o.status.success())
+        .unwrap_or(false)
+}
+
+/// Mounts need world permissions: the container runs as uid 65534 while host
+/// ownership is the server user. Best effort only.
+fn prepare_mounts(mounts: &[(String, String, bool)]) {
+    for (src, _dst, ro) in mounts {
+        let p = std::path::Path::new(src);
+        if !p.is_dir() {
+            continue;
+        }
+        let mode = if *ro { 0o755 } else { 0o777 };
+        let _ = std::fs::set_permissions(p, std::fs::Permissions::from_mode(mode));
+        // Files inside ro input dirs must be world readable; output files are
+        // written by the container with its umask.
+        if *ro {
+            if let Ok(rd) = std::fs::read_dir(p) {
+                for e in rd.flatten() {
+                    let fmode = if e.path().is_file() {
+                        std::fs::Permissions::from_mode(0o644)
+                    } else {
+                        std::fs::Permissions::from_mode(0o755)
+                    };
+                    let _ = std::fs::set_permissions(e.path(), fmode);
+                }
+            }
+        }
+    }
+}
+
+/// Run the worker image with a fixed container name so cancel maps to one
+/// specific container id (never a global prune).
+pub async fn run_named(
+    cx: &Arc<AppState>,
+    network: bool,
+    mounts: &[(String, String, bool)],
+    args: &[String],
+    name: &str,
+    timeout_secs: u64,
+) -> AppResult<ContainerResult> {
+    let cfg = &cx.cfg;
+    prepare_mounts(mounts);
+    let full = docker_args(network, mounts, &cfg.worker_image, name, args);
+    execute_docker(&full, name, timeout_secs).await
+}
+
+/// Instrument catalog search through the worker container (network enabled).
+/// Returns the JSON array printed by `worker.main search`; failures are real
+/// errors, never an empty fake success. Results are cached briefly per query so
+/// repeated keystrokes reuse the actual provider identity (full item payloads).
+pub async fn search_instruments(
+    cx: &Arc<AppState>,
+    query: &str,
+    limit: i64,
+) -> AppResult<Vec<serde_json::Value>> {
+    let query = query.trim().to_string();
+    if query.is_empty() {
+        return Ok(Vec::new());
+    }
+    let limit = if (1..=100).contains(&limit) { limit } else { 50 };
+    let cache_key = format!("q={query}&limit={limit}");
+    if let Some(items) = catalog_cache_get(&cache_key) {
+        return Ok(items);
+    }
+    let args = vec![
+        "python".into(),
+        "-m".into(),
+        "worker.main".into(),
+        "search".into(),
+        "--query".into(),
+        query,
+        "--limit".into(),
+        limit.to_string(),
+    ];
+    // The worker enforces <=4s per HTTP source; the container including startup
+    // is bounded here. This is the outer bound for the whole search round trip.
+    let res = run_named(cx, true, &[], &args, &random_name(), 12).await?;
+    if !res.ok() {
+        return Err(AppError::bad("search_failed", truncate(&res.stderr, 500)));
+    }
+    // The contract is a single JSON object envelope on stdout:
+    // {"status":"ready"|"failed","items":[...],"error":{...},...}
+    // A `failed` status is surfaced as an error, never as an empty success.
+    let envelope: serde_json::Value = serde_json::from_str(res.stdout.trim())
+        .map_err(|e| AppError::internal(format!("invalid search envelope: {e}")))?;
+    let status = envelope
+        .get("status")
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| AppError::internal("invalid search envelope: missing status"))?;
+    if status != "ready" {
+        let err = envelope.get("error").cloned().unwrap_or(serde_json::Value::Null);
+        let code = err
+            .get("code")
+            .and_then(|v| v.as_str())
+            .unwrap_or("search_unavailable");
+        let message = err
+            .get("message")
+            .and_then(|v| v.as_str())
+            .unwrap_or("instrument search providers unavailable");
+        // provider error codes are dynamic; they ride in the message so the
+        // HTTP layer keeps static error codes
+        return Err(AppError::bad(
+            "search_unavailable",
+            format!("[{code}] {message}"),
+        ));
+    }
+    let items: Vec<serde_json::Value> = serde_json::from_value(
+        envelope.get("items").cloned().unwrap_or(serde_json::Value::Null),
+    )
+    .map_err(|e| AppError::internal(format!("invalid search envelope items: {e}")))?;
+    catalog_cache_put(&cache_key, &items);
+    Ok(items)
+}
+
+/// Small in-process TTL cache for catalog search results (provider identity).
+const CATALOG_TTL_SECS: u64 = 300;
+const CATALOG_MAX_ENTRIES: usize = 128;
+
+fn catalog_cache() -> &'static tokio::sync::Mutex<std::collections::HashMap<String, (std::time::Instant, Vec<serde_json::Value>)>> {
+    static MAP: std::sync::OnceLock<tokio::sync::Mutex<std::collections::HashMap<String, (std::time::Instant, Vec<serde_json::Value>)>>> =
+        std::sync::OnceLock::new();
+    MAP.get_or_init(|| tokio::sync::Mutex::new(std::collections::HashMap::new()))
+}
+
+fn catalog_cache_get(key: &str) -> Option<Vec<serde_json::Value>> {
+    // Instant checks must not block behind stdio work; try_lock is fine here.
+    let map = catalog_cache().try_lock().ok()?;
+    let (at, items) = map.get(key)?;
+    if at.elapsed() < std::time::Duration::from_secs(CATALOG_TTL_SECS) {
+        Some(items.clone())
+    } else {
+        None
+    }
+}
+
+fn catalog_cache_put(key: &str, items: &[serde_json::Value]) {
+    if let Ok(mut map) = catalog_cache().try_lock() {
+        if map.len() >= CATALOG_MAX_ENTRIES {
+            map.clear();
+        }
+        map.insert(key.to_string(), (std::time::Instant::now(), items.to_vec()));
+    }
+}
+
+fn random_name() -> String {
+    use rand::RngCore;
+    let mut buf = [0u8; 8];
+    rand::rngs::OsRng.fill_bytes(&mut buf);
+    format!("sl-run-{}", hex::encode(buf))
+}
+
+pub fn truncate(s: &str, n: usize) -> String {
+    s.chars().take(n).collect()
+}
+
+const CONCAT_USER: &str = "--user=65534:65534";
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn run_cli_never_proxies_signals_and_names_the_worker() {
+        // Cause-establishing regression: systemd restart stall happened because
+        // the runner CLI swallowed SIGTERM while relaying to the container.
+        // The CLI (and every other container invokation) must use sig-proxy=false.
+        for network in [true, false] {
+            let full = docker_args(
+                network,
+                &[("/job/output".into(), "/output".into(), false)],
+                "strategy-lab-worker:local",
+                "sl-run-t",
+                &["python".into()],
+            );
+            let s = full.join(" ");
+            assert!(s.contains("--sig-proxy=false"), "{s}");
+            assert!(s.contains("--rm --sig-proxy=false --name sl-run-t"));
+            assert!(s.contains(format!("--user={NONROOT_UID}:").as_str()));
+        }
+    }
+
+    #[tokio::test]
+    async fn runner_child_is_reaped_when_the_job_future_is_dropped() {
+        // Focused cause test: a runner child that ignores SIGTERM must not
+        // outlive its owning future (systemd restart stall cause). We emulate
+        // with a stub runner in a UNIQUE tempdir that ignores SIGTERM and then
+        // `exec`s into sleep so the stub PID IS the sleep process: kill_on_drop
+        // removes it entirely, leaving no grandchild orphan. Wait supervision
+        // uses exact PID checks (`/proc/<pid>`), never process-name scans.
+        let td = tempfile::tempdir_in("/tmp/opencode").unwrap();
+        let stub_path = td.path().join("runner-stub.sh");
+        let pid_file = td.path().join("stub.pid");
+        std::fs::write(&stub_path, format!(
+            "#!/bin/sh\ntrap '' TERM INT\necho $$ > {}\nexec sleep 500\n", pid_file.display()
+        )).unwrap();
+        {
+            use std::os::unix::fs::PermissionsExt;
+            std::fs::set_permissions(&stub_path, std::fs::Permissions::from_mode(0o755)).unwrap();
+        }
+        let stub = stub_path.clone();
+        let task = tokio::spawn(async move {
+            let _res = execute_docker_named(
+                stub.to_str().unwrap(),
+                &["long-running-stub".into()],
+                "sl-run-stub",
+                5,
+            ).await;
+        });
+        // wait until the stub published its exact PID
+        let mut stub_pid: Option<i32> = None;
+        for _ in 0..50 {
+            if let Ok(s) = std::fs::read_to_string(&pid_file) {
+                stub_pid = s.trim().parse().ok();
+            }
+            if stub_pid.is_some() { break; }
+            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+        }
+        let stub_pid: i32 = match stub_pid {
+            Some(p) => p,
+            None => panic!("stub never published its PID; test setup broken"),
+        };
+        let proc_dir = format!("/proc/{stub_pid}");
+        // sanity: stub alive; and after `exec` it IS the sleep grandchild
+        assert!(std::path::Path::new(&proc_dir).exists(), "stub pid {stub_pid} must be alive before the drop");
+        let cmd = std::fs::read_to_string(format!("{proc_dir}/cmdline")).unwrap_or_default();
+        assert!(cmd.contains("sleep"), "exec replace failed; test would leave an orphan: {cmd:?}");
+        task.abort(); // drops the execute_docker future mid-flight -> kill_on_drop -> SIGKILL
+        // exact-PID supervision: gone == /proc/<pid> has vanished
+        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(3);
+        let mut gone = false;
+        while std::time::Instant::now() < deadline {
+            if !std::path::Path::new(&proc_dir).exists() {
+                gone = true;
+                break;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+        }
+        assert!(gone, "runner stub pid {stub_pid} was not reaped when its future was dropped");
+        // no grandchild either: the exec'd sleep adopted the same PID, then was
+        // SIGKILLed with the rest; nothing named-scan was used.
+        let _ = stub_path; // file removed with the unique tempdir at scope end
+    }
+    fn mounts() -> Vec<(String, String, bool)> {
+        vec![
+            ("/job/input".into(), "/input".into(), true),
+            ("/job/output".into(), "/output".into(), false),
+            ("/data/objects/x.csv".into(), "/data/x.csv".into(), true),
+        ]
+    }
+
+    #[test]
+    fn backtest_container_flags_are_bounded_isolated_nonroot() {
+        let args = vec!["python".to_string(), "-m".to_string(), "worker.main".to_string()];
+        let full = docker_args(false, &mounts(), "strategy-lab-worker:local", "sl-run-x", &args);
+        let s = full.join(" ");
+        assert!(s.contains("strategy-lab-worker:local"));
+        assert!(s.contains("--network=none"), "runner must not use network: {s}");
+        assert!(s.contains(format!("--user={NONROOT_UID}:").as_str()), "{s}");
+        assert!(s.contains("--cap-drop=ALL"));
+        assert!(s.contains("--security-opt=no-new-privileges"));
+        assert!(s.contains("--read-only"));
+        assert!(s.contains("--pids-limit=128"));
+        assert!(s.contains("--memory=2g"));
+        assert!(s.contains("--cpus=2"));
+        assert!(s.contains("--tmpfs=/tmp:"));
+        assert!(!s.contains("/var/run/docker.sock"), "no docker socket in worker: {s}");
+        // mount directions preserved
+        assert!(s.contains("/job/input:/input:ro"));
+        assert!(s.contains("/job/output:/output"));
+        assert!(s.contains("/data/objects/x.csv:/data/x.csv:ro"));
+    }
+
+    #[test]
+    fn fetch_container_has_network_and_same_isolation() {
+        let full = docker_args(true, &[], "img", "sl-fetch-1", &["python".into()]);
+        let s = full.join(" ");
+        assert!(!s.contains("--network=none"));
+        assert!(s.contains("--cap-drop=ALL") && s.contains(CONCAT_USER));
+        assert!(s.contains("--rm --sig-proxy=false --name sl-fetch-1"));
+    }
+
+    #[test]
+    fn truncate_is_char_safe() {
+        let s = "中文内容";
+        let t = truncate(s, 4);
+        assert!(t.chars().count() <= 4);
+        assert_eq!(truncate("short", 100), "short");
+    }
+}