diff --git a/pipeline_service/__init__.py b/pipeline_service/__init__.py
index 0381f16..3695bc5 100644
--- a/pipeline_service/__init__.py
+++ b/pipeline_service/__init__.py
@@ -42,4 +42,14 @@ from .deploy_account import (
     sanitize_username,
 )
 
+# 工作环境管理（沙箱本地/远程切换 + 目录迁移）
+from .work_env import (
+    get_work_env,
+    set_work_env,
+    migrate_work_dir,
+    run_remote_sandbox,
+    run_in_work_env,
+    ensure_remote_bwrap,
+)
+
 __version__ = "3.5.0"
diff --git a/pipeline_service/init.py b/pipeline_service/init.py
index fe43c61..0320c51 100644
--- a/pipeline_service/init.py
+++ b/pipeline_service/init.py
@@ -648,6 +648,18 @@ def load_pipeline_service():
                     "UNIQUE KEY uk_user_id (user_id), KEY idx_status (status)"
                     ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4", {})
                 _debug("sd_deploy_accounts table ready")
+                # 工作环境表（沙箱本地/远程切换）
+                await sor.sqlExe(
+                    "CREATE TABLE IF NOT EXISTS sd_work_envs ("
+                    "id varchar(32) NOT NULL, owner_type varchar(10) NOT NULL,"
+                    "owner_id varchar(32) NOT NULL, mode varchar(10) NOT NULL DEFAULT 'local',"
+                    "remote_host varchar(200), remote_port int DEFAULT 22,"
+                    "remote_user varchar(100), remote_key_path varchar(500), remote_dir varchar(500),"
+                    "created_at datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,"
+                    "updated_at datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,"
+                    "PRIMARY KEY (id), UNIQUE KEY uk_owner (owner_type, owner_id)"
+                    ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4", {})
+                _debug("sd_work_envs table ready")
                 # agent_config columns
                 try:
                     await sor.sqlExe("ALTER TABLE pipelines ADD COLUMN agent_config text", {})
diff --git a/pipeline_service/work_env.py b/pipeline_service/work_env.py
new file mode 100644
index 0000000..7298ba7
--- /dev/null
+++ b/pipeline_service/work_env.py
@@ -0,0 +1,313 @@
+"""
+pipeline_service/work_env.py - 工作环境管理（沙箱的本地/远程切换 + 目录迁移）
+
+工作环境以「用户 / 机构」为单位组织，缺省本地模式（bwrap 沙箱）。
+远程模式 = 免密 SSH 到其他主机，远程主机同样部署 bwrap 走沙箱。
+
+目录层级约定：项目目录位于「工作目录」（逻辑账号 deploy_dir）之下，
+所以迁移只需同步工作目录顶层，即可覆盖所有项目内容。
+
+- get_work_env: 查询用户/机构工作环境（user 优先 → org 兜底 → 缺省 local）
+- set_work_env: 设置工作环境（切换 mode 时触发目录迁移）
+- migrate_work_dir: 目录迁移（rsync 本地 deploy_dir ↔ 远程 remote_dir）
+- run_remote_sandbox: 远程免密 SSH + bwrap 沙箱执行
+"""
+
+import os
+import shlex
+import shutil
+import asyncio
+import logging
+
+logger = logging.getLogger("pipeline.work_env")
+
+from .deploy_account import (
+    BWRAP, deploy_dir_for, sanitize_username, account_name_for,
+)
+
+# rsync 可执行文件（迁移必需）
+RSYNC = shutil.which("rsync")
+
+
+def _row_to_env(row) -> dict:
+    """DB 行 → 工作环境 dict。"""
+    return {
+        "mode": getattr(row, "mode", "local") or "local",
+        "remote_host": getattr(row, "remote_host", "") or "",
+        "remote_port": getattr(row, "remote_port", 22) or 22,
+        "remote_user": getattr(row, "remote_user", "") or "",
+        "remote_key_path": getattr(row, "remote_key_path", "") or "",
+        "remote_dir": getattr(row, "remote_dir", "") or "",
+    }
+
+
+async def get_work_env(sor, user_id: str, org_id: str = "") -> dict:
+    """获取工作环境。优先级：user 级 → org 级 → 缺省 local。"""
+    if user_id:
+        recs = await sor.sqlExe(
+            "SELECT * FROM sd_work_envs WHERE owner_type='user' AND owner_id=${o}$ LIMIT 1",
+            {"o": user_id})
+        if recs:
+            return _row_to_env(recs[0])
+    if org_id:
+        recs = await sor.sqlExe(
+            "SELECT * FROM sd_work_envs WHERE owner_type='org' AND owner_id=${o}$ LIMIT 1",
+            {"o": org_id})
+        if recs:
+            return _row_to_env(recs[0])
+    return {"mode": "local", "remote_host": "", "remote_port": 22,
+            "remote_user": "", "remote_key_path": "", "remote_dir": ""}
+
+
+def _ssh_target(env: dict) -> str:
+    """构造 scp/rsync 的 user@host 目标。"""
+    user = env.get("remote_user", "")
+    host = env.get("remote_host", "")
+    if user:
+        return f"{user}@{host}"
+    return host
+
+
+def _ssh_common_args(env: dict) -> list:
+    """ssh/scp/rsync 共用的 SSH 参数。"""
+    args = ["-o", "StrictHostKeyChecking=no", "-o", "ConnectTimeout=10"]
+    key = env.get("remote_key_path", "")
+    if key:
+        args += ["-i", key]
+    port = env.get("remote_port", 22) or 22
+    args += ["-p", str(port)]
+    return args
+
+
+# ═══════════════════════════════════════════════════════════
+# 目录迁移
+# ═══════════════════════════════════════════════════════════
+
+async def migrate_work_dir(account_name: str, from_mode: str, to_mode: str, env: dict) -> dict:
+    """把工作目录内容从旧环境迁移到新环境。
+
+    from_mode/to_mode ∈ {'local', 'remote'}
+    - local → remote: rsync 本地 deploy_dir → remote_dir
+    - remote → local: rsync remote_dir → 本地 deploy_dir
+    """
+    local_dir = deploy_dir_for(account_name)
+    if to_mode == "local" and not os.path.isdir(local_dir):
+        os.makedirs(local_dir, exist_ok=True)
+
+    if not RSYNC:
+        return {"ok": False, "error": "rsync 不可用，无法迁移"}
+
+    # 构造 rsync 命令
+    try:
+        if from_mode == "local" and to_mode == "remote":
+            # 本地 → 远程
+            remote_dir = env.get("remote_dir", "")
+            if not remote_dir:
+                return {"ok": False, "error": "缺少远程目录 remote_dir"}
+            target = _ssh_target(env) + ":" + remote_dir
+            cmd = ["rsync", "-az", "--delete", "-e",
+                   "ssh " + " ".join(_ssh_common_args(env)),
+                   local_dir.rstrip("/") + "/", target.rstrip("/") + "/"]
+        elif from_mode == "remote" and to_mode == "local":
+            # 远程 → 本地
+            remote_dir = env.get("remote_dir", "")
+            if not remote_dir:
+                return {"ok": False, "error": "缺少远程目录 remote_dir"}
+            src = _ssh_target(env) + ":" + remote_dir
+            cmd = ["rsync", "-az", "--delete", "-e",
+                   "ssh " + " ".join(_ssh_common_args(env)),
+                   src.rstrip("/") + "/", local_dir.rstrip("/") + "/"]
+        else:
+            # 同模式或无效，无需迁移
+            return {"ok": True, "migrated": False}
+
+        proc = await asyncio.create_subprocess_exec(
+            *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
+        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=300)
+        if proc.returncode != 0:
+            return {"ok": False, "error": "迁移失败: " + stderr.decode("utf-8", "replace")[-1000:]}
+        return {"ok": True, "migrated": True,
+                "detail": stdout.decode("utf-8", "replace")[-500:]}
+    except asyncio.TimeoutError:
+        return {"ok": False, "error": "迁移超时（300s）"}
+    except Exception as e:
+        return {"ok": False, "error": str(e)[:500]}
+
+
+# ═══════════════════════════════════════════════════════════
+# 设置工作环境
+# ═══════════════════════════════════════════════════════════
+
+async def set_work_env(sor, owner_type: str, owner_id: str, mode: str,
+                       remote_config: dict = None, account_name: str = None) -> dict:
+    """设置工作环境。切换 mode 时触发目录迁移。
+
+    Returns: {"ok": bool, "mode": str, "migrated": {...}}
+    """
+    if owner_type not in ("user", "org"):
+        return {"ok": False, "error": "owner_type 必须是 user 或 org"}
+    if mode not in ("local", "remote"):
+        return {"ok": False, "error": "mode 必须是 local 或 remote"}
+
+    remote_config = remote_config or {}
+    if mode == "remote":
+        if not remote_config.get("remote_host"):
+            return {"ok": False, "error": "远程模式需要 remote_host"}
+        if not remote_config.get("remote_user"):
+            return {"ok": False, "error": "远程模式需要 remote_user"}
+        if not remote_config.get("remote_dir"):
+            return {"ok": False, "error": "远程模式需要 remote_dir"}
+
+    # 查旧环境（用于判断是否需要迁移）
+    old_recs = await sor.sqlExe(
+        "SELECT mode FROM sd_work_envs WHERE owner_type=${t}$ AND owner_id=${o}$ LIMIT 1",
+        {"t": owner_type, "o": owner_id})
+    old_mode = getattr(old_recs[0], "mode", "local") if old_recs else "local"
+
+    # upsert
+    from appPublic.uniqueID import getID
+    data = {
+        "owner_type": owner_type,
+        "owner_id": owner_id,
+        "mode": mode,
+        "remote_host": remote_config.get("remote_host", ""),
+        "remote_port": int(remote_config.get("remote_port", 22) or 22),
+        "remote_user": remote_config.get("remote_user", ""),
+        "remote_key_path": remote_config.get("remote_key_path", ""),
+        "remote_dir": remote_config.get("remote_dir", ""),
+    }
+    if old_recs:
+        await sor.sqlExe(
+            "UPDATE sd_work_envs SET mode=${mode}$, remote_host=${h}$, remote_port=${p}$, "
+            "remote_user=${u}$, remote_key_path=${k}$, remote_dir=${d}$ "
+            "WHERE owner_type=${t}$ AND owner_id=${o}$",
+            {"mode": mode, "h": data["remote_host"], "p": data["remote_port"],
+             "u": data["remote_user"], "k": data["remote_key_path"],
+             "d": data["remote_dir"], "t": owner_type, "o": owner_id})
+    else:
+        data["id"] = getID()
+        await sor.C("sd_work_envs", data)
+
+    # 触发迁移（模式切换 + 提供了 account_name 时）
+    migrated = {"migrated": False}
+    if account_name and old_mode != mode:
+        env = _row_to_env_from_dict(data)
+        migrated = await migrate_work_dir(account_name, old_mode, mode, env)
+
+    return {"ok": True, "mode": mode, "old_mode": old_mode, "migration": migrated}
+
+
+def _row_to_env_from_dict(data: dict) -> dict:
+    return {
+        "mode": data.get("mode", "local"),
+        "remote_host": data.get("remote_host", ""),
+        "remote_port": data.get("remote_port", 22),
+        "remote_user": data.get("remote_user", ""),
+        "remote_key_path": data.get("remote_key_path", ""),
+        "remote_dir": data.get("remote_dir", ""),
+    }
+
+
+# ═══════════════════════════════════════════════════════════
+# 远程沙箱执行
+# ═══════════════════════════════════════════════════════════
+
+def _remote_bwrap_cmd(env: dict, deploy_dir: str, command: str, workdir: str = "") -> list:
+    """构造远程主机的 bwrap 沙箱命令（通过 SSH 执行）。"""
+    # 远程主机上 bwrap 的路径（假设部署在 ~/bin/bwrap 或系统 PATH）
+    remote_bwrap = "bwrap"
+    parts = [
+        remote_bwrap,
+        "--unshare-user", "--unshare-pid", "--unshare-ipc", "--unshare-uts",
+        "--die-with-parent",
+        "--ro-bind", "/usr", "/usr",
+        "--ro-bind", "/bin", "/bin",
+        "--ro-bind", "/sbin", "/sbin",
+        "--ro-bind", "/lib", "/lib",
+        "--ro-bind", "/lib64", "/lib64",
+        "--ro-bind", "/etc", "/etc",
+        "--proc", "/proc",
+        "--dev", "/dev",
+        "--tmpfs", "/tmp",
+        "--bind", deploy_dir, "/home",
+        "--chdir", "/home",
+        "--setenv", "HOME", "/home",
+        "--", "bash", "-c", command,
+    ]
+    return parts
+
+
+async def run_remote_sandbox(env: dict, deploy_dir: str, command: str,
+                             workdir: str = "", timeout: int = 120) -> dict:
+    """远程免密 SSH + bwrap 沙箱执行命令。"""
+    ssh_args = _ssh_common_args(env)
+    target = _ssh_target(env)
+
+    bwrap_parts = _remote_bwrap_cmd(env, deploy_dir, command, workdir)
+    inner = " ".join(shlex.quote(p) for p in bwrap_parts)
+
+    ssh_cmd = ["ssh"] + ssh_args + [target, inner]
+
+    try:
+        proc = await asyncio.create_subprocess_exec(
+            *ssh_cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
+        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout + 30)
+    except asyncio.TimeoutError:
+        return {"rc": -1, "stdout": "", "stderr": f"远程命令超时（{timeout}s）", "sandbox": True, "remote": True}
+    except Exception as e:
+        return {"rc": -1, "stdout": "", "stderr": f"SSH 连接失败: {str(e)[:300]}", "sandbox": True, "remote": True}
+
+    return {
+        "rc": proc.returncode or 0,
+        "stdout": stdout.decode("utf-8", "replace")[-8000:],
+        "stderr": stderr.decode("utf-8", "replace")[-4000:],
+        "sandbox": True,
+        "remote": True,
+    }
+
+
+async def run_in_work_env(sor, account_name: str, user_id: str, org_id: str,
+                          command: str, workdir: str = "", timeout: int = 120,
+                          network: str = "shared") -> dict:
+    """按工作环境 mode 分发执行：本地 bwrap 或远程 SSH+bwrap。"""
+    env = await get_work_env(sor, user_id, org_id)
+    if env.get("mode") == "remote":
+        remote_dir = env.get("remote_dir", "")
+        if not remote_dir:
+            return {"rc": -1, "stdout": "", "stderr": "远程模式缺少 remote_dir",
+                    "sandbox": True, "remote": True}
+        return await run_remote_sandbox(env, remote_dir, command, workdir, timeout)
+    from .deploy_account import run_in_sandbox
+    return await run_in_sandbox(sor, account_name, command, workdir, timeout, network)
+
+
+# ═══════════════════════════════════════════════════════════
+# 远程 bwrap 部署（远程主机装 bwrap，零 root）
+# ═══════════════════════════════════════════════════════════
+
+async def ensure_remote_bwrap(env: dict) -> dict:
+    """确保远程主机已安装 bwrap（零 root：apt download + dpkg -x）。"""
+    ssh_args = _ssh_common_args(env)
+    target = _ssh_target(env)
+
+    script = (
+        'mkdir -p ~/bin && '
+        'if ! command -v bwrap >/dev/null 2>&1 && [ ! -x ~/bin/bwrap ]; then '
+        'tmpd=$(mktemp -d) && (cd "$tmpd" && apt download bubblewrap >/dev/null 2>&1 '
+        '&& dpkg -x bubblewrap*.deb . && cp usr/bin/bwrap ~/bin/bwrap) 2>/dev/null; '
+        'rm -rf "$tmpd"; fi; '
+        'if command -v bwrap >/dev/null 2>&1; then echo PATH_BWRAP; '
+        'elif [ -x ~/bin/bwrap ]; then echo HOME_BWRAP; '
+        'else echo NO_BWRAP; fi'
+    )
+    ssh_cmd = ["ssh"] + ssh_args + [target, script]
+    try:
+        proc = await asyncio.create_subprocess_exec(
+            *ssh_cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
+        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=60)
+        out = stdout.decode("utf-8", "replace").strip()
+        if "NO_BWRAP" in out:
+            return {"ok": False, "error": "远程主机无法获取 bwrap: " + stderr.decode("utf-8", "replace")[-300:]}
+        return {"ok": True, "location": out.strip().splitlines()[-1] if out else "unknown"}
+    except Exception as e:
+        return {"ok": False, "error": str(e)[:300]}