diff --git a/.env.template b/.env.template
index 9115d6f7..4fc1da84 100644
--- a/.env.template
+++ b/.env.template
@@ -12,11 +12,25 @@ SANDBOX_PROVISIONER_URL=http://sandbox-provisioner:8002
 SANDBOX_VIRTUAL_PATH_PREFIX=/mnt/user-data
 SANDBOX_EXEC_TIMEOUT_SECONDS=180
 SANDBOX_MAX_OUTPUT_BYTES=262144
-# sandbox-provisioner backend: memory | kubernetes
-SANDBOX_PROVISIONER_BACKEND=memory
+SANDBOX_KEEPALIVE_INTERVAL_SECONDS=30
+SANDBOX_IDLE_TIMEOUT_SECONDS=120
+SANDBOX_IDLE_CHECK_INTERVAL_SECONDS=10
+# sandbox-provisioner backend: memory | local | docker | kubernetes
+SANDBOX_PROVISIONER_BACKEND=local
+# local/docker backend defaults (deerFlow local_backend style)
+# SANDBOX_CONTAINER_PORT=8080
+# SANDBOX_DOCKER_SANDBOX_HOST=host.docker.internal
+# Docker provisioner options (used when SANDBOX_PROVISIONER_BACKEND=local/docker)
+# SANDBOX_DOCKER_NETWORK=yuxi-know_app-network
+# SANDBOX_DOCKER_THREADS_HOST_PATH=
+# SANDBOX_DOCKER_SKILLS_HOST_PATH=
+# SANDBOX_DOCKER_SANDBOX_PREFIX=yuxi-sandbox
+# Optional proxy for sandbox-provisioner container
+# SANDBOX_HTTP_PROXY=http://host.docker.internal:7897
+# SANDBOX_HTTPS_PROXY=http://host.docker.internal:7897
 # K8s provisioner options (used when SANDBOX_PROVISIONER_BACKEND=kubernetes)
 # SANDBOX_K8S_NAMESPACE=yuxi-know
-# SANDBOX_IMAGE=ghcr.io/bytedance/deer-flow-sandbox:latest
+# SANDBOX_IMAGE=enterprise-public-cn-beijing.cr.volces.com/vefaas-public/all-in-one-sandbox:latest
 # SANDBOX_SKILLS_HOST_PATH=/app/saves/skills
 # SANDBOX_THREADS_HOST_PATH=/app/saves/threads
 # SANDBOX_NODE_HOST=host.docker.internal
diff --git a/AGENTS.md b/AGENTS.md
index 3706ab94..26937877 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -37,7 +37,7 @@ make format        # 格式化代码
 # 直接在容器内执行命令
 docker compose exec api uv run python test/your_script.py  # 放在 test 文件夹
 ```
-
+安装依赖太慢时可以使用代理端口 7897
 注意：
 - Python 代码要符合 Python 的规范，符合 pythonic 风格
 - 尽量使用较新的语法，避免使用旧版本的语法（版本兼容到 3.12+）
diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml
index c6f18f47..f4c10315 100644
--- a/docker-compose.prod.yml
+++ b/docker-compose.prod.yml
@@ -37,8 +37,8 @@ services:
       - SANDBOX_VIRTUAL_PATH_PREFIX=${SANDBOX_VIRTUAL_PATH_PREFIX:-/mnt/user-data}
       - SANDBOX_EXEC_TIMEOUT_SECONDS=${SANDBOX_EXEC_TIMEOUT_SECONDS:-180}
       - SANDBOX_MAX_OUTPUT_BYTES=${SANDBOX_MAX_OUTPUT_BYTES:-262144}
-      - NO_PROXY=localhost,127.0.0.1,milvus,graph,milvus-minio,milvus-etcd,etcd,minio,mineru,paddlex,sandbox-provisioner,api.siliconflow.cn
-      - no_proxy=localhost,127.0.0.1,milvus,graph,milvus-minio,milvus-etcd,etcd,minio,mineru,paddlex,sandbox-provisioner,api.siliconflow.cn
+      - NO_PROXY=localhost,127.0.0.1,host.docker.internal,milvus,graph,milvus-minio,milvus-etcd,etcd,minio,mineru,paddlex,sandbox-provisioner,api.siliconflow.cn
+      - no_proxy=localhost,127.0.0.1,host.docker.internal,milvus,graph,milvus-minio,milvus-etcd,etcd,minio,mineru,paddlex,sandbox-provisioner,api.siliconflow.cn
     command: uv run --no-dev uvicorn server.main:app --host 0.0.0.0 --port 5050
     restart: unless-stopped
     healthcheck:
@@ -97,8 +97,8 @@ services:
       - SANDBOX_VIRTUAL_PATH_PREFIX=${SANDBOX_VIRTUAL_PATH_PREFIX:-/mnt/user-data}
       - SANDBOX_EXEC_TIMEOUT_SECONDS=${SANDBOX_EXEC_TIMEOUT_SECONDS:-180}
       - SANDBOX_MAX_OUTPUT_BYTES=${SANDBOX_MAX_OUTPUT_BYTES:-262144}
-      - NO_PROXY=localhost,127.0.0.1,milvus,graph,milvus-minio,milvus-etcd,etcd,minio,mineru,paddlex,sandbox-provisioner,api.siliconflow.cn
-      - no_proxy=localhost,127.0.0.1,milvus,graph,milvus-minio,milvus-etcd,etcd,minio,mineru,paddlex,sandbox-provisioner,api.siliconflow.cn
+      - NO_PROXY=localhost,127.0.0.1,host.docker.internal,milvus,graph,milvus-minio,milvus-etcd,etcd,minio,mineru,paddlex,sandbox-provisioner,api.siliconflow.cn
+      - no_proxy=localhost,127.0.0.1,host.docker.internal,milvus,graph,milvus-minio,milvus-etcd,etcd,minio,mineru,paddlex,sandbox-provisioner,api.siliconflow.cn
     command: uv run --no-dev arq server.worker_main.WorkerSettings
     restart: unless-stopped
     depends_on:
@@ -120,19 +120,29 @@ services:
     container_name: sandbox-provisioner
     volumes:
       - ./saves:/app/saves
+      - /var/run/docker.sock:/var/run/docker.sock
     networks:
       - app-network
     extra_hosts:
       - "host.docker.internal:host-gateway"
     environment:
-      - PROVISIONER_BACKEND=${SANDBOX_PROVISIONER_BACKEND:-memory}
+      - PROVISIONER_BACKEND=${SANDBOX_PROVISIONER_BACKEND:-local}
       - K8S_NAMESPACE=${SANDBOX_K8S_NAMESPACE:-yuxi-know}
-      - SANDBOX_IMAGE=${SANDBOX_IMAGE:-ghcr.io/bytedance/deer-flow-sandbox:latest}
+      - SANDBOX_IMAGE=${SANDBOX_IMAGE:-enterprise-public-cn-beijing.cr.volces.com/vefaas-public/all-in-one-sandbox:latest}
+      - SANDBOX_CONTAINER_PORT=${SANDBOX_CONTAINER_PORT:-8080}
       - SKILLS_HOST_PATH=${SANDBOX_SKILLS_HOST_PATH:-/app/saves/skills}
       - THREADS_HOST_PATH=${SANDBOX_THREADS_HOST_PATH:-/app/saves/threads}
       - NODE_HOST=${SANDBOX_NODE_HOST:-host.docker.internal}
       - KUBECONFIG_PATH=${KUBECONFIG_PATH:-}
       - MEMORY_SANDBOX_URL_TEMPLATE=${MEMORY_SANDBOX_URL_TEMPLATE:-http://agent-sandbox:8000}
+      - DOCKER_NETWORK=${SANDBOX_DOCKER_NETWORK:-yuxi-know_app-network}
+      - DOCKER_THREADS_HOST_PATH=${SANDBOX_DOCKER_THREADS_HOST_PATH:-}
+      - DOCKER_SKILLS_HOST_PATH=${SANDBOX_DOCKER_SKILLS_HOST_PATH:-}
+      - DOCKER_SANDBOX_PREFIX=${SANDBOX_DOCKER_SANDBOX_PREFIX:-yuxi-sandbox}
+      - DOCKER_SANDBOX_HOST=${SANDBOX_DOCKER_SANDBOX_HOST:-host.docker.internal}
+      - SANDBOX_IDLE_TIMEOUT_SECONDS=${SANDBOX_IDLE_TIMEOUT_SECONDS:-120}
+      - SANDBOX_IDLE_CHECK_INTERVAL_SECONDS=${SANDBOX_IDLE_CHECK_INTERVAL_SECONDS:-10}
+      - SANDBOX_EXEC_TIMEOUT_SECONDS=${SANDBOX_EXEC_TIMEOUT_SECONDS:-180}
     healthcheck:
       test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8002/health').read()"]
       interval: 10s
diff --git a/docker-compose.yml b/docker-compose.yml
index 3648323b..f262905d 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -46,8 +46,8 @@ services:
       - SANDBOX_VIRTUAL_PATH_PREFIX=${SANDBOX_VIRTUAL_PATH_PREFIX:-/mnt/user-data}
       - SANDBOX_EXEC_TIMEOUT_SECONDS=${SANDBOX_EXEC_TIMEOUT_SECONDS:-180}
       - SANDBOX_MAX_OUTPUT_BYTES=${SANDBOX_MAX_OUTPUT_BYTES:-262144}
-      - NO_PROXY=localhost,127.0.0.1,milvus,graph,milvus-minio,milvus-etcd-dev,etcd,minio,mineru,paddlex,sandbox-provisioner,api.siliconflow.cn
-      - no_proxy=localhost,127.0.0.1,milvus,graph,milvus-minio,milvus-etcd-dev,etcd,minio,mineru,paddlex,sandbox-provisioner,api.siliconflow.cn
+      - NO_PROXY=localhost,127.0.0.1,host.docker.internal,milvus,graph,milvus-minio,milvus-etcd-dev,etcd,minio,mineru,paddlex,sandbox-provisioner,api.siliconflow.cn
+      - no_proxy=localhost,127.0.0.1,host.docker.internal,milvus,graph,milvus-minio,milvus-etcd-dev,etcd,minio,mineru,paddlex,sandbox-provisioner,api.siliconflow.cn
     # endregion api_envs
     command: uv run --no-dev uvicorn server.main:app --host 0.0.0.0 --port 5050 --reload
     restart: unless-stopped
@@ -115,8 +115,8 @@ services:
       - SANDBOX_VIRTUAL_PATH_PREFIX=${SANDBOX_VIRTUAL_PATH_PREFIX:-/mnt/user-data}
       - SANDBOX_EXEC_TIMEOUT_SECONDS=${SANDBOX_EXEC_TIMEOUT_SECONDS:-180}
       - SANDBOX_MAX_OUTPUT_BYTES=${SANDBOX_MAX_OUTPUT_BYTES:-262144}
-      - NO_PROXY=localhost,127.0.0.1,milvus,graph,milvus-minio,milvus-etcd-dev,etcd,minio,mineru,paddlex,sandbox-provisioner,api.siliconflow.cn
-      - no_proxy=localhost,127.0.0.1,milvus,graph,milvus-minio,milvus-etcd-dev,etcd,minio,mineru,paddlex,sandbox-provisioner,api.siliconflow.cn
+      - NO_PROXY=localhost,127.0.0.1,host.docker.internal,milvus,graph,milvus-minio,milvus-etcd-dev,etcd,minio,mineru,paddlex,sandbox-provisioner,api.siliconflow.cn
+      - no_proxy=localhost,127.0.0.1,host.docker.internal,milvus,graph,milvus-minio,milvus-etcd-dev,etcd,minio,mineru,paddlex,sandbox-provisioner,api.siliconflow.cn
     command: uv run --no-dev arq server.worker_main.WorkerSettings
     restart: unless-stopped
     depends_on:
@@ -138,19 +138,36 @@ services:
     container_name: sandbox-provisioner
     volumes:
       - ./saves:/app/saves
+      - ./docker/sandbox_provisioner/app.py:/app/app.py:ro
+      - /var/run/docker.sock:/var/run/docker.sock
     networks:
       - app-network
     extra_hosts:
       - "host.docker.internal:host-gateway"
     environment:
-      - PROVISIONER_BACKEND=${SANDBOX_PROVISIONER_BACKEND:-memory}
+      - PROVISIONER_BACKEND=${SANDBOX_PROVISIONER_BACKEND:-local}
       - K8S_NAMESPACE=${SANDBOX_K8S_NAMESPACE:-yuxi-know}
-      - SANDBOX_IMAGE=${SANDBOX_IMAGE:-ghcr.io/bytedance/deer-flow-sandbox:latest}
+      - SANDBOX_IMAGE=${SANDBOX_IMAGE:-enterprise-public-cn-beijing.cr.volces.com/vefaas-public/all-in-one-sandbox:latest}
+      - SANDBOX_CONTAINER_PORT=${SANDBOX_CONTAINER_PORT:-8080}
       - SKILLS_HOST_PATH=${SANDBOX_SKILLS_HOST_PATH:-/app/saves/skills}
       - THREADS_HOST_PATH=${SANDBOX_THREADS_HOST_PATH:-/app/saves/threads}
       - NODE_HOST=${SANDBOX_NODE_HOST:-host.docker.internal}
       - KUBECONFIG_PATH=${KUBECONFIG_PATH:-}
       - MEMORY_SANDBOX_URL_TEMPLATE=${MEMORY_SANDBOX_URL_TEMPLATE:-http://agent-sandbox:8000}
+      - HTTP_PROXY=${SANDBOX_HTTP_PROXY:-}
+      - HTTPS_PROXY=${SANDBOX_HTTPS_PROXY:-}
+      - NO_PROXY=localhost,127.0.0.1,host.docker.internal
+      - DOCKER_NETWORK=${SANDBOX_DOCKER_NETWORK:-yuxi-know_app-network}
+      - DOCKER_THREADS_HOST_PATH=${SANDBOX_DOCKER_THREADS_HOST_PATH:-}
+      - DOCKER_SKILLS_HOST_PATH=${SANDBOX_DOCKER_SKILLS_HOST_PATH:-}
+      - DOCKER_SANDBOX_PREFIX=${SANDBOX_DOCKER_SANDBOX_PREFIX:-yuxi-sandbox}
+      - DOCKER_SANDBOX_HOST=${SANDBOX_DOCKER_SANDBOX_HOST:-host.docker.internal}
+      - SANDBOX_IDLE_TIMEOUT_SECONDS=${SANDBOX_IDLE_TIMEOUT_SECONDS:-120}
+      - SANDBOX_IDLE_CHECK_INTERVAL_SECONDS=${SANDBOX_IDLE_CHECK_INTERVAL_SECONDS:-10}
+      - SANDBOX_EXEC_TIMEOUT_SECONDS=${SANDBOX_EXEC_TIMEOUT_SECONDS:-180}
+    command: >
+      sh -lc "python -c 'import docker' >/dev/null 2>&1 || pip install --no-cache-dir 'docker>=7.1.0';
+      uvicorn app:app --host 0.0.0.0 --port 8002"
     healthcheck:
       test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8002/health').read()"]
       interval: 10s
diff --git a/docker/sandbox_provisioner/app.py b/docker/sandbox_provisioner/app.py
index 2c8887be..87a957e5 100644
--- a/docker/sandbox_provisioner/app.py
+++ b/docker/sandbox_provisioner/app.py
@@ -1,12 +1,19 @@
 from __future__ import annotations
 
+import logging
 import os
 import threading
+import time
+from contextlib import asynccontextmanager
 from dataclasses import dataclass
+from pathlib import Path
+from urllib import request
 
 from fastapi import FastAPI, HTTPException
 from pydantic import BaseModel
 
+logger = logging.getLogger(__name__)
+
 
 class CreateSandboxRequest(BaseModel):
     sandbox_id: str
@@ -24,6 +31,12 @@ class DeleteSandboxResponse(BaseModel):
     sandbox_id: str
 
 
+class TouchSandboxResponse(BaseModel):
+    ok: bool
+    sandbox_id: str
+    status: str | None = None
+
+
 class ListSandboxesResponse(BaseModel):
     sandboxes: list[SandboxResponse]
     count: int
@@ -75,17 +88,243 @@ class MemoryProvisionerBackend:
             self._records.pop(sandbox_id, None)
 
 
+def wait_for_sandbox_ready(sandbox_url: str, timeout_seconds: int = 30) -> bool:
+    deadline = time.time() + timeout_seconds
+    opener = request.build_opener(request.ProxyHandler({}))
+    while time.time() < deadline:
+        try:
+            with opener.open(f"{sandbox_url.rstrip('/')}/v1/sandbox", timeout=3) as response:
+                status_code = getattr(response, "status", 200)
+            if status_code == 200:
+                return True
+        except Exception:
+            pass
+        time.sleep(1)
+    return False
+
+
+class LocalContainerProvisionerBackend:
+    def __init__(self):
+        import docker
+        from docker.errors import DockerException
+
+        self._docker = docker
+        self._lock = threading.Lock()
+        self._container_port = int(os.getenv("SANDBOX_CONTAINER_PORT", "8080"))
+        self._sandbox_image = os.getenv(
+            "SANDBOX_IMAGE",
+            "enterprise-public-cn-beijing.cr.volces.com/vefaas-public/all-in-one-sandbox:latest",
+        )
+        self._network = os.getenv("DOCKER_NETWORK")
+        self._threads_host_path = os.getenv("DOCKER_THREADS_HOST_PATH")
+        self._skills_host_path = os.getenv("DOCKER_SKILLS_HOST_PATH")
+        self._container_prefix = os.getenv("DOCKER_SANDBOX_PREFIX", "yuxi-sandbox")
+        self._sandbox_host = os.getenv("DOCKER_SANDBOX_HOST", "host.docker.internal")
+        self._health_timeout_seconds = int(os.getenv("SANDBOX_HEALTH_TIMEOUT_SECONDS", "30"))
+
+        try:
+            self._client = docker.from_env()
+            self._client.ping()
+        except DockerException as exc:
+            raise RuntimeError(f"docker backend unavailable: {exc}") from exc
+
+        self._resolve_host_paths()
+
+    @staticmethod
+    def _validate_thread_id(thread_id: str) -> str:
+        candidate = str(thread_id or "").strip()
+        if not candidate:
+            raise ValueError("thread_id is required")
+        if any(ch in candidate for ch in ("/", "\\", "\x00")):
+            raise ValueError("thread_id must be a single safe path segment")
+        if candidate in {".", ".."} or ".." in candidate:
+            raise ValueError("thread_id contains invalid path traversal sequence")
+        return candidate
+
+    @staticmethod
+    def _sanitize_id(value: str) -> str:
+        sanitized = "".join(ch if ch.isalnum() or ch in "-_" else "-" for ch in value.strip().lower())
+        return sanitized[:48] or "sandbox"
+
+    def _container_name(self, sandbox_id: str) -> str:
+        return f"{self._container_prefix}-{self._sanitize_id(sandbox_id)}"
+
+    def _resolve_host_paths(self) -> None:
+        if self._threads_host_path and self._skills_host_path:
+            return
+
+        container_id = os.getenv("HOSTNAME", "").strip()
+        if not container_id:
+            raise RuntimeError("HOSTNAME is required to infer docker backend host paths")
+
+        inspected = self._client.api.inspect_container(container_id)
+        mounts = inspected.get("Mounts") or []
+
+        saves_source = None
+        for mount in mounts:
+            destination = (mount.get("Destination") or "").rstrip("/")
+            if destination == "/app/saves":
+                saves_source = mount.get("Source")
+                break
+
+        if not saves_source:
+            raise RuntimeError("cannot infer host path for /app/saves mount")
+
+        base = Path(saves_source)
+        if not self._threads_host_path:
+            self._threads_host_path = str(base / "threads")
+        if not self._skills_host_path:
+            self._skills_host_path = str(base / "skills")
+
+    def _host_port_for(self, container) -> int | None:
+        ports = (container.attrs.get("NetworkSettings") or {}).get("Ports") or {}
+        bindings = ports.get(f"{self._container_port}/tcp")
+        if not bindings:
+            return None
+        host_port = bindings[0].get("HostPort")
+        if not host_port:
+            return None
+        return int(host_port)
+
+    def _sandbox_url(self, host_port: int) -> str:
+        return f"http://{self._sandbox_host}:{host_port}"
+
+    def _to_record(self, container, sandbox_id: str) -> SandboxRecord:
+        state = (container.attrs.get("State") or {}).get("Status")
+        host_port = self._host_port_for(container)
+        sandbox_url = self._sandbox_url(host_port) if host_port is not None else ""
+        return SandboxRecord(
+            sandbox_id=sandbox_id,
+            sandbox_url=sandbox_url,
+            status=state or "unknown",
+        )
+
+    @staticmethod
+    def _ensure_user_data_writable(container) -> None:
+        cmd = (
+            "sh -lc "
+            "\"mkdir -p /mnt/user-data/workspace /mnt/user-data/uploads /mnt/user-data/outputs "
+            "&& chmod -R a+rwX /mnt/user-data\""
+        )
+        result = container.exec_run(cmd, user="0:0")
+        if result.exit_code != 0:
+            output = result.output.decode("utf-8", errors="ignore") if isinstance(result.output, bytes) else str(result.output)
+            raise RuntimeError(f"failed to ensure writable thread user-data mount: {output}")
+
+    def _get_container(self, sandbox_id: str):
+        from docker.errors import NotFound
+
+        name = self._container_name(sandbox_id)
+        try:
+            return self._client.containers.get(name)
+        except NotFound:
+            return None
+
+    def create(self, sandbox_id: str, thread_id: str) -> SandboxRecord:
+        with self._lock:
+            safe_thread_id = self._validate_thread_id(thread_id)
+            existing = self._get_container(sandbox_id)
+            if existing is not None:
+                if existing.status != "running":
+                    existing.start()
+                    existing.reload()
+                self._ensure_user_data_writable(existing)
+                record = self._to_record(existing, sandbox_id)
+                if not record.sandbox_url:
+                    raise RuntimeError(f"sandbox {sandbox_id} has no mapped host port")
+                if not wait_for_sandbox_ready(record.sandbox_url, timeout_seconds=self._health_timeout_seconds):
+                    raise RuntimeError(f"sandbox {sandbox_id} is not ready at {record.sandbox_url}")
+                return record
+
+            threads_root = Path(self._threads_host_path).resolve()
+            thread_user_data = (threads_root / safe_thread_id / "user-data").resolve()
+            try:
+                thread_user_data.relative_to(threads_root)
+            except ValueError as exc:
+                raise ValueError("thread_id resolved outside threads host root") from exc
+            thread_user_data.mkdir(parents=True, exist_ok=True)
+
+            skills_path = Path(self._skills_host_path)
+            skills_path.mkdir(parents=True, exist_ok=True)
+
+            container_name = self._container_name(sandbox_id)
+            run_kwargs = {
+                "name": container_name,
+                "detach": True,
+                "labels": {
+                    "app": "yuxi-sandbox",
+                    "sandbox-id": sandbox_id,
+                    "thread-id": thread_id,
+                    "managed-by": "yuxi-sandbox-provisioner",
+                },
+                "volumes": {
+                    str(thread_user_data): {"bind": "/mnt/user-data", "mode": "rw"},
+                    str(skills_path): {"bind": "/mnt/skills", "mode": "ro"},
+                },
+                "ports": {f"{self._container_port}/tcp": None},
+                "security_opt": ["seccomp=unconfined"],
+            }
+            if self._network:
+                run_kwargs["network"] = self._network
+
+            container = self._client.containers.run(self._sandbox_image, **run_kwargs)
+            container.reload()
+            self._ensure_user_data_writable(container)
+            record = self._to_record(container, sandbox_id)
+            if not record.sandbox_url:
+                raise RuntimeError(f"sandbox {sandbox_id} has no mapped host port")
+            if not wait_for_sandbox_ready(record.sandbox_url, timeout_seconds=self._health_timeout_seconds):
+                raise RuntimeError(f"sandbox {sandbox_id} is not ready at {record.sandbox_url}")
+            return record
+
+    def discover(self, sandbox_id: str) -> SandboxRecord | None:
+        container = self._get_container(sandbox_id)
+        if container is None:
+            return None
+        container.reload()
+        record = self._to_record(container, sandbox_id)
+        if not record.sandbox_url:
+            return None
+        if not wait_for_sandbox_ready(record.sandbox_url, timeout_seconds=5):
+            return None
+        return record
+
+    def list(self) -> list[SandboxRecord]:
+        containers = self._client.containers.list(
+            all=True, filters={"label": ["app=yuxi-sandbox", "managed-by=yuxi-sandbox-provisioner"]}
+        )
+        records: list[SandboxRecord] = []
+        for container in containers:
+            labels = container.labels or {}
+            sandbox_id = labels.get("sandbox-id")
+            if sandbox_id:
+                container.reload()
+                records.append(self._to_record(container, sandbox_id))
+        return records
+
+    def delete(self, sandbox_id: str) -> None:
+        container = self._get_container(sandbox_id)
+        if container is None:
+            return
+        if container.status == "running":
+            container.stop(timeout=10)
+        container.remove(v=True, force=True)
+
+
 class KubernetesProvisionerBackend:
     def __init__(self):
         from kubernetes import client, config
 
         self._lock = threading.Lock()
         self._namespace = os.getenv("K8S_NAMESPACE", "yuxi-know")
-        self._sandbox_image = os.getenv("SANDBOX_IMAGE", "ghcr.io/bytedance/deer-flow-sandbox:latest")
+        self._sandbox_image = os.getenv(
+            "SANDBOX_IMAGE",
+            "enterprise-public-cn-beijing.cr.volces.com/vefaas-public/all-in-one-sandbox:latest",
+        )
         self._skills_host_path = os.getenv("SKILLS_HOST_PATH", "/app/skills")
         self._threads_host_path = os.getenv("THREADS_HOST_PATH", "/app/saves/threads")
         self._node_host = os.getenv("NODE_HOST", "host.docker.internal")
-        self._container_port = int(os.getenv("SANDBOX_CONTAINER_PORT", "8000"))
+        self._container_port = int(os.getenv("SANDBOX_CONTAINER_PORT", "8080"))
 
         kubeconfig_path = os.getenv("KUBECONFIG_PATH")
         if kubeconfig_path:
@@ -262,28 +501,130 @@ class KubernetesProvisionerBackend:
                     raise
 
 
+class SandboxIdleReaper:
+    def __init__(self, backend):
+        self._backend = backend
+        self._lock = threading.Lock()
+        self._last_activity_at: dict[str, float] = {}
+        self._stop_event = threading.Event()
+        self._thread: threading.Thread | None = None
+        self._exec_timeout_seconds = int(os.getenv("SANDBOX_EXEC_TIMEOUT_SECONDS", "180"))
+        configured_idle_timeout = int(os.getenv("SANDBOX_IDLE_TIMEOUT_SECONDS", "120"))
+        if 0 < configured_idle_timeout <= self._exec_timeout_seconds:
+            logger.warning(
+                "SANDBOX_IDLE_TIMEOUT_SECONDS=%s is <= SANDBOX_EXEC_TIMEOUT_SECONDS=%s; "
+                "adjusting idle timeout to %s seconds to avoid reaping running commands",
+                configured_idle_timeout,
+                self._exec_timeout_seconds,
+                self._exec_timeout_seconds + 30,
+            )
+            configured_idle_timeout = self._exec_timeout_seconds + 30
+        self._idle_timeout_seconds = configured_idle_timeout
+        self._check_interval_seconds = max(1, int(os.getenv("SANDBOX_IDLE_CHECK_INTERVAL_SECONDS", "10")))
+
+    def touch(self, sandbox_id: str) -> None:
+        with self._lock:
+            self._last_activity_at[sandbox_id] = time.time()
+
+    def forget(self, sandbox_id: str) -> None:
+        with self._lock:
+            self._last_activity_at.pop(sandbox_id, None)
+
+    def _seed_existing(self) -> None:
+        try:
+            records = self._backend.list()
+        except Exception as exc:  # noqa: BLE001
+            logger.warning(f"Failed to seed sandbox activity for idle reaper: {exc}")
+            return
+
+        now = time.time()
+        with self._lock:
+            for record in records:
+                self._last_activity_at.setdefault(record.sandbox_id, now)
+
+    def _collect_expired_sandbox_ids(self) -> list[str]:
+        if self._idle_timeout_seconds <= 0:
+            return []
+        cutoff = time.time() - self._idle_timeout_seconds
+        with self._lock:
+            return [sandbox_id for sandbox_id, last_at in self._last_activity_at.items() if last_at <= cutoff]
+
+    def _run(self) -> None:
+        while not self._stop_event.wait(self._check_interval_seconds):
+            expired_ids = self._collect_expired_sandbox_ids()
+            for sandbox_id in expired_ids:
+                try:
+                    self._backend.delete(sandbox_id)
+                    logger.info(f"Deleted idle sandbox: {sandbox_id}")
+                    self.forget(sandbox_id)
+                except Exception as exc:  # noqa: BLE001
+                    logger.warning(f"Failed to delete idle sandbox {sandbox_id}: {exc}")
+
+    def start(self) -> None:
+        if self._idle_timeout_seconds <= 0:
+            logger.info("Idle reaper disabled (SANDBOX_IDLE_TIMEOUT_SECONDS <= 0)")
+            return
+        self._seed_existing()
+        self._thread = threading.Thread(target=self._run, name="sandbox-idle-reaper", daemon=True)
+        self._thread.start()
+        logger.info(
+            "Started sandbox idle reaper with timeout=%ss interval=%ss",
+            self._idle_timeout_seconds,
+            self._check_interval_seconds,
+        )
+
+    def shutdown(self) -> None:
+        self._stop_event.set()
+        if self._thread is not None:
+            self._thread.join(timeout=3)
+
+
 def _build_backend():
     backend = (os.getenv("PROVISIONER_BACKEND", "memory") or "memory").strip().lower()
+    if backend in {"docker", "local"}:
+        return LocalContainerProvisionerBackend(), backend
     if backend == "kubernetes":
         return KubernetesProvisionerBackend(), backend
     return MemoryProvisionerBackend(), backend
 
 
-app = FastAPI(title="Yuxi Sandbox Provisioner")
 backend_impl, backend_name = _build_backend()
+idle_reaper = SandboxIdleReaper(backend_impl)
+
+
+@asynccontextmanager
+async def lifespan(_app: FastAPI):
+    idle_reaper.start()
+    try:
+        yield
+    finally:
+        idle_reaper.shutdown()
+
+
+app = FastAPI(title="Yuxi Sandbox Provisioner", lifespan=lifespan)
 
 
 @app.get("/health")
 def health():
-    return {"status": "ok", "backend": backend_name}
+    tracked = len(idle_reaper._last_activity_at)  # noqa: SLF001
+    return {
+        "status": "ok",
+        "backend": backend_name,
+        "idle_timeout_seconds": idle_reaper._idle_timeout_seconds,  # noqa: SLF001
+        "idle_check_interval_seconds": idle_reaper._check_interval_seconds,  # noqa: SLF001
+        "tracked_sandboxes": tracked,
+    }
 
 
 @app.post("/api/sandboxes", response_model=SandboxResponse)
 def create_sandbox(payload: CreateSandboxRequest):
     try:
         record = backend_impl.create(payload.sandbox_id, payload.thread_id)
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
     except Exception as exc:  # noqa: BLE001
         raise HTTPException(status_code=500, detail=str(exc)) from exc
+    idle_reaper.touch(record.sandbox_id)
     return SandboxResponse(
         sandbox_id=record.sandbox_id,
         sandbox_url=record.sandbox_url,
@@ -300,6 +641,7 @@ def get_sandbox(sandbox_id: str):
 
     if record is None:
         raise HTTPException(status_code=404, detail="sandbox not found")
+    idle_reaper.touch(record.sandbox_id)
 
     return SandboxResponse(
         sandbox_id=record.sandbox_id,
@@ -308,6 +650,18 @@ def get_sandbox(sandbox_id: str):
     )
 
 
+@app.post("/api/sandboxes/{sandbox_id}/touch", response_model=TouchSandboxResponse)
+def touch_sandbox(sandbox_id: str):
+    try:
+        record = backend_impl.discover(sandbox_id)
+    except Exception as exc:  # noqa: BLE001
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+    if record is None:
+        raise HTTPException(status_code=404, detail="sandbox not found")
+    idle_reaper.touch(sandbox_id)
+    return TouchSandboxResponse(ok=True, sandbox_id=sandbox_id, status=record.status)
+
+
 @app.get("/api/sandboxes", response_model=ListSandboxesResponse)
 def list_sandboxes():
     try:
@@ -332,5 +686,6 @@ def delete_sandbox(sandbox_id: str):
         backend_impl.delete(sandbox_id)
     except Exception as exc:  # noqa: BLE001
         raise HTTPException(status_code=500, detail=str(exc)) from exc
+    idle_reaper.forget(sandbox_id)
 
     return DeleteSandboxResponse(ok=True, sandbox_id=sandbox_id)
diff --git a/docker/sandbox_provisioner/requirements.txt b/docker/sandbox_provisioner/requirements.txt
index d8500503..7feabc1f 100644
--- a/docker/sandbox_provisioner/requirements.txt
+++ b/docker/sandbox_provisioner/requirements.txt
@@ -1,3 +1,4 @@
 fastapi>=0.121
 uvicorn[standard]>=0.34.2
 kubernetes>=31.0.0
+docker>=7.1.0
diff --git a/docs/plans/2026-03-05-refactor-thread-upload-state-source-files-plan.md b/docs/plans/2026-03-05-refactor-thread-upload-state-source-files-plan.md
new file mode 100644
index 00000000..fcc6566b
--- /dev/null
+++ b/docs/plans/2026-03-05-refactor-thread-upload-state-source-files-plan.md
@@ -0,0 +1,90 @@
+---
+title: refactor: remove auto attachment conversion and persist uploads in state
+type: refactor
+date: 2026-03-05
+---
+
+# refactor: remove auto attachment conversion and persist uploads in state
+
+## Overview
+
+将聊天附件链路从“上传即转换为 markdown 并写入 state.files”重构为“上传即保存原文件并写入 state.uploads 元数据”。目标是把附件处理从预处理模式改为按需读取模式，避免在 agent 未使用文件前做不必要转换，并统一线程目录下的原始文件作为唯一事实来源。
+
+## Problem Statement / Motivation
+
+当前实现在上传阶段存在较强耦合：
+- 上传流程在 `src/services/conversation_service.py` 同时做文件保存、可选 markdown 转换、状态同步。
+- state 写入依赖 `attachment.markdown -> files[path].content`（`conversation_service.py`、`chat_stream_service.py` 的 `_build_state_files`）。
+- 中间件提示词依赖 `attachments` 列表，但 UI 和状态还混用了 `virtual_path/original_virtual_path/file_path/markdown/truncated` 等字段。
+
+这导致：
+- 上传阶段成本高（转换前置）。
+- 状态语义混乱（`attachments` 既是展示元数据又承载转换结果）。
+- 后续扩展“按需解析/按类型处理”时改动面过大。
+
+## Proposed Solution
+
+采用“上传原文件优先”的单一路径：
+
+1. 上传阶段不再触发自动 markdown 转换（移除 `doc_converter` 依赖链路）。
+2. 原文件统一保存到线程目录 `.../user-data/uploads`，并同步到 sandbox 同路径。
+3. LangGraph state 新增并使用 `uploads` 作为附件事实来源（每个条目包含文件标识、路径、大小、类型、时间等必要元数据）。
+4. `attachments` 退出状态主链路，`uploads` 成为唯一状态来源。
+5. agent 读取文件内容统一通过现有文件工具（如 `read_file`）在需要时访问 `uploads` 中声明的路径。
+
+## Key Decisions (Locked)
+
+- 不采用双写：不同时维护 `attachments` 与 `uploads` 两套状态事实来源。
+- 单次切换：本次重构同时更新后端与前端读取路径，直接切到 `uploads`。
+- 不维护 `state.files` 镜像：停止由附件内容派生 `files`，避免冗余状态与同步复杂度。
+
+## Technical Considerations
+
+- 需要统一以下使用点的数据语义：
+  - `src/services/conversation_service.py`：上传、删除、state 同步。
+  - `src/services/chat_stream_service.py`：`get_agent_state_view` 的回填逻辑。
+  - `src/agents/common/middlewares/attachment_middleware.py`：提示词来源从 `attachments` 迁移/对齐到 `uploads`。
+  - `web/src/components/AgentChatComponent.vue` 与附件组件：字段从 `original_virtual_path/file_path` 对齐到单一上传路径。
+- 兼容策略：不做运行时兼容层，不做双写；接口与前端在同一版本内完成同步切换。
+
+## Acceptance Criteria
+
+- [ ] 上传任意受支持大小文件后，不再触发自动 markdown 转换。
+- [ ] 原文件可在线程 `uploads` 目录中稳定落盘并可下载。
+- [ ] state 中存在 `uploads`，且能完整表达当前线程附件集合。
+- [ ] agent 在无预转换 markdown 的前提下，仍可基于上传文件路径正常执行读取流程。
+- [ ] 删除附件后，state.uploads 与线程目录保持一致。
+- [ ] 前端附件列表与提及能力在新数据结构下行为一致（无回归）。
+
+## Success Metrics
+
+- 上传接口平均耗时下降（相对当前版本）。
+- 上传失败率不高于当前基线。
+- 相关会话流程无新增 P1 回归（上传、查看、下载、删除、提及）。
+
+## Dependencies & Risks
+
+- 风险：前端仍依赖旧字段（`markdown/truncated/original_virtual_path`）导致展示回归。
+- 风险：`get_agent_state_view` 仍按旧逻辑回填 `files`，可能与 `uploads` 语义冲突。
+- 风险：历史会话中仅存在旧 `attachments` 结构的数据，在切换后可能不再自动注入到模型上下文。
+
+## SpecFlow Gaps / Edge Cases
+
+- 同名文件覆盖策略（覆盖、重命名、拒绝）需明确。
+- 无扩展名文件与非常规 MIME 文件在 UI 展示规则需明确。
+- 历史会话首次读取时是否做一次性状态归一化需明确。
+
+## References & Research
+
+- `src/services/conversation_service.py`
+- `src/services/chat_stream_service.py`
+- `src/agents/common/middlewares/attachment_middleware.py`
+- `src/repositories/conversation_repository.py`
+- `server/routers/chat_router.py`
+- `web/src/components/AgentChatComponent.vue`
+
+## Scope Boundaries
+
+- 本次不包含上传后自动解析（markdown/OCR）链路；若需要解析，后续按“显式触发”另开能力。
+- 本次不引入运行时兼容分支（如旧字段回填、双状态同步）。
+- 历史数据迁移若需要，采用一次性离线迁移，不在在线请求链路增加复杂度。
diff --git a/server/routers/chat_router.py b/server/routers/chat_router.py
index 4153e150..037b8bdf 100644
--- a/server/routers/chat_router.py
+++ b/server/routers/chat_router.py
@@ -679,11 +679,8 @@ class AttachmentResponse(BaseModel):
     file_size: int
     status: str
     uploaded_at: str
-    truncated: bool | None = False
-    virtual_path: str | None = None
+    path: str
     artifact_url: str | None = None
-    original_virtual_path: str | None = None
-    original_artifact_url: str | None = None
     minio_url: str | None = None
 
 
@@ -787,7 +784,7 @@ async def upload_thread_attachment(
     db: AsyncSession = Depends(get_db),
     current_user: User = Depends(get_required_user),
 ):
-    """上传并解析附件为 Markdown，附加到指定对话线程。"""
+    """上传原始附件并关联到指定对话线程。"""
     return await upload_thread_attachment_view(
         thread_id=thread_id,
         file=file,
diff --git a/src/agents/common/backends/minio_backend.py b/src/agents/common/backends/minio_backend.py
new file mode 100644
index 00000000..e300a330
--- /dev/null
+++ b/src/agents/common/backends/minio_backend.py
@@ -0,0 +1,59 @@
+"""
+MinIO Backend for Deep Agents
+
+基于 S3_backend 适配的 MinIO 后端实现。
+使用环境变量配置 MinIO 连接，专用于 Agent 状态存储。
+"""
+
+from __future__ import annotations
+
+import os
+
+from src.utils.S3_backend import S3Backend, S3Config
+
+__all__ = ["MinIOBackend", "get_minio_backend"]
+
+
+def get_minio_backend(runtime) -> MinIOBackend:
+    thread_id = getattr(runtime, "config", {}).get("configurable", {}).get("thread_id")
+    """获取 MinIO 后端实例（单例）"""
+    return MinIOBackend(thread_id)
+
+
+class MinIOBackend(S3Backend):
+    """
+    基于 S3Backend 的 MinIO 后端。
+
+    使用环境变量配置：
+        - MINIO_URI: MinIO 端点地址（默认: http://milvus-minio:9000）
+        - MINIO_ACCESS_KEY: 访问密钥（默认: minioadmin）
+        - MINIO_SECRET_KEY: 密钥（默认: minioadmin）
+
+    默认配置：
+        - bucket: "state-bucket"
+        - prefix: "threads"
+    """
+
+    def __init__(self, thread_id: str) -> None:
+        endpoint = os.getenv("MINIO_URI") or "http://milvus-minio:9000"
+        access_key = os.getenv("MINIO_ACCESS_KEY") or "minioadmin"
+        secret_key = os.getenv("MINIO_SECRET_KEY") or "minioadmin"
+        
+
+        # 从 endpoint 提取 region（MinIO 通常用 us-east-1）
+        region = os.getenv("MINIO_REGION") or "us-east-1"
+
+        config = S3Config(
+            bucket="agent-state-bucket",
+            prefix=f"threads/{thread_id}",
+            region=region,
+            endpoint_url=endpoint,
+            access_key_id=access_key,
+            secret_access_key=secret_key,
+            use_ssl=endpoint.startswith("https://"),
+            max_pool_connections=50,
+            connect_timeout=5.0,
+            read_timeout=30.0,
+            max_retries=3,
+        )
+        super().__init__(config)
\ No newline at end of file
diff --git a/src/agents/common/middlewares/attachment_middleware.py b/src/agents/common/middlewares/attachment_middleware.py
index 0e423508..cd6bef55 100644
--- a/src/agents/common/middlewares/attachment_middleware.py
+++ b/src/agents/common/middlewares/attachment_middleware.py
@@ -1,6 +1,7 @@
-"""附件注入中间件 - 使用 LangChain 标准中间件实现
+"""Attachment prompt injection middleware.
 
-从 State 中读取附件信息，注入提示词让模型使用 read_file 工具读取附件内容。
+Read uploaded file metadata from LangGraph state and inject readable paths
+into the system prompt so the model can use `read_file` on demand.
 """
 
 from __future__ import annotations
@@ -18,67 +19,51 @@ ATTACHMENT_PROMPT_MARKER = "<!-- attachment_context -->"
 
 
 class AttachmentState(AgentState):
-    """扩展 AgentState 以支持附件"""
+    """Extended state schema with uploaded files."""
 
-    attachments: NotRequired[list[dict]]
+    uploads: NotRequired[list[dict]]
 
 
-def _build_attachment_prompt(attachments: Sequence[dict]) -> str | None:
-    """Render attachments into a system prompt block with file paths.
-
-    提示模型使用 read_file 工具读取附件内容。
-    """
-    if not attachments:
+def _build_attachment_prompt(uploads: Sequence[dict]) -> str | None:
+    """Render uploads into a concise prompt block."""
+    if not uploads:
         return None
 
-    valid_attachments = [a for a in attachments if a.get("status") == "parsed"]
+    valid_uploads: list[tuple[str, str]] = []
+    for upload in uploads:
+        path = upload.get("path")
+        if not isinstance(path, str) or not path.strip():
+            continue
+        file_name = upload.get("file_name", "未知文件")
+        valid_uploads.append((str(file_name), path))
 
-    if not valid_attachments:
+    if not valid_uploads:
         return None
 
-    attachment_infos: list[str] = []
-    for attachment in valid_attachments:
-        file_name = attachment.get("file_name", "未知文件")
-        file_path = attachment.get("file_path", "")
-        truncated = "（已截断）" if attachment.get("truncated") else ""
-
-        if file_path:
-            attachment_infos.append(f"- {file_name}{truncated}: {file_path}")
-        else:
-            attachment_infos.append(f"- {file_name}{truncated}")
-
+    upload_infos = [f"- {file_name}: {path}" for file_name, path in valid_uploads]
     lines = [
-        "用户上传了以下附件：",
+        "用户上传了以下文件：",
         "",
-        *attachment_infos,
+        *upload_infos,
         "",
-        "请使用 read_file 工具读取附件内容后，再回答用户的问题。",
+        "请优先使用 `read_file` 工具读取这些路径中的文件内容，再回答用户问题。",
     ]
-
     return "\n".join(lines)
 
 
 class AttachmentMiddleware(AgentMiddleware[AttachmentState]):
-    """
-    LangChain 标准中间件：从 State 中读取附件并注入提示词。
-
-    LangGraph 会自动从 checkpointer 恢复 state，包括 attachments。
-    从 request.state 中读取附件，将其转换为上下文块 并注入到系统提示词中。
-    """
+    """Inject upload context from state.uploads into system prompt."""
 
     state_schema = AttachmentState
 
     async def awrap_model_call(
         self, request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]
     ) -> ModelResponse:
-        # 从 state 获取附件（LangGraph 自动从 checkpointer 恢复）
-        attachments = request.state.get("attachments", [])
-        logger.info(f"AttachmentMiddleware: found {len(attachments)} attachments in state")
-
-        if attachments:
-            # 构建附件提示
-            attachment_prompt = _build_attachment_prompt(attachments)
+        uploads = request.state.get("uploads", [])
+        logger.info(f"AttachmentMiddleware: found {len(uploads)} uploads in state")
 
+        if uploads:
+            attachment_prompt = _build_attachment_prompt(uploads)
             if attachment_prompt:
                 logger.info("AttachmentMiddleware: injecting attachment prompt")
                 existing_blocks = list(request.system_message.content_blocks) if request.system_message else []
@@ -100,8 +85,5 @@ class AttachmentMiddleware(AgentMiddleware[AttachmentState]):
         return await handler(request)
 
 
-# 创建中间件实例，供其他模块使用
 save_attachments_to_fs = AttachmentMiddleware()
-
-# 保留旧名称以保持向后兼容（已废弃）
 inject_attachment_context = save_attachments_to_fs
diff --git a/src/config/app.py b/src/config/app.py
index 200791d7..20f24253 100644
--- a/src/config/app.py
+++ b/src/config/app.py
@@ -43,6 +43,7 @@ class Config(BaseModel):
     sandbox_virtual_path_prefix: str = Field(default="/mnt/user-data")
     sandbox_exec_timeout_seconds: int = Field(default=180)
     sandbox_max_output_bytes: int = Field(default=262144)
+    sandbox_keepalive_interval_seconds: int = Field(default=30)
 
     model_names: dict[str, ChatModelProvider] = Field(
         default_factory=lambda: DEFAULT_CHAT_MODEL_PROVIDERS.copy(),
@@ -166,6 +167,9 @@ class Config(BaseModel):
         self.sandbox_max_output_bytes = int(
             os.getenv("SANDBOX_MAX_OUTPUT_BYTES") or self.sandbox_max_output_bytes or 262144
         )
+        self.sandbox_keepalive_interval_seconds = int(
+            os.getenv("SANDBOX_KEEPALIVE_INTERVAL_SECONDS") or self.sandbox_keepalive_interval_seconds or 30
+        )
 
         if self.sandbox_provider.lower() != "provisioner":
             raise ValueError("Only sandbox_provider=provisioner is supported.")
diff --git a/src/sandbox/backend.py b/src/sandbox/backend.py
index 55862658..c54178af 100644
--- a/src/sandbox/backend.py
+++ b/src/sandbox/backend.py
@@ -130,6 +130,37 @@ class ProvisionerSandboxBackend(BaseSandbox):
         except Exception:  # noqa: BLE001
             return content.encode("utf-8")
 
+    def read(
+        self,
+        file_path: str,
+        offset: int = 0,
+        limit: int = 2000,
+    ) -> str:
+        """Read file content directly via file API to avoid shell-output false positives."""
+        normalized_path = _normalize_path(file_path)
+        start = max(0, int(offset))
+        size = max(0, int(limit))
+
+        try:
+            content = self._read_binary(normalized_path)
+        except Exception:  # noqa: BLE001
+            return f"Error: File '{file_path}' not found"
+
+        if not content:
+            return "System reminder: File exists but has empty contents"
+
+        text = content.decode("utf-8", errors="replace")
+        lines = text.splitlines()
+        selected_lines = lines[start : start + size]
+
+        if not selected_lines:
+            return ""
+
+        return "\n".join(
+            f"{start + idx + 1:6d}\t{line}"
+            for idx, line in enumerate(selected_lines)
+        )
+
     def execute(self, command: str) -> ExecuteResponse:
         try:
             result = self._shell_exec(command)
diff --git a/src/sandbox/provider.py b/src/sandbox/provider.py
index fd848117..dfc2964b 100644
--- a/src/sandbox/provider.py
+++ b/src/sandbox/provider.py
@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import hashlib
 import threading
+import time
 from dataclasses import dataclass
 
 from src import config as conf
@@ -36,6 +37,8 @@ class ProvisionerSandboxProvider:
         self._lock = threading.Lock()
         self._thread_locks: dict[str, threading.Lock] = {}
         self._connections: dict[str, SandboxConnection] = {}
+        self._last_touch_at: dict[str, float] = {}
+        self._touch_interval_seconds = int(getattr(conf, "sandbox_keepalive_interval_seconds", 30))
 
     def _thread_lock(self, thread_id: str) -> threading.Lock:
         with self._lock:
@@ -52,14 +55,37 @@ class ProvisionerSandboxProvider:
             sandbox_url=record.sandbox_url,
         )
         self._connections[thread_id] = connection
+        self._last_touch_at[thread_id] = time.time()
         return connection
 
+    def _should_touch(self, thread_id: str) -> bool:
+        if self._touch_interval_seconds <= 0:
+            return False
+        last_touch = self._last_touch_at.get(thread_id)
+        if last_touch is None:
+            return True
+        return (time.time() - last_touch) >= self._touch_interval_seconds
+
+    def _touch_if_needed(self, connection: SandboxConnection) -> bool:
+        if not self._should_touch(connection.thread_id):
+            return True
+        is_alive = self._client.touch(connection.sandbox_id)
+        self._last_touch_at[connection.thread_id] = time.time()
+        return is_alive
+
     def acquire(self, thread_id: str) -> str:
         lock = self._thread_lock(thread_id)
         with lock:
             current = self._connections.get(thread_id)
             if current:
-                return current.sandbox_id
+                try:
+                    if self._touch_if_needed(current):
+                        return current.sandbox_id
+                    self._connections.pop(thread_id, None)
+                    self._last_touch_at.pop(thread_id, None)
+                except Exception as exc:  # noqa: BLE001
+                    logger.warning(f"Failed to touch sandbox {current.sandbox_id} for thread {thread_id}: {exc}")
+                    return current.sandbox_id
 
             sandbox_id = sandbox_id_for_thread(thread_id)
             record = self._client.discover(sandbox_id)
@@ -75,6 +101,17 @@ class ProvisionerSandboxProvider:
     def get(self, thread_id: str, *, create_if_missing: bool = False) -> SandboxConnection | None:
         lock = self._thread_lock(thread_id)
         with lock:
+            current = self._connections.get(thread_id)
+            if current:
+                try:
+                    if self._touch_if_needed(current):
+                        return current
+                    self._connections.pop(thread_id, None)
+                    self._last_touch_at.pop(thread_id, None)
+                except Exception as exc:  # noqa: BLE001
+                    logger.warning(f"Failed to touch sandbox {current.sandbox_id} for thread {thread_id}: {exc}")
+                    return current
+
             current = self._connections.get(thread_id)
             if current:
                 return current
@@ -92,6 +129,7 @@ class ProvisionerSandboxProvider:
         with self._lock:
             connections = list(self._connections.values())
             self._connections.clear()
+            self._last_touch_at.clear()
 
         for connection in connections:
             try:
diff --git a/src/sandbox/provisioner_client.py b/src/sandbox/provisioner_client.py
index cdaa4832..8cbb2412 100644
--- a/src/sandbox/provisioner_client.py
+++ b/src/sandbox/provisioner_client.py
@@ -63,6 +63,17 @@ class ProvisionerClient:
             status=payload.get("status"),
         )
 
+    def touch(self, sandbox_id: str) -> bool:
+        response = self._request("POST", f"/api/sandboxes/{sandbox_id}/touch")
+        if response.status_code == 404:
+            return False
+        if response.status_code >= 400:
+            raise RuntimeError(
+                f"failed to touch sandbox {sandbox_id}: "
+                f"{response.status_code} {response.text}"
+            )
+        return True
+
     def delete(self, sandbox_id: str) -> None:
         response = self._request("DELETE", f"/api/sandboxes/{sandbox_id}")
         if response.status_code in {200, 404}:
diff --git a/src/services/chat_stream_service.py b/src/services/chat_stream_service.py
index 1ac94927..da120f52 100644
--- a/src/services/chat_stream_service.py
+++ b/src/services/chat_stream_service.py
@@ -3,7 +3,6 @@ import json
 import traceback
 import uuid
 from collections.abc import AsyncIterator
-from datetime import UTC, datetime
 
 from langchain.messages import AIMessage, AIMessageChunk, HumanMessage
 from langgraph.types import Command
@@ -18,39 +17,29 @@ from src.storage.postgres.manager import pg_manager
 from src.utils.logging_config import logger
 
 
-def _build_state_files(attachments: list[dict]) -> dict:
-    """将附件列表转换为 StateBackend 格式的 files 字典
-
-    StateBackend 期望的格式:
-    {
-        "/attachments/file.md": {
-            "content": ["line1", "line2", ...],
-            "created_at": "...",
-            "modified_at": "...",
-        }
-    }
-    """
-    files = {}
+def _build_state_uploads(attachments: list[dict]) -> list[dict]:
+    """Convert persisted attachment metadata into state.uploads entries."""
+    uploads: list[dict] = []
     for attachment in attachments:
-        if attachment.get("status") != "parsed":
+        file_path = attachment.get("path")
+        if not isinstance(file_path, str) or not file_path.strip():
             continue
 
-        file_path = attachment.get("file_path")
-        markdown = attachment.get("markdown")
+        uploads.append(
+            {
+                "file_id": attachment.get("file_id"),
+                "file_name": attachment.get("file_name"),
+                "file_type": attachment.get("file_type"),
+                "file_size": attachment.get("file_size", 0),
+                "status": attachment.get("status", "uploaded"),
+                "uploaded_at": attachment.get("uploaded_at"),
+                "path": file_path,
+                "artifact_url": attachment.get("artifact_url"),
+            }
+        )
 
-        if not file_path or not markdown:
-            continue
+    return uploads
 
-        now = datetime.now(UTC).isoformat()
-        # 将 markdown 内容按行拆分
-        content_lines = markdown.split("\n")
-        files[file_path] = {
-            "content": content_lines,
-            "created_at": attachment.get("uploaded_at", now),
-            "modified_at": attachment.get("uploaded_at", now),
-        }
-
-    return files
 
 
 async def _get_langgraph_messages(agent_instance, config_dict):
@@ -65,20 +54,20 @@ async def _get_langgraph_messages(agent_instance, config_dict):
 
 
 def extract_agent_state(values: dict) -> dict:
-    """从 LangGraph state 中提取 agent 状态"""
+    """Extract agent state payload returned to frontend."""
     if not isinstance(values, dict):
         return {}
 
-    # 直接获取，信任 state 的数据结构
     todos = values.get("todos")
     result = {
         "todos": list(todos)[:20] if todos else [],
-        "files": values.get("files") or {},
+        "uploads": values.get("uploads") or [],
     }
 
     return result
 
 
+
 async def _get_existing_message_ids(conv_repo: ConversationRepository, thread_id: str) -> set[str]:
     existing_messages = await conv_repo.get_messages_by_thread_id(thread_id)
     return {
@@ -379,7 +368,7 @@ async def stream_agent_chat(
         # 先构建 langgraph_config
         langgraph_config = {"configurable": {"thread_id": thread_id, "user_id": user_id}}
 
-        # 注意：LangGraph 会自动从 checkpointer 恢复 state（包括 attachments 和 files）
+        # LangGraph 会自动从 checkpointer 恢复 state（包括 uploads）
         # 无需手动加载或传递
 
         # 根据用户权限过滤知识库
@@ -654,22 +643,15 @@ async def get_agent_state_view(
     state = await graph.aget_state(langgraph_config)
     agent_state = extract_agent_state(getattr(state, "values", {})) if state else {}
 
-    # 如果 state 中没有 files，从附件构建
-    # 这确保了上传附件后立即可以在文件列表中看到文件
-    if not agent_state.get("files") or agent_state["files"] == {}:
+    # 如果 state 中暂时没有 uploads，则从持久化附件记录回填
+    if not isinstance(agent_state.get("uploads"), list) or not agent_state["uploads"]:
         try:
             attachments = await conv_repo.get_attachments_by_thread_id(thread_id)
             logger.info(f"[get_agent_state_view] found {len(attachments)} attachments in DB")
             if attachments:
-                first_status = attachments[0].get("status")
-                first_has_markdown = bool(attachments[0].get("markdown"))
-                logger.info(
-                    f"[get_agent_state_view] first attachment status: {first_status}, "
-                    f"has markdown: {first_has_markdown}"
-                )
-                files = _build_state_files(attachments)
-                agent_state["files"] = files
-                logger.info(f"[get_agent_state_view] Built files from attachments: {len(files)} files")
+                uploads = _build_state_uploads(attachments)
+                agent_state["uploads"] = uploads
+                logger.info(f"[get_agent_state_view] Built uploads from attachments: {len(uploads)} files")
         except Exception as e:
             logger.warning(f"Failed to fetch attachments for thread {thread_id}: {e}")
 
diff --git a/src/services/conversation_service.py b/src/services/conversation_service.py
index 36493509..0d7ece07 100644
--- a/src/services/conversation_service.py
+++ b/src/services/conversation_service.py
@@ -1,6 +1,5 @@
 import shlex
 import uuid
-from datetime import UTC, datetime
 from pathlib import Path
 
 from fastapi import HTTPException, UploadFile
@@ -14,16 +13,11 @@ from src.sandbox import (
     get_sandbox_provider,
     sandbox_uploads_dir,
 )
-from src.services.doc_converter import (
-    ATTACHMENT_ALLOWED_EXTENSIONS,
-    MAX_ATTACHMENT_SIZE_BYTES,
-    convert_upload_to_markdown,
-)
+from src.services.doc_converter import ATTACHMENT_ALLOWED_EXTENSIONS, MAX_ATTACHMENT_SIZE_BYTES
 from src.utils.datetime_utils import utc_isoformat
 from src.utils.logging_config import logger
 
 UPLOADS_VIRTUAL_PREFIX = "/mnt/user-data/uploads"
-LEGACY_ATTACHMENTS_PREFIX = "/attachments/"
 
 
 async def require_user_conversation(conv_repo: ConversationRepository, thread_id: str, user_id: str):
@@ -33,21 +27,7 @@ async def require_user_conversation(conv_repo: ConversationRepository, thread_id
     return conversation
 
 
-def _sanitize_file_stem(file_name: str) -> str:
-    base_name = file_name
-    for ext in [".docx", ".txt", ".html", ".htm", ".pdf", ".md"]:
-        if file_name.lower().endswith(ext):
-            base_name = file_name[: -len(ext)]
-            break
-    safe_name = base_name.replace("/", "_").replace("\\", "_").strip(" .")
-    return safe_name or "attachment"
-
-
-def _make_attachment_markdown_virtual_path(file_name: str) -> str:
-    return f"{UPLOADS_VIRTUAL_PREFIX}/{_sanitize_file_stem(file_name)}.md"
-
-
-def _make_attachment_original_virtual_path(file_name: str) -> str:
+def _make_upload_virtual_path(file_name: str) -> str:
     safe_name = file_name.replace("/", "_").replace("\\", "_").strip(" .")
     return f"{UPLOADS_VIRTUAL_PREFIX}/{safe_name or 'attachment.bin'}"
 
@@ -56,27 +36,29 @@ def _artifact_url(thread_id: str, virtual_path: str) -> str:
     return f"/api/chat/thread/{thread_id}/artifacts/{virtual_path.lstrip('/')}"
 
 
-def _build_state_files(attachments: list[dict]) -> dict:
-    files = {}
+def _build_state_uploads(attachments: list[dict]) -> list[dict]:
+    uploads: list[dict] = []
     for attachment in attachments:
-        if attachment.get("status") != "parsed":
+        path = attachment.get("path")
+        if not isinstance(path, str) or not path.strip():
             continue
 
-        file_path = attachment.get("file_path")
-        markdown = attachment.get("markdown")
-        if not file_path or not markdown:
-            continue
-
-        now = datetime.now(UTC).isoformat()
-        files[file_path] = {
-            "content": markdown.split("\n"),
-            "created_at": attachment.get("uploaded_at", now),
-            "modified_at": attachment.get("uploaded_at", now),
-        }
-    return files
+        uploads.append(
+            {
+                "file_id": attachment.get("file_id"),
+                "file_name": attachment.get("file_name"),
+                "file_type": attachment.get("file_type"),
+                "file_size": attachment.get("file_size", 0),
+                "status": attachment.get("status", "uploaded"),
+                "uploaded_at": attachment.get("uploaded_at"),
+                "path": path,
+                "artifact_url": attachment.get("artifact_url"),
+            }
+        )
+    return uploads
 
 
-async def _sync_thread_attachment_state(
+async def _sync_thread_upload_state(
     *,
     thread_id: str,
     user_id: str,
@@ -86,55 +68,33 @@ async def _sync_thread_attachment_state(
     try:
         agent = agent_manager.get_agent(agent_id)
         if not agent:
-            logger.warning(f"Skip attachment state sync: agent not found ({agent_id})")
+            logger.warning(f"Skip upload state sync: agent not found ({agent_id})")
             return
 
         graph = await agent.get_graph()
         config = {"configurable": {"thread_id": thread_id, "user_id": str(user_id)}}
 
-        state = await graph.aget_state(config)
-        state_values = getattr(state, "values", {}) if state else {}
-        existing_files = state_values.get("files", {}) if isinstance(state_values, dict) else {}
-        if not isinstance(existing_files, dict):
-            existing_files = {}
-
-        next_attachment_files = _build_state_files(attachments)
-        prev_attachment_paths = {
-            path
-            for path in existing_files.keys()
-            if isinstance(path, str)
-            and (path.startswith(LEGACY_ATTACHMENTS_PREFIX) or path.startswith(f"{UPLOADS_VIRTUAL_PREFIX}/"))
-        }
-        next_attachment_paths = set(next_attachment_files.keys())
-
-        file_updates: dict[str, dict | None] = {**next_attachment_files}
-        for removed_path in prev_attachment_paths - next_attachment_paths:
-            file_updates[removed_path] = None
-
         await graph.aupdate_state(
             config=config,
             values={
-                "attachments": attachments,
-                "files": file_updates,
+                "uploads": _build_state_uploads(attachments),
             },
         )
     except Exception as exc:  # noqa: BLE001
-        logger.warning(f"Failed to sync attachment state for thread {thread_id}: {exc}")
+        logger.warning(f"Failed to sync upload state for thread {thread_id}: {exc}")
 
 
 def serialize_attachment(record: dict) -> dict:
+    path = record.get("path")
     return {
         "file_id": record.get("file_id"),
         "file_name": record.get("file_name"),
         "file_type": record.get("file_type"),
         "file_size": record.get("file_size", 0),
-        "status": record.get("status", "parsed"),
+        "status": record.get("status", "uploaded"),
         "uploaded_at": record.get("uploaded_at"),
-        "truncated": record.get("truncated", False),
-        "virtual_path": record.get("virtual_path") or record.get("file_path"),
+        "path": path,
         "artifact_url": record.get("artifact_url"),
-        "original_virtual_path": record.get("original_virtual_path"),
-        "original_artifact_url": record.get("original_artifact_url"),
         "minio_url": record.get("minio_url"),
     }
 
@@ -241,29 +201,24 @@ async def upload_thread_attachment_view(
 ) -> dict:
     conv_repo = ConversationRepository(db)
     conversation = await require_user_conversation(conv_repo, thread_id, str(current_user_id))
+    if not file.filename:
+        raise HTTPException(status_code=400, detail="无法识别的文件名")
 
-    try:
-        conversion = await convert_upload_to_markdown(file)
-    except ValueError as exc:
-        raise HTTPException(status_code=400, detail=str(exc)) from exc
-    except Exception as exc:
-        logger.error(f"附件解析失败: {exc}")
-        raise HTTPException(status_code=500, detail="附件解析失败，请稍后重试") from exc
+    file_name = Path(file.filename).name
+    await file.seek(0)
+    file_content = await file.read()
+    file_size = len(file_content)
+    if file_size > MAX_ATTACHMENT_SIZE_BYTES:
+        max_size_mb = MAX_ATTACHMENT_SIZE_BYTES // (1024 * 1024)
+        raise HTTPException(status_code=400, detail=f"附件过大，当前仅支持 {max_size_mb} MB 以内的文件")
 
-    markdown_virtual_path = _make_attachment_markdown_virtual_path(conversion.file_name)
-    original_virtual_path = _make_attachment_original_virtual_path(conversion.file_name)
-    markdown_artifact_url = _artifact_url(thread_id, markdown_virtual_path)
-    original_artifact_url = _artifact_url(thread_id, original_virtual_path)
+    upload_virtual_path = _make_upload_virtual_path(file_name)
+    artifact_url = _artifact_url(thread_id, upload_virtual_path)
 
     ensure_thread_dirs(thread_id)
     uploads_dir = sandbox_uploads_dir(thread_id)
-    markdown_actual_path = uploads_dir / Path(markdown_virtual_path).name
-    original_actual_path = uploads_dir / Path(original_virtual_path).name
-
-    await file.seek(0)
-    file_content = await file.read()
-    markdown_actual_path.write_text(conversion.markdown, encoding="utf-8")
-    original_actual_path.write_bytes(file_content)
+    upload_actual_path = uploads_dir / Path(upload_virtual_path).name
+    upload_actual_path.write_bytes(file_content)
 
     provider = get_sandbox_provider()
     connection = provider.get(thread_id, create_if_missing=False)
@@ -271,32 +226,26 @@ async def upload_thread_attachment_view(
         backend = ProvisionerSandboxBackend(thread_id=thread_id)
         backend.upload_files(
             [
-                (markdown_virtual_path, conversion.markdown.encode("utf-8")),
-                (original_virtual_path, file_content),
+                (upload_virtual_path, file_content),
             ]
         )
 
     attachment_record = {
-        "file_id": conversion.file_id,
-        "file_name": conversion.file_name,
-        "file_type": conversion.file_type,
-        "file_size": conversion.file_size,
-        "status": "parsed",
-        "markdown": conversion.markdown,
+        "file_id": uuid.uuid4().hex,
+        "file_name": file_name,
+        "file_type": file.content_type,
+        "file_size": file_size,
+        "status": "uploaded",
         "uploaded_at": utc_isoformat(),
-        "truncated": conversion.truncated,
-        "file_path": markdown_virtual_path,
-        "virtual_path": markdown_virtual_path,
-        "artifact_url": markdown_artifact_url,
-        "original_virtual_path": original_virtual_path,
-        "original_artifact_url": original_artifact_url,
+        "path": upload_virtual_path,
+        "artifact_url": artifact_url,
         "minio_url": None,
-        "storage_path": str(markdown_actual_path),
-        "original_storage_path": str(original_actual_path),
+        "storage_path": str(upload_actual_path),
     }
+
     await conv_repo.add_attachment(conversation.id, attachment_record)
     all_attachments = await conv_repo.get_attachments(conversation.id)
-    await _sync_thread_attachment_state(
+    await _sync_thread_upload_state(
         thread_id=thread_id,
         user_id=str(current_user_id),
         agent_id=conversation.agent_id,
@@ -342,12 +291,8 @@ async def delete_thread_attachment_view(
         raise HTTPException(status_code=404, detail="附件不存在或已被删除")
 
     if target_attachment:
-        for candidate in [
-            target_attachment.get("storage_path"),
-            target_attachment.get("original_storage_path"),
-        ]:
-            if not candidate:
-                continue
+        candidate = target_attachment.get("storage_path")
+        if candidate:
             try:
                 file_path = Path(candidate)
                 if file_path.exists():
@@ -356,7 +301,7 @@ async def delete_thread_attachment_view(
                 logger.warning(f"Failed to remove attachment file {candidate}: {exc}")
 
     all_attachments = await conv_repo.get_attachments(conversation.id)
-    await _sync_thread_attachment_state(
+    await _sync_thread_upload_state(
         thread_id=thread_id,
         user_id=str(current_user_id),
         agent_id=conversation.agent_id,
@@ -369,12 +314,9 @@ async def delete_thread_attachment_view(
         if connection is not None:
             backend = ProvisionerSandboxBackend(thread_id=thread_id)
             delete_commands = []
-            for path in [
-                target_attachment.get("virtual_path") or target_attachment.get("file_path"),
-                target_attachment.get("original_virtual_path"),
-            ]:
-                if isinstance(path, str) and path.strip():
-                    delete_commands.append(f"rm -f {shlex.quote(path)}")
+            path = target_attachment.get("path")
+            if isinstance(path, str) and path.strip():
+                delete_commands.append(f"rm -f {shlex.quote(path)}")
             if delete_commands:
                 backend.execute(" && ".join(delete_commands))
 
diff --git a/src/services/doc_converter.py b/src/services/doc_converter.py
index 4b0c60f3..edd77692 100644
--- a/src/services/doc_converter.py
+++ b/src/services/doc_converter.py
@@ -13,7 +13,7 @@ from src.config import config as app_config
 from src.knowledge.indexing import process_file_to_markdown
 from src.utils import logger
 
-ATTACHMENT_ALLOWED_EXTENSIONS: tuple[str, ...] = (".txt", ".md", ".docx", ".html", ".htm")
+ATTACHMENT_ALLOWED_EXTENSIONS: tuple[str, ...] = ()
 MAX_ATTACHMENT_SIZE_BYTES = 5 * 1024 * 1024  # 5 MB
 MAX_ATTACHMENT_MARKDOWN_CHARS = 32_000
 
@@ -71,7 +71,7 @@ async def convert_upload_to_markdown(upload: UploadFile) -> ConversionResult:
     file_name = Path(upload.filename).name
     suffix = Path(file_name).suffix.lower()
 
-    if suffix not in ATTACHMENT_ALLOWED_EXTENSIONS:
+    if ATTACHMENT_ALLOWED_EXTENSIONS and suffix not in ATTACHMENT_ALLOWED_EXTENSIONS:
         allowed = ", ".join(ATTACHMENT_ALLOWED_EXTENSIONS)
         raise ValueError(f"不支持的文件类型: {suffix or '未知'}，当前仅支持 {allowed}")
 
diff --git a/src/utils/S3_backend.py b/src/utils/S3_backend.py
new file mode 100644
index 00000000..d50b31e8
--- /dev/null
+++ b/src/utils/S3_backend.py
@@ -0,0 +1,486 @@
+"""
+Deep Agents Remote Backends
+
+S3 backend implementations for LangChain's Deep Agents.
+Supports any S3-compatible storage (AWS S3, MinIO, etc.) 
+with connection pooling for optimal performance.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import fnmatch
+import json
+import re
+import threading
+from contextlib import asynccontextmanager
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import PurePosixPath
+from typing import TYPE_CHECKING, Any, AsyncIterator, Coroutine
+
+import aioboto3
+import wcmatch.glob as wcglob
+from botocore.config import Config as BotoConfig
+from botocore.exceptions import ClientError
+from deepagents.backends.protocol import (
+    BackendProtocol,
+    EditResult,
+    FileDownloadResponse,
+    FileInfo,
+    FileUploadResponse,
+    GrepMatch,
+    WriteResult,
+)
+from deepagents.backends.utils import (
+    check_empty_content,
+    format_content_with_line_numbers,
+    perform_string_replacement,
+)
+
+if TYPE_CHECKING:
+    from types_aiobotocore_s3 import S3Client
+
+__all__ = ["S3Backend", "S3Config"]
+
+
+def run_async_safely[T](coroutine: Coroutine[Any, Any, T], timeout: float | None = None) -> T:
+    try:
+        asyncio.get_running_loop()
+    except RuntimeError:
+        return asyncio.run(coroutine)
+
+    result: dict[str, T] = {}
+    error: dict[str, Exception] = {}
+
+    def _run() -> None:
+        try:
+            result["value"] = asyncio.run(coroutine)
+        except Exception as exc:  # noqa: BLE001
+            error["value"] = exc
+
+    thread = threading.Thread(target=_run, daemon=True)
+    thread.start()
+    thread.join(timeout)
+
+    if thread.is_alive():
+        raise TimeoutError("Timed out while waiting for coroutine result")
+    if "value" in error:
+        raise error["value"]
+
+    return result["value"]
+
+
+# =============================================================================
+# S3 Backend (S3-compatible: AWS S3, MinIO, etc.)
+# =============================================================================
+
+
+@dataclass
+class S3Config:
+    """Configuration for S3-compatible storage."""
+
+    bucket: str
+    prefix: str = ""
+    region: str = "us-east-1"
+    endpoint_url: str | None = None
+    access_key_id: str | None = None
+    secret_access_key: str | None = None
+    use_ssl: bool = True
+    max_pool_connections: int = 50
+    connect_timeout: float = 5.0
+    read_timeout: float = 30.0
+    max_retries: int = 3
+
+
+class S3Backend(BackendProtocol):
+    """
+    S3-compatible backend for Deep Agents file operations.
+
+    Supports AWS S3, MinIO, and any S3-compatible object storage.
+    All operations are async-native using aioboto3.
+
+    Files are stored as objects with paths mapping to S3 keys.
+    Content is stored as JSON with the structure:
+    {"content": [...lines], "created_at": "...", "modified_at": "..."}
+    """
+
+    def __init__(self, config: S3Config) -> None:
+        self._config = config
+        self._prefix = config.prefix.strip("/")
+        if self._prefix:
+            self._prefix += "/"
+
+        self._boto_config = BotoConfig(
+            region_name=config.region,
+            signature_version="s3v4",
+            retries={"max_attempts": config.max_retries, "mode": "adaptive"},
+            max_pool_connections=config.max_pool_connections,
+            connect_timeout=config.connect_timeout,
+            read_timeout=config.read_timeout,
+        )
+
+        session_kwargs: dict[str, Any] = {}
+        if config.access_key_id:
+            session_kwargs["aws_access_key_id"] = config.access_key_id
+        if config.secret_access_key:
+            session_kwargs["aws_secret_access_key"] = config.secret_access_key
+
+        self._session = aioboto3.Session(**session_kwargs)
+        self._bucket = config.bucket
+
+    def _s3_key(self, path: str) -> str:
+        """Convert virtual path to S3 key."""
+        clean = path.lstrip("/")
+        return f"{self._prefix}{clean}"
+
+    def _virtual_path(self, key: str) -> str:
+        """Convert S3 key to virtual path."""
+        if self._prefix and key.startswith(self._prefix):
+            key = key[len(self._prefix) :]
+        return "/" + key.lstrip("/")
+
+    @asynccontextmanager
+    async def _client(self) -> AsyncIterator["S3Client"]:
+        """Get S3 client context."""
+        async with self._session.client(
+            "s3",
+            config=self._boto_config,
+            endpoint_url=self._config.endpoint_url,
+            use_ssl=self._config.use_ssl,
+        ) as client:
+            yield client
+
+    async def _get_file_data(self, path: str) -> dict[str, Any] | None:
+        """Get file data dict from S3."""
+        key = self._s3_key(path)
+        try:
+            async with self._client() as client:
+                response = await client.get_object(Bucket=self._bucket, Key=key)
+                async with response["Body"] as stream:
+                    content = await stream.read()
+                return json.loads(content.decode("utf-8"))
+        except ClientError as e:
+            if e.response["Error"]["Code"] == "NoSuchKey":
+                return None
+            raise
+
+    async def _put_file_data(
+        self, path: str, data: dict[str, Any], *, update_modified: bool = True
+    ) -> None:
+        """Put file data dict to S3."""
+        key = self._s3_key(path)
+        if update_modified:
+            data["modified_at"] = datetime.now(timezone.utc).isoformat()
+        content = json.dumps(data).encode("utf-8")
+        async with self._client() as client:
+            await client.put_object(
+                Bucket=self._bucket,
+                Key=key,
+                Body=content,
+                ContentType="application/json",
+            )
+
+    async def _exists(self, path: str) -> bool:
+        """Check if file exists in S3."""
+        key = self._s3_key(path)
+        try:
+            async with self._client() as client:
+                await client.head_object(Bucket=self._bucket, Key=key)
+            return True
+        except ClientError as e:
+            if e.response["Error"]["Code"] == "404":
+                return False
+            raise
+
+    async def _list_keys(self, prefix: str = "") -> list[dict[str, Any]]:
+        """List all keys with a prefix."""
+        full_prefix = self._s3_key(prefix)
+        results: list[dict[str, Any]] = []
+        async with self._client() as client:
+            paginator = client.get_paginator("list_objects_v2")
+            async for page in paginator.paginate(
+                Bucket=self._bucket, Prefix=full_prefix
+            ):
+                for obj in page.get("Contents", []):
+                    results.append(obj)
+        return results
+
+    # -------------------------------------------------------------------------
+    # BackendProtocol Implementation
+    # -------------------------------------------------------------------------
+
+    def ls_info(self, path: str) -> list[FileInfo]:
+        """Sync wrapper for als_info."""
+        return run_async_safely(self.als_info(path))
+
+    async def als_info(self, path: str) -> list[FileInfo]:
+        """List files in a directory."""
+        prefix = path.lstrip("/")
+        if prefix and not prefix.endswith("/"):
+            prefix += "/"
+
+        objects = await self._list_keys(prefix)
+        results: list[FileInfo] = []
+        seen_dirs: set[str] = set()
+
+        for obj in objects:
+            key = obj["Key"]
+            vpath = self._virtual_path(key)
+
+            # Check if this is a direct child or nested
+            rel = vpath[len("/" + prefix) :] if prefix else vpath[1:]
+            if "/" in rel:
+                # This is in a subdirectory, add the directory entry
+                dir_name = rel.split("/")[0]
+                dir_path = "/" + prefix + dir_name + "/"
+                if dir_path not in seen_dirs:
+                    seen_dirs.add(dir_path)
+                    results.append({"path": dir_path, "is_dir": True})
+            else:
+                # Direct file
+                results.append(
+                    {
+                        "path": vpath,
+                        "is_dir": False,
+                        "size": obj.get("Size", 0),
+                        "modified_at": obj["LastModified"].isoformat()
+                        if "LastModified" in obj
+                        else None,
+                    }
+                )
+
+        results.sort(key=lambda x: x.get("path", ""))
+        return results
+
+    def read(self, file_path: str, offset: int = 0, limit: int = 2000) -> str:
+        """Sync wrapper for aread."""
+        return run_async_safely(
+            self.aread(file_path, offset, limit)
+        )
+
+    async def aread(self, file_path: str, offset: int = 0, limit: int = 2000) -> str:
+        """Read file content with line numbers."""
+        data = await self._get_file_data(file_path)
+        if data is None:
+            return f"Error: File '{file_path}' not found"
+
+        lines = data.get("content", [])
+        if not lines:
+            empty_msg = check_empty_content("")
+            if empty_msg:
+                return empty_msg
+
+        if offset >= len(lines):
+            return f"Error: Line offset {offset} exceeds file length ({len(lines)} lines)"
+
+        selected = lines[offset : offset + limit]
+        return format_content_with_line_numbers(selected, start_line=offset + 1)
+
+    def write(self, file_path: str, content: str) -> WriteResult:
+        """Sync wrapper for awrite."""
+        return run_async_safely(
+            self.awrite(file_path, content)
+        )
+
+    async def awrite(self, file_path: str, content: str) -> WriteResult:
+        """Create a new file."""
+        if await self._exists(file_path):
+            return WriteResult(
+                error=f"Cannot write to {file_path} because it already exists. "
+                "Read and then make an edit, or write to a new path."
+            )
+
+        now = datetime.now(timezone.utc).isoformat()
+        data = {
+            "content": content.splitlines(),
+            "created_at": now,
+            "modified_at": now,
+        }
+        try:
+            await self._put_file_data(file_path, data, update_modified=False)
+            return WriteResult(path=file_path, files_update=None)
+        except Exception as e:
+            return WriteResult(error=f"Error writing file '{file_path}': {e}")
+
+    def edit(
+        self,
+        file_path: str,
+        old_string: str,
+        new_string: str,
+        replace_all: bool = False,
+    ) -> EditResult:
+        """Sync wrapper for aedit."""
+        return run_async_safely(
+            self.aedit(file_path, old_string, new_string, replace_all)
+        )
+
+    async def aedit(
+        self,
+        file_path: str,
+        old_string: str,
+        new_string: str,
+        replace_all: bool = False,
+    ) -> EditResult:
+        """Edit file by replacing strings."""
+        data = await self._get_file_data(file_path)
+        if data is None:
+            return EditResult(error=f"Error: File '{file_path}' not found")
+
+        content = "\n".join(data.get("content", []))
+        result = perform_string_replacement(content, old_string, new_string, replace_all)
+
+        if isinstance(result, str):
+            return EditResult(error=result)
+
+        new_content, occurrences = result
+        data["content"] = new_content.splitlines()
+
+        try:
+            await self._put_file_data(file_path, data)
+            return EditResult(
+                path=file_path, files_update=None, occurrences=int(occurrences)
+            )
+        except Exception as e:
+            return EditResult(error=f"Error editing file '{file_path}': {e}")
+
+    def grep_raw(
+        self, pattern: str, path: str | None = None, glob: str | None = None
+    ) -> list[GrepMatch] | str:
+        """Sync wrapper for agrep_raw."""
+        return run_async_safely(
+            self.agrep_raw(pattern, path, glob)
+        )
+
+    async def agrep_raw(
+        self, pattern: str, path: str | None = None, glob: str | None = None
+    ) -> list[GrepMatch] | str:
+        """Search for pattern in files."""
+        try:
+            regex = re.compile(pattern)
+        except re.error as e:
+            return f"Invalid regex pattern: {e}"
+
+        search_prefix = (path or "/").lstrip("/")
+        objects = await self._list_keys(search_prefix)
+        matches: list[GrepMatch] = []
+
+        for obj in objects:
+            vpath = self._virtual_path(obj["Key"])
+            filename = PurePosixPath(vpath).name
+
+            if glob and not wcglob.globmatch(filename, glob, flags=wcglob.BRACE):
+                continue
+
+            data = await self._get_file_data(vpath)
+            if data is None:
+                continue
+
+            for line_num, line in enumerate(data.get("content", []), 1):
+                if regex.search(line):
+                    matches.append({"path": vpath, "line": line_num, "text": line})
+
+        return matches
+
+    def glob_info(self, pattern: str, path: str = "/") -> list[FileInfo]:
+        """Sync wrapper for aglob_info."""
+        return run_async_safely(
+            self.aglob_info(pattern, path)
+        )
+
+    async def aglob_info(self, pattern: str, path: str = "/") -> list[FileInfo]:
+        """Find files matching a glob pattern."""
+        search_prefix = path.lstrip("/")
+        objects = await self._list_keys(search_prefix)
+        results: list[FileInfo] = []
+
+        for obj in objects:
+            vpath = self._virtual_path(obj["Key"])
+            rel_path = vpath[len(path) :].lstrip("/") if path != "/" else vpath[1:]
+
+            if fnmatch.fnmatch(rel_path, pattern) or fnmatch.fnmatch(vpath, pattern):
+                results.append(
+                    {
+                        "path": vpath,
+                        "is_dir": False,
+                        "size": obj.get("Size", 0),
+                        "modified_at": obj["LastModified"].isoformat()
+                        if "LastModified" in obj
+                        else None,
+                    }
+                )
+
+        results.sort(key=lambda x: x.get("path", ""))
+        return results
+
+    def upload_files(self, files: list[tuple[str, bytes]]) -> list[FileUploadResponse]:
+        """Sync wrapper for aupload_files."""
+        return run_async_safely(self.aupload_files(files))
+
+    async def aupload_files(
+        self, files: list[tuple[str, bytes]]
+    ) -> list[FileUploadResponse]:
+        """Upload multiple files."""
+        responses: list[FileUploadResponse] = []
+        async with self._client() as client:
+            for path, content in files:
+                try:
+                    key = self._s3_key(path)
+                    await client.put_object(
+                        Bucket=self._bucket, Key=key, Body=content
+                    )
+                    responses.append(FileUploadResponse(path=path, error=None))
+                except ClientError as e:
+                    code = e.response["Error"]["Code"]
+                    if code == "AccessDenied":
+                        responses.append(
+                            FileUploadResponse(path=path, error="permission_denied")
+                        )
+                    else:
+                        responses.append(
+                            FileUploadResponse(path=path, error="invalid_path")
+                        )
+                except Exception:
+                    responses.append(
+                        FileUploadResponse(path=path, error="invalid_path")
+                    )
+        return responses
+
+    def download_files(self, paths: list[str]) -> list[FileDownloadResponse]:
+        """Sync wrapper for adownload_files."""
+        return run_async_safely(self.adownload_files(paths))
+
+    async def adownload_files(self, paths: list[str]) -> list[FileDownloadResponse]:
+        """Download multiple files."""
+        responses: list[FileDownloadResponse] = []
+        async with self._client() as client:
+            for path in paths:
+                try:
+                    key = self._s3_key(path)
+                    response = await client.get_object(Bucket=self._bucket, Key=key)
+                    async with response["Body"] as stream:
+                        content = await stream.read()
+                    responses.append(
+                        FileDownloadResponse(path=path, content=content, error=None)
+                    )
+                except ClientError as e:
+                    code = e.response["Error"]["Code"]
+                    if code == "NoSuchKey":
+                        responses.append(
+                            FileDownloadResponse(
+                                path=path, content=None, error="file_not_found"
+                            )
+                        )
+                    elif code == "AccessDenied":
+                        responses.append(
+                            FileDownloadResponse(
+                                path=path, content=None, error="permission_denied"
+                            )
+                        )
+                    else:
+                        responses.append(
+                            FileDownloadResponse(
+                                path=path, content=None, error="invalid_path"
+                            )
+                        )
+        return responses
diff --git a/web/src/apis/agent_api.js b/web/src/apis/agent_api.js
index 3f269798..76fa8579 100644
--- a/web/src/apis/agent_api.js
+++ b/web/src/apis/agent_api.js
@@ -370,6 +370,15 @@ export const threadApi = {
     return `/api/chat/thread/${threadId}/artifacts/${encodedPath}${query}`
   },
 
+  /**
+   * 下载线程文件（带鉴权）
+   * @param {string} threadId
+   * @param {string} path
+   * @returns {Promise<Response>}
+   */
+  downloadThreadArtifact: (threadId, path) =>
+    apiGet(threadApi.getThreadArtifactUrl(threadId, path, true), {}, true, 'blob'),
+
   /**
    * 上传附件
    * @param {string} threadId
diff --git a/web/src/components/AgentChatComponent.vue b/web/src/components/AgentChatComponent.vue
index 19ca8e54..655bad17 100644
--- a/web/src/components/AgentChatComponent.vue
+++ b/web/src/components/AgentChatComponent.vue
@@ -386,10 +386,7 @@ const mentionConfig = computed(() => {
     })
 
   currentThreadAttachments.value.forEach((item) => {
-    const candidates = [
-      [item?.virtual_path, item?.artifact_url],
-      [item?.original_virtual_path, item?.original_artifact_url]
-    ]
+    const candidates = [[item?.path, item?.artifact_url]]
     candidates.forEach(([path, artifactUrl]) => {
       if (typeof path !== 'string' || !path) return
       if (!fileMap.has(path)) {
diff --git a/web/src/components/AgentPanel.vue b/web/src/components/AgentPanel.vue
index 7d4598d7..6e8dff33 100644
--- a/web/src/components/AgentPanel.vue
+++ b/web/src/components/AgentPanel.vue
@@ -386,18 +386,19 @@ const closeModal = () => {
   currentFilePath.value = ''
 }
 
-const downloadFile = (fileItem) => {
+const downloadFile = async (fileItem) => {
+  if (!props.threadId || !fileItem?.path) return
   try {
-    if (fileItem.artifact_url) {
-      const link = document.createElement('a')
-      link.href = fileItem.artifact_url
-      link.download = getFileName(fileItem)
-      link.target = '_blank'
-      document.body.appendChild(link)
-      link.click()
-      document.body.removeChild(link)
-      return
-    }
+    const response = await threadApi.downloadThreadArtifact(props.threadId, fileItem.path)
+    const blob = await response.blob()
+    const objectUrl = window.URL.createObjectURL(blob)
+    const link = document.createElement('a')
+    link.href = objectUrl
+    link.download = getFileName(fileItem)
+    document.body.appendChild(link)
+    link.click()
+    document.body.removeChild(link)
+    window.URL.revokeObjectURL(objectUrl)
   } catch (error) {
     console.error('下载文件失败:', error)
   }
diff --git a/web/src/components/AttachmentInputPanel.vue b/web/src/components/AttachmentInputPanel.vue
index 56cb1742..2f561420 100644
--- a/web/src/components/AttachmentInputPanel.vue
+++ b/web/src/components/AttachmentInputPanel.vue
@@ -4,7 +4,6 @@
       <input
         type="file"
         multiple
-        accept=".txt,.md,.docx,.html,.htm"
         :disabled="disabled || isUploading"
         @change="handleFileChange"
       />
@@ -53,7 +52,7 @@ const props = defineProps({
 const emit = defineEmits(['upload', 'remove'])
 
 const extensionsText = computed(() => {
-  if (!props.limits?.allowed_extensions?.length) return 'txt/md/docx/html'
+  if (!props.limits?.allowed_extensions?.length) return '任意文件'
   return props.limits.allowed_extensions.map((item) => item.replace('.', '')).join(' / ')
 })
 
@@ -64,10 +63,8 @@ const sizeHint = computed(() => {
 })
 
 const statusLabel = (item) => {
-  if (item.status === 'parsed') {
-    return item.truncated ? '已解析（截断）' : '已解析'
-  }
-  if (item.status === 'failed') return '解析失败'
+  if (item.status === 'uploaded') return '已上传'
+  if (item.status === 'failed') return '上传失败'
   return '处理中'
 }
 
diff --git a/web/src/components/AttachmentOptionsComponent.vue b/web/src/components/AttachmentOptionsComponent.vue
index 8c3b5bb1..3b3cda7f 100644
--- a/web/src/components/AttachmentOptionsComponent.vue
+++ b/web/src/components/AttachmentOptionsComponent.vue
@@ -6,12 +6,11 @@
           ref="fileInputRef"
           type="file"
           multiple
-          accept=".txt,.md,.docx,.html,.htm"
           :disabled="disabled"
           @change="handleFileChange"
           style="display: none"
         />
-        <a-tooltip title="支持 txt/md/docx/html 格式 ≤ 5 MB" placement="right">
+        <a-tooltip title="支持任意文件格式 ≤ 5 MB" placement="right">
           <div class="option-content">
             <FileText :size="14" class="option-icon" />
             <span class="option-text">添加附件</span>
diff --git a/web/src/composables/useAgentStreamHandler.js b/web/src/composables/useAgentStreamHandler.js
index 0941c5bd..b509f8f8 100644
--- a/web/src/composables/useAgentStreamHandler.js
+++ b/web/src/composables/useAgentStreamHandler.js
@@ -126,13 +126,13 @@ export function useAgentStreamHandler({
           currentAgentId: unref(currentAgentId),
           hasAgentState: !!chunk.agent_state,
           todoCount: Array.isArray(chunk.agent_state?.todos) ? chunk.agent_state.todos.length : 0,
-          fileKeys: chunk.agent_state?.files ? Object.keys(chunk.agent_state.files) : []
+          uploadCount: Array.isArray(chunk.agent_state?.uploads) ? chunk.agent_state.uploads.length : 0
         })
         if (chunk.agent_state) {
           console.log(`${debugPrefix}[agent_state_apply]`, {
             threadId,
             todos: chunk.agent_state?.todos || [],
-            files: chunk.agent_state?.files || []
+            uploads: chunk.agent_state?.uploads || []
           })
           threadState.agentState = chunk.agent_state
         } else {
@@ -164,7 +164,7 @@ export function useAgentStreamHandler({
               {
                 threadId,
                 todos: threadState.agentState?.todos || [],
-                files: threadState.agentState?.files || []
+                uploads: threadState.agentState?.uploads || []
               }
             )
           }