fix(data): ETF 身份契约与挂载权限修复（159399 QA 回归）

[问题原因]
前端手动录入表单保存的 market 字段是 SH/SZ/BJ 原文，但 worker 的
split_identity 只接受 market:"cn" 并按代码前缀推交易所，导致
market:"SZ" 的 159399 在发起任何网络请求前就报 unsupported_market。
同时 server 为非 root 容器（uid 65534）准备挂载时，完整文件挂载从不
被 chmod，且所有 chmod 失败都被静默吞掉，worker 容器读数据集可能
Permission denied 且不易定位。

[问题根因]
身份契约在 server 持久化层与 worker 适配层不一致；挂载权限处理是
best-effort 静默吞错，缺少最小作用域约束。

[修复方案]
worker/data.py: market 原样映射 IDENTITY_EXCHANGES（SH/SZ/BJ），
"cn" 保持代码前缀推断（SH=3/6/9 开头，否则 SZ），未知市场仍显式拒绝
绝不猜测；sina 数据源的警告文案改为只陈述可证实事实（无日期参数、
不除权、成交量单位为股且不换算），删除与其它 provider 的 100x 换算
断言。server/src/worker.rs: prepare_mounts 严格化——独立文件挂载仅在
"只读 + /data/ 前缀"范围内 chmod 0644，拒绝符号链接，目录挂载保持
0755/0777，所有 chmod 失败作为错误返回而非吞掉。tests/worker/
test_data.py 新增身份契约用例：SZ/SH 原文、cn 推断不变、SZ 直达
sina provider 的端到端 fetch。

[影响范围]
worker 数据获取链路与 server 任务编排；server/target/release 二进制
已含本改动（2026-09-17 构建）；pytest 数据模块全绿。

1 files changed, 134 insertions, 15 deletions

diff --git a/server/src/worker.rs b/server/src/worker.rs
index 8e28de5..d046277 100644
--- a/server/src/worker.rs
+++ b/server/src/worker.rs
@@ -194,31 +194,73 @@ pub async fn docker_available() -> bool {
         .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 {
+/// Mounts need world permissions for the non-root container (uid 65534).
+///
+/// Strict scope (leader review 2026-09-17):
+///  - standalone FILE mounts are touched ONLY when they are read-only mount
+///    of a worker data feed (dst starts with `/data/`) — never arbitrary
+///    host files; symlinks are rejected, chmod errors surface, nothing is
+///    silently swallowed.
+///  - directory mounts (job input/output dirs) still normalize perms; any
+///    failure is now returned instead of swallowed.
+fn prepare_mounts(mounts: &[(String, String, bool)]) -> AppResult<()> {
+    for (src, dst, ro) in mounts {
         let p = std::path::Path::new(src);
+        // Standalone file mounts (e.g. dataset objects bound directly to
+        // /data/<name> for backtests) must independently become world
+        // readable; they do not live under a prepare_mounts ro directory.
         if !p.is_dir() {
+            let md = std::fs::symlink_metadata(p)
+                .map_err(|e| crate::error::AppError::internal(format!("mount {src:?} unreachable: {e}")))?;
+            if md.file_type().is_symlink() {
+                return Err(crate::error::AppError::internal(format!(
+                    "symlinked mount rejected: {src}"
+                )));
+            }
+            let data_ro = *ro && dst.starts_with("/data/");
+            if data_ro {
+                if !p.is_file() {
+                    return Err(crate::error::AppError::internal(format!(
+                        "read-only data mount is not a regular file: {src}"
+                    )));
+                }
+                std::fs::set_permissions(p, std::fs::Permissions::from_mode(0o644)).map_err(
+                    |e| {
+                        crate::error::AppError::internal(format!(
+                            "cannot make data mount readable: {src}: {e}"
+                        ))
+                    },
+                )?;
+            }
+            // Anything else (non-/data or non-ro) is intentionally untouched.
             continue;
         }
         let mode = if *ro { 0o755 } else { 0o777 };
-        let _ = std::fs::set_permissions(p, std::fs::Permissions::from_mode(mode));
+        std::fs::set_permissions(p, std::fs::Permissions::from_mode(mode)).map_err(|e| {
+            crate::error::AppError::internal(format!("cannot set dir mount permissions: {src}: {e}"))
+        })?;
         // 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);
-                }
+            for e in std::fs::read_dir(p)
+                .map_err(|e| crate::error::AppError::internal(format!("cannot read mount dir {src}: {e}")))?
+                .flatten()
+            {
+                let fmode = if e.path().is_file() {
+                    std::fs::Permissions::from_mode(0o644)
+                } else {
+                    std::fs::Permissions::from_mode(0o755)
+                };
+                std::fs::set_permissions(e.path(), fmode).map_err(|err| {
+                    crate::error::AppError::internal(format!(
+                        "cannot fix ro mount entry {:?}: {err}",
+                        e.path()
+                    ))
+                })?;
             }
         }
     }
+    Ok(())
 }
 
 /// Run the worker image with a fixed container name so cancel maps to one
@@ -232,7 +274,9 @@ pub async fn run_named(
     timeout_secs: u64,
 ) -> AppResult<ContainerResult> {
     let cfg = &cx.cfg;
-    prepare_mounts(mounts);
+    prepare_mounts(mounts).map_err(|e| {
+        crate::error::AppError::internal(format!("worker mount preparation failed: {}", e.message))
+    })?;
     let full = docker_args(network, mounts, &cfg.worker_image, name, args);
     execute_docker(&full, name, timeout_secs).await
 }
@@ -480,3 +524,78 @@ mod tests {
         assert_eq!(truncate("short", 100), "short");
     }
 }
+
+
+
+#[cfg(test)]
+mod mount_perm_tests {
+    use super::prepare_mounts;
+    use std::os::unix::fs::PermissionsExt;
+
+    fn mode_of(p: &std::path::Path) -> u32 {
+        std::fs::metadata(p).unwrap().permissions().mode() & 0o777
+    }
+
+    /// 159399 candidate QA regression (2026-09-17): a run failed because the
+    /// directly-mounted dataset object file kept the server process' umask
+    /// perms (0o600) while the container runs as uid 65534 =>
+    /// PermissionError in the backtest worker. prepare_mounts already fixed
+    /// files *inside* ro directories but skipped standalone file mounts.
+    #[test]
+    fn standalone_data_file_mounts_become_world_readable() {
+        let td = tempfile::tempdir().unwrap();
+        let f = td.path().join("48c.csv");
+        std::fs::write(&f, b"date,open\n2026-01-02,1.0\n").unwrap();
+        std::fs::set_permissions(&f, std::fs::Permissions::from_mode(0o600)).unwrap();
+        prepare_mounts(&[(f.display().to_string(), "/data/48c.csv".into(), true)]).unwrap();
+        assert_eq!(mode_of(&f) & 0o004, 0o004, "others-read must be set on data mounts");
+    }
+
+    /// Narrow scope: file mounts that are NOT read-only /data feeds must be
+    /// left exactly as they are (no blanket chmod of arbitrary paths).
+    #[test]
+    fn non_data_file_mounts_are_untouched() {
+        let td = tempfile::tempdir().unwrap();
+        for (dst, ro) in [("/input/f.json", false), ("/data/x", false)] {
+            let f = td.path().join(format!("f{}.bin", dst.replace('/', "_")));
+            std::fs::write(&f, b"x").unwrap();
+            std::fs::set_permissions(&f, std::fs::Permissions::from_mode(0o600)).unwrap();
+            prepare_mounts(&[(f.display().to_string(), dst.into(), ro)]).unwrap();
+            assert_eq!(mode_of(&f) & 0o077, 0, "non-data file mount must be untouched: {}", dst);
+        }
+    }
+
+    /// Symlinks are rejected with an explicit error; the link target must not
+    /// be widened (no chmod via a link).
+    #[test]
+    fn symlinked_mount_is_rejected_not_chmodded() {
+        let td = tempfile::tempdir().unwrap();
+        let target = td.path().join("real-data.csv");
+        std::fs::write(&target, b"date\n2026-01-02\n").unwrap();
+        std::fs::set_permissions(&target, std::fs::Permissions::from_mode(0o600)).unwrap();
+        let link = td.path().join("data.csv");
+        std::os::unix::fs::symlink(&target, &link).unwrap();
+        let err = prepare_mounts(&[(
+            link.display().to_string(),
+            "/data/data.csv".into(),
+            true,
+        )])
+        .expect_err("symlink must be rejected");
+        assert_eq!(mode_of(&target) & 0o077, 0o000, "target must stay untouched");
+        assert!(err.message.contains("symlink"), "{:?}", err.message);
+    }
+
+    #[test]
+    fn directory_mounts_keep_fixing_files_inside_ro() {
+        let td = tempfile::tempdir().unwrap();
+        let d = td.path().join("input");
+        std::fs::create_dir_all(&d).unwrap();
+        let f = d.join("request.json");
+        std::fs::write(&f, b"{}").unwrap();
+        std::fs::set_permissions(&d, std::fs::Permissions::from_mode(0o700)).unwrap();
+        std::fs::set_permissions(&f, std::fs::Permissions::from_mode(0o600)).unwrap();
+        prepare_mounts(&[(d.display().to_string(), "/input".into(), true)]).unwrap();
+        assert_eq!(mode_of(&d) & 0o044, 0o044, "ro dir needs traversal");
+        assert_eq!(mode_of(&f) & 0o004, 0o004, "files inside ro dir need others-read");
+    }
+}