diff --git a/backend/package/yuxi/agents/backends/composite.py b/backend/package/yuxi/agents/backends/composite.py index 4507bd65..0b7149d6 100644 --- a/backend/package/yuxi/agents/backends/composite.py +++ b/backend/package/yuxi/agents/backends/composite.py @@ -94,6 +94,23 @@ def _extract_thread_id(runtime) -> str: raise ValueError("thread_id is required in runtime configurable context") +def _extract_user_id(runtime) -> str: + config = getattr(runtime, "config", None) + if isinstance(config, dict): + configurable = config.get("configurable", {}) + if isinstance(configurable, dict): + user_id = configurable.get("user_id") + if isinstance(user_id, str) and user_id.strip(): + return user_id.strip() + + context = getattr(runtime, "context", None) + user_id = getattr(context, "user_id", None) + if isinstance(user_id, str) and user_id.strip(): + return user_id.strip() + + raise ValueError("user_id is required in runtime configurable context") + + def _get_visible_knowledge_bases_from_runtime(runtime) -> list[dict]: context = getattr(runtime, "context", None) selected = getattr(context, "_visible_knowledge_bases", None) @@ -105,9 +122,10 @@ def _get_visible_knowledge_bases_from_runtime(runtime) -> list[dict]: def create_agent_composite_backend(runtime) -> CompositeBackend: visible_skills = _get_visible_skills_from_runtime(runtime) thread_id = _extract_thread_id(runtime) + user_id = _extract_user_id(runtime) visible_kbs = _get_visible_knowledge_bases_from_runtime(runtime) return CustomCompositeBackend( - default=ProvisionerSandboxBackend(thread_id=thread_id, visible_skills=visible_skills), + default=ProvisionerSandboxBackend(thread_id=thread_id, user_id=user_id, visible_skills=visible_skills), routes={ "/skills/": SelectedSkillsReadonlyBackend(selected_slugs=visible_skills), "/home/gem/kbs/": KnowledgeBaseReadonlyBackend(visible_kbs=visible_kbs), diff --git a/backend/package/yuxi/agents/backends/sandbox/backend.py b/backend/package/yuxi/agents/backends/sandbox/backend.py index a238eb50..63d49ca8 100644 --- a/backend/package/yuxi/agents/backends/sandbox/backend.py +++ b/backend/package/yuxi/agents/backends/sandbox/backend.py @@ -62,10 +62,13 @@ def _looks_like_binary(content: bytes) -> bool: class ProvisionerSandboxBackend(BaseSandbox): - def __init__(self, thread_id: str, *, visible_skills: list[str] | None = None): + def __init__(self, thread_id: str, *, user_id: str, visible_skills: list[str] | None = None): self._thread_id = str(thread_id or "").strip() if not self._thread_id: raise ValueError("thread_id is required for ProvisionerSandboxBackend") + self._user_id = str(user_id or "").strip() + if not self._user_id: + raise ValueError("user_id is required for ProvisionerSandboxBackend") self._visible_skills = list(visible_skills or []) self._provider = get_sandbox_provider() @@ -91,7 +94,7 @@ class ProvisionerSandboxBackend(BaseSandbox): def _get_client(self) -> Any: sync_thread_visible_skills(self._thread_id, self._visible_skills) - connection = self._provider.get(self._thread_id, create_if_missing=True) + connection = self._provider.get(self._thread_id, user_id=self._user_id, create_if_missing=True) if connection is None: raise RuntimeError(f"sandbox is unavailable for thread {self._thread_id}") diff --git a/backend/package/yuxi/agents/backends/sandbox/paths.py b/backend/package/yuxi/agents/backends/sandbox/paths.py index eafb85e9..b7dd23b9 100644 --- a/backend/package/yuxi/agents/backends/sandbox/paths.py +++ b/backend/package/yuxi/agents/backends/sandbox/paths.py @@ -6,7 +6,7 @@ from pathlib import Path from yuxi import config as conf from yuxi.utils.paths import OUTPUTS_DIR_NAME, UPLOADS_DIR_NAME, VIRTUAL_PATH_PREFIX, WORKSPACE_DIR_NAME -_SAFE_THREAD_ID_RE = re.compile(r"^[A-Za-z0-9_-]+$") +_SAFE_ID_RE = re.compile(r"^[A-Za-z0-9_-]+$") def get_virtual_path_prefix() -> str: @@ -17,7 +17,7 @@ def _validate_thread_id(thread_id: str) -> str: value = str(thread_id or "").strip() if not value: raise ValueError("thread_id is required") - if not _SAFE_THREAD_ID_RE.match(value): + if not _SAFE_ID_RE.match(value): raise ValueError("thread_id contains invalid characters") return value @@ -27,18 +27,28 @@ def _thread_root_dir(thread_id: str) -> Path: return Path(conf.save_dir) / "threads" / safe_thread_id / "user-data" -def _global_user_data_dir() -> Path: - """Return the shared host-side directory used for thread workspace files.""" - return Path(conf.save_dir) / "threads" / "shared" +def _validate_user_id(user_id: str) -> str: + value = str(user_id or "").strip() + if not value: + raise ValueError("user_id is required") + if not _SAFE_ID_RE.match(value): + raise ValueError("user_id contains invalid characters") + return value + + +def _global_user_data_dir(user_id: str) -> Path: + """Return the shared host-side directory used for one user's workspace files.""" + safe_user_id = _validate_user_id(user_id) + return Path(conf.save_dir) / "threads" / "shared" / safe_user_id def sandbox_user_data_dir(thread_id: str) -> Path: return _thread_root_dir(thread_id) -def sandbox_workspace_dir(thread_id: str) -> Path: +def sandbox_workspace_dir(thread_id: str, user_id: str) -> Path: _validate_thread_id(thread_id) - return _global_user_data_dir() / WORKSPACE_DIR_NAME + return _global_user_data_dir(user_id) / WORKSPACE_DIR_NAME def sandbox_uploads_dir(thread_id: str) -> Path: @@ -49,14 +59,14 @@ def sandbox_outputs_dir(thread_id: str) -> Path: return _thread_root_dir(thread_id) / OUTPUTS_DIR_NAME -def ensure_thread_dirs(thread_id: str) -> None: - _global_user_data_dir().mkdir(parents=True, exist_ok=True) - sandbox_workspace_dir(thread_id).mkdir(parents=True, exist_ok=True) +def ensure_thread_dirs(thread_id: str, user_id: str) -> None: + _global_user_data_dir(user_id).mkdir(parents=True, exist_ok=True) + sandbox_workspace_dir(thread_id, user_id).mkdir(parents=True, exist_ok=True) sandbox_uploads_dir(thread_id).mkdir(parents=True, exist_ok=True) sandbox_outputs_dir(thread_id).mkdir(parents=True, exist_ok=True) -def _resolve_user_data_base_dir(thread_id: str, relative_path: str) -> tuple[Path, Path]: +def _resolve_user_data_base_dir(thread_id: str, user_id: str, relative_path: str) -> tuple[Path, Path]: """Map a virtual user-data path to the correct host-side base directory.""" parts = Path(relative_path).parts if not parts: @@ -65,8 +75,8 @@ def _resolve_user_data_base_dir(thread_id: str, relative_path: str) -> tuple[Pat namespace = parts[0] if namespace == WORKSPACE_DIR_NAME: - # Workspace is shared across threads, so it lives outside the per-thread root. - base_dir = sandbox_workspace_dir(thread_id) + # Workspace is shared across one user's threads, so it lives outside the per-thread root. + base_dir = sandbox_workspace_dir(thread_id, user_id) target_path = base_dir.joinpath(*parts[1:]) if len(parts) > 1 else base_dir return base_dir.resolve(), target_path.resolve() if namespace == UPLOADS_DIR_NAME: @@ -82,7 +92,7 @@ def _resolve_user_data_base_dir(thread_id: str, relative_path: str) -> tuple[Pat return base_dir.resolve(), (base_dir / relative_path).resolve() -def resolve_virtual_path(thread_id: str, virtual_path: str) -> Path: +def resolve_virtual_path(thread_id: str, virtual_path: str, *, user_id: str) -> Path: clean_virtual_path = "/" + str(virtual_path or "").strip().lstrip("/") virtual_prefix = get_virtual_path_prefix() @@ -90,7 +100,7 @@ def resolve_virtual_path(thread_id: str, virtual_path: str) -> Path: raise ValueError(f"path must start with {virtual_prefix}") relative_path = clean_virtual_path[len(virtual_prefix) :].lstrip("/") - base_dir, target_path = _resolve_user_data_base_dir(thread_id, relative_path) + base_dir, target_path = _resolve_user_data_base_dir(thread_id, user_id, relative_path) try: target_path.relative_to(base_dir) @@ -100,10 +110,10 @@ def resolve_virtual_path(thread_id: str, virtual_path: str) -> Path: return target_path -def virtual_path_for_thread_file(thread_id: str, path: str | Path) -> str: +def virtual_path_for_thread_file(thread_id: str, path: str | Path, *, user_id: str) -> str: target_path = Path(path).resolve() thread_root = sandbox_user_data_dir(thread_id).resolve() - global_workspace_root = sandbox_workspace_dir(thread_id).resolve() + global_workspace_root = sandbox_workspace_dir(thread_id, user_id).resolve() try: relative_path = target_path.relative_to(global_workspace_root) diff --git a/backend/package/yuxi/agents/backends/sandbox/provider.py b/backend/package/yuxi/agents/backends/sandbox/provider.py index 9b1bc079..32856774 100644 --- a/backend/package/yuxi/agents/backends/sandbox/provider.py +++ b/backend/package/yuxi/agents/backends/sandbox/provider.py @@ -19,6 +19,7 @@ def sandbox_id_for_thread(thread_id: str) -> str: @dataclass(slots=True) class SandboxConnection: thread_id: str + user_id: str sandbox_id: str sandbox_url: str @@ -48,9 +49,10 @@ class ProvisionerSandboxProvider: self._thread_locks[thread_id] = lock return lock - def _record_to_connection(self, thread_id: str, record: SandboxRecord) -> SandboxConnection: + def _record_to_connection(self, thread_id: str, user_id: str, record: SandboxRecord) -> SandboxConnection: connection = SandboxConnection( thread_id=thread_id, + user_id=user_id, sandbox_id=record.sandbox_id, sandbox_url=record.sandbox_url, ) @@ -73,7 +75,7 @@ class ProvisionerSandboxProvider: self._last_touch_at[connection.thread_id] = time.time() return is_alive - def acquire(self, thread_id: str) -> str: + def acquire(self, thread_id: str, *, user_id: str) -> str: lock = self._thread_lock(thread_id) with lock: current = self._connections.get(thread_id) @@ -91,14 +93,14 @@ class ProvisionerSandboxProvider: record = self._client.discover(sandbox_id) if record is None: logger.info(f"Creating sandbox {sandbox_id} for thread {thread_id}") - record = self._client.create(sandbox_id, thread_id) + record = self._client.create(sandbox_id, thread_id, user_id) else: logger.info(f"Reusing sandbox {sandbox_id} for thread {thread_id}") - connection = self._record_to_connection(thread_id, record) + connection = self._record_to_connection(thread_id, user_id, record) return connection.sandbox_id - def get(self, thread_id: str, *, create_if_missing: bool = False) -> SandboxConnection | None: + def get(self, thread_id: str, *, user_id: str, create_if_missing: bool = False) -> SandboxConnection | None: lock = self._thread_lock(thread_id) with lock: current = self._connections.get(thread_id) @@ -111,19 +113,19 @@ class ProvisionerSandboxProvider: 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 + if current.user_id == user_id: + return current + self._connections.pop(thread_id, None) + self._last_touch_at.pop(thread_id, None) sandbox_id = sandbox_id_for_thread(thread_id) record = self._client.discover(sandbox_id) if record is None: if not create_if_missing: return None - record = self._client.create(sandbox_id, thread_id) + record = self._client.create(sandbox_id, thread_id, user_id) - return self._record_to_connection(thread_id, record) + return self._record_to_connection(thread_id, user_id, record) def shutdown(self) -> None: with self._lock: diff --git a/backend/package/yuxi/agents/backends/sandbox/provisioner_client.py b/backend/package/yuxi/agents/backends/sandbox/provisioner_client.py index 19e7e4fe..f738a575 100644 --- a/backend/package/yuxi/agents/backends/sandbox/provisioner_client.py +++ b/backend/package/yuxi/agents/backends/sandbox/provisioner_client.py @@ -29,11 +29,11 @@ class ProvisionerClient: response = self._request("GET", "/health") return response.status_code == 200 - def create(self, sandbox_id: str, thread_id: str) -> SandboxRecord: + def create(self, sandbox_id: str, thread_id: str, user_id: str) -> SandboxRecord: response = self._request( "POST", "/api/sandboxes", - json={"sandbox_id": sandbox_id, "thread_id": thread_id}, + json={"sandbox_id": sandbox_id, "thread_id": thread_id, "user_id": user_id}, ) if response.status_code >= 400: raise RuntimeError(f"failed to create sandbox {sandbox_id}: {response.status_code} {response.text}") diff --git a/backend/package/yuxi/agents/toolkits/buildin/tools.py b/backend/package/yuxi/agents/toolkits/buildin/tools.py index fc946844..add9b1f5 100644 --- a/backend/package/yuxi/agents/toolkits/buildin/tools.py +++ b/backend/package/yuxi/agents/toolkits/buildin/tools.py @@ -83,8 +83,11 @@ def _normalize_presented_artifact_path(filepath: str, runtime: ToolRuntime) -> s thread_id = getattr(runtime_context, "thread_id", None) if not thread_id: raise ValueError("当前运行时缺少 thread_id") + user_id = getattr(runtime_context, "user_id", None) + if not user_id: + raise ValueError("当前运行时缺少 user_id") - ensure_thread_dirs(thread_id) + ensure_thread_dirs(thread_id, str(user_id)) outputs_dir = sandbox_outputs_dir(thread_id).resolve() normalized_input = str(filepath or "").strip() if not normalized_input: @@ -93,7 +96,7 @@ def _normalize_presented_artifact_path(filepath: str, runtime: ToolRuntime) -> s stripped = normalized_input.lstrip("/") virtual_prefix = VIRTUAL_PATH_PREFIX.lstrip("/") if stripped == virtual_prefix or stripped.startswith(f"{virtual_prefix}/"): - actual_path = resolve_virtual_path(thread_id, normalized_input) + actual_path = resolve_virtual_path(thread_id, normalized_input, user_id=str(user_id)) else: actual_path = Path(normalized_input).expanduser().resolve() diff --git a/backend/package/yuxi/services/conversation_service.py b/backend/package/yuxi/services/conversation_service.py index 01e33f10..e71dfa0f 100644 --- a/backend/package/yuxi/services/conversation_service.py +++ b/backend/package/yuxi/services/conversation_service.py @@ -209,12 +209,13 @@ def serialize_attachment(record: dict) -> dict: async def _materialize_attachment_files( *, thread_id: str, + user_id: str, upload: UploadFile, file_name: str, file_content: bytes, ) -> dict: """将原始附件与可选 markdown 副本落盘到线程 user-data。""" - ensure_thread_dirs(thread_id) + ensure_thread_dirs(thread_id, user_id) upload_virtual_path = _make_upload_virtual_path(file_name) uploads_dir = sandbox_uploads_dir(thread_id) @@ -384,6 +385,7 @@ async def upload_thread_attachment_view( raise HTTPException(status_code=400, detail=f"附件过大,当前仅支持 {max_size_mb} MB 以内的文件") materialized = await _materialize_attachment_files( thread_id=thread_id, + user_id=str(conversation.user_id), upload=file, file_name=file_name, file_content=file_content, diff --git a/backend/package/yuxi/services/filesystem_service.py b/backend/package/yuxi/services/filesystem_service.py index e80ca679..1bc1b8b7 100644 --- a/backend/package/yuxi/services/filesystem_service.py +++ b/backend/package/yuxi/services/filesystem_service.py @@ -67,7 +67,7 @@ async def _resolve_filesystem_state( runtime_context.user_id = str(user.id) await resolve_visible_knowledge_bases_for_context(runtime_context) - sandbox_backend = ProvisionerSandboxBackend(thread_id=thread_id) + sandbox_backend = ProvisionerSandboxBackend(thread_id=thread_id, user_id=str(user.id)) return conversation, runtime_context, sandbox_backend diff --git a/backend/package/yuxi/services/thread_files_service.py b/backend/package/yuxi/services/thread_files_service.py index c30cf63a..3e04139a 100644 --- a/backend/package/yuxi/services/thread_files_service.py +++ b/backend/package/yuxi/services/thread_files_service.py @@ -35,12 +35,13 @@ async def list_thread_files_view( recursive: bool = False, ) -> dict: conv_repo = ConversationRepository(db) - await require_user_conversation(conv_repo, thread_id, str(current_user_id)) + conversation = await require_user_conversation(conv_repo, thread_id, str(current_user_id)) + user_id = str(conversation.user_id) - ensure_thread_dirs(thread_id) + ensure_thread_dirs(thread_id, user_id) virtual_path = path or _get_virtual_root() try: - actual_path = resolve_virtual_path(thread_id, virtual_path) + actual_path = resolve_virtual_path(thread_id, virtual_path, user_id=user_id) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc @@ -51,16 +52,16 @@ async def list_thread_files_view( if recursive: if virtual_path.rstrip("/") == _get_virtual_root(): - return _list_user_data_root_entries(thread_id, virtual_path, recursive=True) - return _list_files_recursive(thread_id, actual_path, virtual_path) + return _list_user_data_root_entries(thread_id, user_id, virtual_path, recursive=True) + return _list_files_recursive(thread_id, user_id, actual_path, virtual_path) if virtual_path.rstrip("/") == _get_virtual_root(): - return _list_user_data_root_entries(thread_id, virtual_path) + return _list_user_data_root_entries(thread_id, user_id, virtual_path) entries: list[dict[str, Any]] = [] for child in sorted(actual_path.iterdir(), key=lambda item: (not item.is_dir(), item.name.lower())): stat = child.stat() - child_virtual_path = virtual_path_for_thread_file(thread_id, child) + child_virtual_path = virtual_path_for_thread_file(thread_id, child, user_id=user_id) entries.append( { "path": child_virtual_path, @@ -77,13 +78,13 @@ async def list_thread_files_view( return {"path": virtual_path, "files": entries} -def _list_user_data_root_entries(thread_id: str, virtual_path: str, recursive: bool = False) -> dict: - """List the thread root and inject the shared workspace entry if needed.""" +def _list_user_data_root_entries(thread_id: str, user_id: str, virtual_path: str, recursive: bool = False) -> dict: + """List the thread root and inject the user workspace entry if needed.""" entries: list[dict[str, Any]] = [] thread_root = sandbox_user_data_dir(thread_id) for child in sorted(thread_root.iterdir(), key=lambda item: (not item.is_dir(), item.name.lower())): stat = child.stat() - child_virtual_path = virtual_path_for_thread_file(thread_id, child) + child_virtual_path = virtual_path_for_thread_file(thread_id, child, user_id=user_id) if child.is_dir() and not child_virtual_path.endswith("/"): child_virtual_path = f"{child_virtual_path}/" entries.append( @@ -99,11 +100,11 @@ def _list_user_data_root_entries(thread_id: str, virtual_path: str, recursive: b } ) if recursive and child.is_dir(): - nested = _list_files_recursive(thread_id, child, child_virtual_path) + nested = _list_files_recursive(thread_id, user_id, child, child_virtual_path) entries.extend(nested["files"]) - workspace_dir = sandbox_workspace_dir(thread_id) - workspace_virtual_path = virtual_path_for_thread_file(thread_id, workspace_dir) + workspace_dir = sandbox_workspace_dir(thread_id, user_id) + workspace_virtual_path = virtual_path_for_thread_file(thread_id, workspace_dir, user_id=user_id) if workspace_virtual_path.rstrip("/") not in {str(entry["path"]).rstrip("/") for entry in entries}: # workspace lives outside the per-thread root, so expose it as a top-level entry. stat = workspace_dir.stat() @@ -120,12 +121,12 @@ def _list_user_data_root_entries(thread_id: str, virtual_path: str, recursive: b } ) if recursive: - nested = _list_files_recursive(thread_id, workspace_dir, workspace_virtual_path) + nested = _list_files_recursive(thread_id, user_id, workspace_dir, workspace_virtual_path) entries.extend(nested["files"]) return {"path": virtual_path, "files": entries} -def _list_files_recursive(thread_id: str, actual_path: Path, virtual_path: str) -> dict: +def _list_files_recursive(thread_id: str, user_id: str, actual_path: Path, virtual_path: str) -> dict: """Recursively scan a directory while preserving viewer virtual paths.""" entries: list[dict[str, Any]] = [] @@ -133,7 +134,7 @@ def _list_files_recursive(thread_id: str, actual_path: Path, virtual_path: str) try: for child in sorted(base_actual_path.iterdir(), key=lambda item: (not item.is_dir(), item.name.lower())): stat = child.stat() - child_virtual_path = virtual_path_for_thread_file(thread_id, child) + child_virtual_path = virtual_path_for_thread_file(thread_id, child, user_id=user_id) entries.append( { "path": child_virtual_path, @@ -165,10 +166,11 @@ async def read_thread_file_content_view( limit: int = 2000, ) -> dict: conv_repo = ConversationRepository(db) - await require_user_conversation(conv_repo, thread_id, str(current_user_id)) + conversation = await require_user_conversation(conv_repo, thread_id, str(current_user_id)) + user_id = str(conversation.user_id) try: - actual_path = resolve_virtual_path(thread_id, path) + actual_path = resolve_virtual_path(thread_id, path, user_id=user_id) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc @@ -201,13 +203,14 @@ async def resolve_thread_artifact_view( path: str, ) -> Path: conv_repo = ConversationRepository(db) - await require_user_conversation(conv_repo, thread_id, str(current_user_id)) + conversation = await require_user_conversation(conv_repo, thread_id, str(current_user_id)) + user_id = str(conversation.user_id) - ensure_thread_dirs(thread_id) + ensure_thread_dirs(thread_id, user_id) normalized = "/" + path.lstrip("/") try: - actual_path = resolve_virtual_path(thread_id, normalized) + actual_path = resolve_virtual_path(thread_id, normalized, user_id=user_id) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc @@ -218,7 +221,7 @@ async def resolve_thread_artifact_view( resolved_path = actual_path.resolve() allowed_roots = ( - sandbox_workspace_dir(thread_id).resolve(), + sandbox_workspace_dir(thread_id, user_id).resolve(), sandbox_uploads_dir(thread_id).resolve(), sandbox_outputs_dir(thread_id).resolve(), ) @@ -242,14 +245,17 @@ async def save_thread_artifact_to_workspace_view( path=path, ) - target_dir = sandbox_workspace_dir(thread_id) / "saved_artifacts" + conv_repo = ConversationRepository(db) + conversation = await require_user_conversation(conv_repo, thread_id, str(current_user_id)) + user_id = str(conversation.user_id) + target_dir = sandbox_workspace_dir(thread_id, user_id) / "saved_artifacts" target_dir.mkdir(parents=True, exist_ok=True) target_path = _next_available_artifact_path(target_dir, source_path.name) with source_path.open("rb") as src, target_path.open("wb") as dst: shutil.copyfileobj(src, dst) - saved_virtual_path = virtual_path_for_thread_file(thread_id, target_path) + saved_virtual_path = virtual_path_for_thread_file(thread_id, target_path, user_id=user_id) return { "name": target_path.name, "source_path": "/" + path.lstrip("/"), diff --git a/backend/package/yuxi/services/viewer_filesystem_service.py b/backend/package/yuxi/services/viewer_filesystem_service.py index 780f9b25..031cbdfd 100644 --- a/backend/package/yuxi/services/viewer_filesystem_service.py +++ b/backend/package/yuxi/services/viewer_filesystem_service.py @@ -226,13 +226,13 @@ def _sort_entries(entries: list[dict]) -> list[dict]: ) -def _list_local_entries(thread_id: str, actual_path) -> list[dict]: +def _list_local_entries(thread_id: str, user_id: str, actual_path) -> list[dict]: """List a local directory and remap children back into viewer virtual paths.""" entries: list[dict] = [] for child in sorted(actual_path.iterdir(), key=lambda item: (not item.is_dir(), item.name.lower())): stat = child.stat() is_dir = child.is_dir() - display_path = virtual_path_for_thread_file(thread_id, child) + display_path = virtual_path_for_thread_file(thread_id, child, user_id=user_id) if is_dir and not display_path.endswith("/"): display_path = f"{display_path}/" entries.append( @@ -247,12 +247,12 @@ def _list_local_entries(thread_id: str, actual_path) -> list[dict]: return entries -def _list_user_data_root_entries(thread_id: str) -> list[dict]: - """Expose thread-root files while keeping the shared workspace entry visible.""" - entries = _list_local_entries(thread_id, sandbox_user_data_dir(thread_id)) +def _list_user_data_root_entries(thread_id: str, user_id: str) -> list[dict]: + """Expose thread-root files while keeping the user workspace entry visible.""" + entries = _list_local_entries(thread_id, user_id, sandbox_user_data_dir(thread_id)) visible_paths = {str(entry.get("path") or "").rstrip("/") for entry in entries} - workspace_dir = sandbox_workspace_dir(thread_id) - workspace_virtual_path = virtual_path_for_thread_file(thread_id, workspace_dir).rstrip("/") + workspace_dir = sandbox_workspace_dir(thread_id, user_id) + workspace_virtual_path = virtual_path_for_thread_file(thread_id, workspace_dir, user_id=user_id).rstrip("/") if workspace_virtual_path not in visible_paths: # workspace is stored outside the per-thread root, so add it explicitly when needed. stat = workspace_dir.stat() @@ -327,16 +327,17 @@ async def list_viewer_filesystem_tree( try: if _is_user_data_path(normalized_path): - ensure_thread_dirs(thread_id) + user_id = str(current_user.id) + ensure_thread_dirs(thread_id, user_id) if normalized_path == USER_DATA_PATH: - entries = await asyncio.to_thread(_list_user_data_root_entries, thread_id) + entries = await asyncio.to_thread(_list_user_data_root_entries, thread_id, user_id) return {"entries": _sort_entries(entries)} - actual_path = resolve_virtual_path(thread_id, normalized_path) + actual_path = resolve_virtual_path(thread_id, normalized_path, user_id=user_id) if not actual_path.exists(): return {"entries": []} if not actual_path.is_dir(): raise HTTPException(status_code=400, detail="当前路径不是目录") - entries = await asyncio.to_thread(_list_local_entries, thread_id, actual_path) + entries = await asyncio.to_thread(_list_local_entries, thread_id, user_id, actual_path) return {"entries": _sort_entries(entries)} if _is_skills_path(normalized_path): @@ -378,7 +379,7 @@ async def read_viewer_file_content( try: if _is_user_data_path(normalized_path): - actual_path = resolve_virtual_path(thread_id, normalized_path) + actual_path = resolve_virtual_path(thread_id, normalized_path, user_id=str(current_user.id)) if not actual_path.exists(): raise HTTPException(status_code=404, detail="文件不存在") if not actual_path.is_file(): @@ -471,7 +472,7 @@ async def download_viewer_file( try: if _is_user_data_path(normalized_path): - actual_path = resolve_virtual_path(thread_id, normalized_path) + actual_path = resolve_virtual_path(thread_id, normalized_path, user_id=str(current_user.id)) if not actual_path.exists(): raise HTTPException(status_code=404, detail="文件不存在") if not actual_path.is_file(): @@ -545,7 +546,7 @@ async def delete_viewer_file( raise HTTPException(status_code=400, detail="当前目录不允许删除") try: - actual_path = resolve_virtual_path(thread_id, normalized_path) + actual_path = resolve_virtual_path(thread_id, normalized_path, user_id=str(current_user.id)) if not actual_path.exists(): raise HTTPException(status_code=404, detail="文件不存在") if actual_path.is_dir(): diff --git a/backend/test/e2e/conftest.py b/backend/test/e2e/conftest.py index 70a0e6a3..6e5d06dd 100644 --- a/backend/test/e2e/conftest.py +++ b/backend/test/e2e/conftest.py @@ -26,8 +26,7 @@ E2E_TIMEOUT = httpx.Timeout(300.0, connect=10.0) def _require_e2e_credentials() -> tuple[str, str]: if not E2E_USERNAME or not E2E_PASSWORD: pytest.skip( - "E2E credentials are not configured via E2E_USERNAME / E2E_PASSWORD " - "or TEST_USERNAME / TEST_PASSWORD." + "E2E credentials are not configured via E2E_USERNAME / E2E_PASSWORD or TEST_USERNAME / TEST_PASSWORD." ) return E2E_USERNAME, E2E_PASSWORD @@ -58,6 +57,15 @@ async def e2e_headers(e2e_client: httpx.AsyncClient) -> dict[str, str]: @pytest_asyncio.fixture(scope="function") async def e2e_agent_context(e2e_client: httpx.AsyncClient, e2e_headers: dict[str, str]) -> dict[str, str | int]: + me_response = await e2e_client.get("/api/auth/me", headers=e2e_headers) + if me_response.status_code != 200: + pytest.fail( + f"Failed to fetch current user for E2E tests (status={me_response.status_code}): {me_response.text}" + ) + user_id = me_response.json().get("id") + if user_id is None: + pytest.fail("Current user payload missing id field for E2E tests.") + default_response = await e2e_client.get("/api/chat/default_agent", headers=e2e_headers) default_agent_id = None if default_response.status_code == 200: @@ -77,8 +85,7 @@ async def e2e_agent_context(e2e_client: httpx.AsyncClient, e2e_headers: dict[str config_response = await e2e_client.get(f"/api/chat/agent/{agent_id}/configs", headers=e2e_headers) if config_response.status_code != 200: pytest.fail( - "Failed to list agent configs for E2E tests " - f"(status={config_response.status_code}): {config_response.text}" + f"Failed to list agent configs for E2E tests (status={config_response.status_code}): {config_response.text}" ) configs = config_response.json().get("configs") or [] @@ -89,4 +96,4 @@ async def e2e_agent_context(e2e_client: httpx.AsyncClient, e2e_headers: dict[str if not config_id: pytest.fail(f"Agent config payload missing id field for agent {agent_id}.") - return {"agent_id": agent_id, "agent_config_id": int(config_id)} + return {"agent_id": agent_id, "agent_config_id": int(config_id), "user_id": int(user_id)} diff --git a/backend/test/e2e/test_viewer_filesystem_e2e.py b/backend/test/e2e/test_viewer_filesystem_e2e.py index 64151357..567065dc 100644 --- a/backend/test/e2e/test_viewer_filesystem_e2e.py +++ b/backend/test/e2e/test_viewer_filesystem_e2e.py @@ -103,14 +103,15 @@ async def test_viewer_filesystem_e2e_respects_workspace_sharing_and_thread_local e2e_agent_context: dict[str, str | int], ): agent_id = str(e2e_agent_context["agent_id"]) + user_id = str(e2e_agent_context["user_id"]) thread_id = await _create_thread(e2e_client, e2e_headers, agent_id) other_thread_id = await _create_thread(e2e_client, e2e_headers, agent_id) - ensure_thread_dirs(thread_id) - ensure_thread_dirs(other_thread_id) + ensure_thread_dirs(thread_id, user_id) + ensure_thread_dirs(other_thread_id, user_id) (sandbox_user_data_dir(thread_id) / "root-note.txt").write_text("root-visible\n", encoding="utf-8") - (sandbox_workspace_dir(thread_id) / "demo.py").write_text("print(42)\n", encoding="utf-8") + (sandbox_workspace_dir(thread_id, user_id) / "demo.py").write_text("print(42)\n", encoding="utf-8") uploads_dir = sandbox_uploads_dir(thread_id) / "attachments" uploads_dir.mkdir(parents=True, exist_ok=True) @@ -218,10 +219,11 @@ async def test_viewer_filesystem_e2e_deletes_workspace_directory_recursively( e2e_agent_context: dict[str, str | int], ): agent_id = str(e2e_agent_context["agent_id"]) + user_id = str(e2e_agent_context["user_id"]) thread_id = await _create_thread(e2e_client, e2e_headers, agent_id) - ensure_thread_dirs(thread_id) - target_dir = sandbox_workspace_dir(thread_id) / "delete-dir" + ensure_thread_dirs(thread_id, user_id) + target_dir = sandbox_workspace_dir(thread_id, user_id) / "delete-dir" nested_dir = target_dir / "deep" nested_dir.mkdir(parents=True) (nested_dir / "artifact.txt").write_text("delete me\n", encoding="utf-8") diff --git a/backend/test/integration/api/test_chat_router.py b/backend/test/integration/api/test_chat_router.py index 91bcdd71..154a30e0 100644 --- a/backend/test/integration/api/test_chat_router.py +++ b/backend/test/integration/api/test_chat_router.py @@ -92,10 +92,11 @@ async def test_setting_default_agent_requires_admin(test_client, admin_headers, async def test_save_thread_artifact_to_workspace_copies_output_file(test_client, standard_user): headers = standard_user["headers"] + user_id = str(standard_user["user"]["id"]) thread_id = await _create_thread_for_user(test_client, headers) filename = f"artifact-{uuid.uuid4().hex[:8]}.md" - ensure_thread_dirs(thread_id) + ensure_thread_dirs(thread_id, user_id) source_path = sandbox_user_data_dir(thread_id) / "outputs" / filename source_path.write_text("# artifact\n", encoding="utf-8") @@ -111,7 +112,7 @@ async def test_save_thread_artifact_to_workspace_copies_output_file(test_client, assert payload["source_path"] == f"/home/gem/user-data/outputs/{filename}" assert payload["saved_path"] == f"/home/gem/user-data/workspace/saved_artifacts/{filename}" - saved_path = sandbox_workspace_dir(thread_id) / "saved_artifacts" / filename + saved_path = sandbox_workspace_dir(thread_id, user_id) / "saved_artifacts" / filename assert saved_path.exists() assert saved_path.read_text(encoding="utf-8") == "# artifact\n" @@ -122,11 +123,12 @@ async def test_save_thread_artifact_to_workspace_copies_output_file(test_client, async def test_save_thread_artifact_to_workspace_auto_renames_conflicts(test_client, standard_user): headers = standard_user["headers"] + user_id = str(standard_user["user"]["id"]) thread_id = await _create_thread_for_user(test_client, headers) filename = f"artifact-{uuid.uuid4().hex[:8]}.txt" renamed_filename = filename.replace(".txt", " (1).txt") - ensure_thread_dirs(thread_id) + ensure_thread_dirs(thread_id, user_id) source_path = sandbox_user_data_dir(thread_id) / "outputs" / filename source_path.write_text("first\n", encoding="utf-8") @@ -150,14 +152,15 @@ async def test_save_thread_artifact_to_workspace_auto_renames_conflicts(test_cli assert first_payload["saved_path"] == f"/home/gem/user-data/workspace/saved_artifacts/{filename}" assert second_payload["saved_path"] == f"/home/gem/user-data/workspace/saved_artifacts/{renamed_filename}" - first_saved = sandbox_workspace_dir(thread_id) / "saved_artifacts" / filename - second_saved = sandbox_workspace_dir(thread_id) / "saved_artifacts" / renamed_filename + first_saved = sandbox_workspace_dir(thread_id, user_id) / "saved_artifacts" / filename + second_saved = sandbox_workspace_dir(thread_id, user_id) / "saved_artifacts" / renamed_filename assert first_saved.read_text(encoding="utf-8") == "first\n" assert second_saved.read_text(encoding="utf-8") == "second\n" async def test_save_thread_artifact_to_workspace_rejects_invalid_paths(test_client, standard_user): headers = standard_user["headers"] + user_id = str(standard_user["user"]["id"]) thread_id = await _create_thread_for_user(test_client, headers) invalid_response = await test_client.post( @@ -167,8 +170,8 @@ async def test_save_thread_artifact_to_workspace_rejects_invalid_paths(test_clie ) assert invalid_response.status_code == 404, invalid_response.text - ensure_thread_dirs(thread_id) - directory_path = sandbox_workspace_dir(thread_id) / "nested-dir" + ensure_thread_dirs(thread_id, user_id) + directory_path = sandbox_workspace_dir(thread_id, user_id) / "nested-dir" directory_path.mkdir(parents=True, exist_ok=True) directory_response = await test_client.post( f"/api/chat/thread/{thread_id}/artifacts/save", diff --git a/backend/test/integration/api/test_viewer_filesystem_router.py b/backend/test/integration/api/test_viewer_filesystem_router.py index 6f3ea5d3..23ebf5dc 100644 --- a/backend/test/integration/api/test_viewer_filesystem_router.py +++ b/backend/test/integration/api/test_viewer_filesystem_router.py @@ -123,10 +123,11 @@ async def test_viewer_tree_root_does_not_require_sandbox_listing(test_client, st async def test_viewer_tree_user_data_uses_local_thread_directory(test_client, standard_user, monkeypatch): headers = standard_user["headers"] + user_id = str(standard_user["user"]["id"]) thread_id = await _create_thread_for_user(test_client, headers) - ensure_thread_dirs(thread_id) - actual_path = sandbox_workspace_dir(thread_id) / "viewer_tree_demo.txt" + ensure_thread_dirs(thread_id, user_id) + actual_path = sandbox_workspace_dir(thread_id, user_id) / "viewer_tree_demo.txt" actual_path.write_text("viewer tree", encoding="utf-8") class _FailingSandbox: @@ -161,9 +162,10 @@ async def test_viewer_tree_user_data_uses_local_thread_directory(test_client, st async def test_viewer_tree_user_data_root_keeps_thread_root_files_visible(test_client, standard_user): headers = standard_user["headers"] + user_id = str(standard_user["user"]["id"]) thread_id = await _create_thread_for_user(test_client, headers) - ensure_thread_dirs(thread_id) + ensure_thread_dirs(thread_id, user_id) root_file = sandbox_user_data_dir(thread_id) / "root-note.txt" root_file.write_text("visible at root", encoding="utf-8") @@ -179,13 +181,14 @@ async def test_viewer_tree_user_data_root_keeps_thread_root_files_visible(test_c assert "/home/gem/user-data/workspace/" in paths -async def test_workspace_is_shared_across_threads_but_uploads_remain_thread_local(test_client, standard_user): +async def test_workspace_is_shared_across_same_user_threads_but_uploads_remain_thread_local(test_client, standard_user): headers = standard_user["headers"] + user_id = str(standard_user["user"]["id"]) thread_id = await _create_thread_for_user(test_client, headers) other_thread_id = await _create_thread_for_user(test_client, headers) - ensure_thread_dirs(thread_id) - shared_path = sandbox_workspace_dir(thread_id) / "shared_across_threads.txt" + ensure_thread_dirs(thread_id, user_id) + shared_path = sandbox_workspace_dir(thread_id, user_id) / "shared_across_threads.txt" shared_path.write_text("shared workspace", encoding="utf-8") upload_path = await _upload_attachment_file(test_client, thread_id, headers, "thread-local.txt", "private upload\n") @@ -232,12 +235,13 @@ async def test_viewer_file_returns_raw_content_without_line_numbers(test_client, async def test_viewer_file_returns_unsupported_for_binary_payload(test_client, standard_user): headers = standard_user["headers"] + user_id = str(standard_user["user"]["id"]) thread_id = await _create_thread_for_user(test_client, headers) - ensure_thread_dirs(thread_id) - actual_path = sandbox_workspace_dir(thread_id) / "binary.bin" + ensure_thread_dirs(thread_id, user_id) + actual_path = sandbox_workspace_dir(thread_id, user_id) / "binary.bin" actual_path.write_bytes(b"\x00\x01\x02\x03binary") - file_path = virtual_path_for_thread_file(thread_id, actual_path) + file_path = virtual_path_for_thread_file(thread_id, actual_path, user_id=user_id) response = await test_client.get( "/api/viewer/filesystem/file", @@ -254,14 +258,15 @@ async def test_viewer_file_returns_unsupported_for_binary_payload(test_client, s async def test_viewer_file_returns_image_preview_metadata(test_client, standard_user): headers = standard_user["headers"] + user_id = str(standard_user["user"]["id"]) thread_id = await _create_thread_for_user(test_client, headers) - ensure_thread_dirs(thread_id) - actual_path = sandbox_workspace_dir(thread_id) / "demo.png" + ensure_thread_dirs(thread_id, user_id) + actual_path = sandbox_workspace_dir(thread_id, user_id) / "demo.png" actual_path.write_bytes( b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x06\x00\x00\x00\x1f\x15\xc4\x89" ) - file_path = virtual_path_for_thread_file(thread_id, actual_path) + file_path = virtual_path_for_thread_file(thread_id, actual_path, user_id=user_id) response = await test_client.get( "/api/viewer/filesystem/file", @@ -278,12 +283,13 @@ async def test_viewer_file_returns_image_preview_metadata(test_client, standard_ async def test_viewer_file_returns_pdf_preview_metadata(test_client, standard_user): headers = standard_user["headers"] + user_id = str(standard_user["user"]["id"]) thread_id = await _create_thread_for_user(test_client, headers) - ensure_thread_dirs(thread_id) - actual_path = sandbox_workspace_dir(thread_id) / "demo.pdf" + ensure_thread_dirs(thread_id, user_id) + actual_path = sandbox_workspace_dir(thread_id, user_id) / "demo.pdf" actual_path.write_bytes(b"%PDF-1.4\n1 0 obj\n<<>>\nendobj\ntrailer\n<<>>\n%%EOF") - file_path = virtual_path_for_thread_file(thread_id, actual_path) + file_path = virtual_path_for_thread_file(thread_id, actual_path, user_id=user_id) response = await test_client.get( "/api/viewer/filesystem/file", @@ -317,13 +323,14 @@ async def test_viewer_download_returns_attachment_response(test_client, standard async def test_viewer_download_returns_full_file_for_large_user_data_content(test_client, standard_user): headers = standard_user["headers"] + user_id = str(standard_user["user"]["id"]) thread_id = await _create_thread_for_user(test_client, headers) large_content = "0123456789abcdef" * 4096 - ensure_thread_dirs(thread_id) - actual_path = sandbox_workspace_dir(thread_id) / "large_download.txt" + ensure_thread_dirs(thread_id, user_id) + actual_path = sandbox_workspace_dir(thread_id, user_id) / "large_download.txt" actual_path.write_text(large_content, encoding="utf-8") - file_path = virtual_path_for_thread_file(thread_id, actual_path) + file_path = virtual_path_for_thread_file(thread_id, actual_path, user_id=user_id) response = await test_client.get( "/api/viewer/filesystem/download", @@ -336,12 +343,13 @@ async def test_viewer_download_returns_full_file_for_large_user_data_content(tes async def test_viewer_delete_removes_user_data_file(test_client, standard_user): headers = standard_user["headers"] + user_id = str(standard_user["user"]["id"]) thread_id = await _create_thread_for_user(test_client, headers) - ensure_thread_dirs(thread_id) - actual_path = sandbox_workspace_dir(thread_id) / "delete_me.txt" + ensure_thread_dirs(thread_id, user_id) + actual_path = sandbox_workspace_dir(thread_id, user_id) / "delete_me.txt" actual_path.write_text("delete me", encoding="utf-8") - file_path = virtual_path_for_thread_file(thread_id, actual_path) + file_path = virtual_path_for_thread_file(thread_id, actual_path, user_id=user_id) delete_response = await test_client.delete( "/api/viewer/filesystem/file", @@ -364,12 +372,13 @@ async def test_viewer_delete_removes_user_data_file(test_client, standard_user): async def test_viewer_delete_removes_empty_user_data_directory(test_client, standard_user): headers = standard_user["headers"] + user_id = str(standard_user["user"]["id"]) thread_id = await _create_thread_for_user(test_client, headers) - ensure_thread_dirs(thread_id) - actual_path = sandbox_workspace_dir(thread_id) / "empty-folder" + ensure_thread_dirs(thread_id, user_id) + actual_path = sandbox_workspace_dir(thread_id, user_id) / "empty-folder" actual_path.mkdir() - dir_path = virtual_path_for_thread_file(thread_id, actual_path) + dir_path = virtual_path_for_thread_file(thread_id, actual_path, user_id=user_id) delete_response = await test_client.delete( "/api/viewer/filesystem/file", @@ -392,15 +401,16 @@ async def test_viewer_delete_removes_empty_user_data_directory(test_client, stan async def test_viewer_delete_recursively_removes_user_data_directory(test_client, standard_user): headers = standard_user["headers"] + user_id = str(standard_user["user"]["id"]) thread_id = await _create_thread_for_user(test_client, headers) - ensure_thread_dirs(thread_id) - actual_path = sandbox_workspace_dir(thread_id) / "nested-folder" + ensure_thread_dirs(thread_id, user_id) + actual_path = sandbox_workspace_dir(thread_id, user_id) / "nested-folder" nested_dir = actual_path / "child" nested_dir.mkdir(parents=True) nested_file = nested_dir / "notes.txt" nested_file.write_text("remove recursively", encoding="utf-8") - dir_path = virtual_path_for_thread_file(thread_id, actual_path) + dir_path = virtual_path_for_thread_file(thread_id, actual_path, user_id=user_id) delete_response = await test_client.delete( "/api/viewer/filesystem/file", @@ -447,9 +457,10 @@ async def test_viewer_delete_rejects_protected_user_data_root_directories( test_client, standard_user, protected_path: str ): headers = standard_user["headers"] + user_id = str(standard_user["user"]["id"]) thread_id = await _create_thread_for_user(test_client, headers) - ensure_thread_dirs(thread_id) + ensure_thread_dirs(thread_id, user_id) response = await test_client.delete( "/api/viewer/filesystem/file", diff --git a/backend/test/unit/backends/test_sandbox_backends.py b/backend/test/unit/backends/test_sandbox_backends.py index a757881f..893ea918 100644 --- a/backend/test/unit/backends/test_sandbox_backends.py +++ b/backend/test/unit/backends/test_sandbox_backends.py @@ -14,13 +14,18 @@ from yuxi.agents.middlewares.skills_middleware import SkillsMiddleware def _runtime( *, thread_id: str | None = "thread-1", + user_id: str | None = "user-1", skills: list[str] | None = None, visible_kbs: list[dict] | None = None, ): - configurable = {"thread_id": thread_id} if thread_id else {} + configurable = {"thread_id": thread_id, "user_id": user_id} if thread_id and user_id else {} return SimpleNamespace( config={"configurable": configurable}, - context=SimpleNamespace(skills=skills or [], _visible_knowledge_bases=visible_kbs or []), + context=SimpleNamespace( + skills=skills or [], + _visible_knowledge_bases=visible_kbs or [], + user_id=user_id, + ), ) @@ -50,12 +55,12 @@ def test_skills_middleware_extracts_slug_for_new_paths() -> None: def test_resolve_virtual_path_rejects_outside_prefix(): with pytest.raises(ValueError, match="path must start with"): - resolve_virtual_path("thread-1", "/etc/passwd") + resolve_virtual_path("thread-1", "/etc/passwd", user_id="user-1") def test_resolve_virtual_path_rejects_path_traversal(): with pytest.raises(ValueError, match="path traversal"): - resolve_virtual_path("thread-1", "/home/gem/user-data/../secrets") + resolve_virtual_path("thread-1", "/home/gem/user-data/../secrets", user_id="user-1") def test_sandbox_id_for_thread_is_stable(): @@ -69,7 +74,7 @@ def test_sandbox_id_for_thread_is_stable(): def test_provisioner_read_reports_binary_files(monkeypatch) -> None: monkeypatch.setattr("yuxi.agents.backends.sandbox.backend.get_sandbox_provider", lambda: object()) - backend = ProvisionerSandboxBackend(thread_id="thread-1") + backend = ProvisionerSandboxBackend(thread_id="thread-1", user_id="user-1") monkeypatch.setattr(backend, "_read_binary", lambda path, offset=0, limit=None: b"\x89PNG\r\n\x1a\n") result = backend.read("/home/gem/user-data/image.png") @@ -79,7 +84,7 @@ def test_provisioner_read_reports_binary_files(monkeypatch) -> None: def test_provisioner_read_reports_invalid_path(monkeypatch) -> None: monkeypatch.setattr("yuxi.agents.backends.sandbox.backend.get_sandbox_provider", lambda: object()) - backend = ProvisionerSandboxBackend(thread_id="thread-1") + backend = ProvisionerSandboxBackend(thread_id="thread-1", user_id="user-1") def _raise_invalid_path(path, offset=0, limit=None): raise ValueError("path traversal is not allowed") @@ -93,7 +98,7 @@ def test_provisioner_read_reports_invalid_path(monkeypatch) -> None: def test_provisioner_download_files_distinguishes_invalid_path_from_read_failure(monkeypatch) -> None: monkeypatch.setattr("yuxi.agents.backends.sandbox.backend.get_sandbox_provider", lambda: object()) - backend = ProvisionerSandboxBackend(thread_id="thread-1") + backend = ProvisionerSandboxBackend(thread_id="thread-1", user_id="user-1") def _fake_read_binary(path, offset=0, limit=None): if path == "/bad-path": @@ -110,7 +115,7 @@ def test_provisioner_download_files_distinguishes_invalid_path_from_read_failure def test_provisioner_execute_returns_error_response_on_client_failure(monkeypatch) -> None: monkeypatch.setattr("yuxi.agents.backends.sandbox.backend.get_sandbox_provider", lambda: object()) - backend = ProvisionerSandboxBackend(thread_id="thread-1") + backend = ProvisionerSandboxBackend(thread_id="thread-1", user_id="user-1") class _FakeClient: class shell: diff --git a/backend/test/unit/services/test_agent_artifacts_state.py b/backend/test/unit/services/test_agent_artifacts_state.py index 49dfcb53..1734a27d 100644 --- a/backend/test/unit/services/test_agent_artifacts_state.py +++ b/backend/test/unit/services/test_agent_artifacts_state.py @@ -9,8 +9,8 @@ from yuxi.agents.toolkits.buildin.tools import _normalize_presented_artifact_pat from yuxi.services.chat_service import extract_agent_state -def _runtime_with_thread(thread_id: str): - context = type("RuntimeContext", (), {"thread_id": thread_id})() +def _runtime_with_thread(thread_id: str, user_id: str = "user-1"): + context = type("RuntimeContext", (), {"thread_id": thread_id, "user_id": user_id})() return type("RuntimeStub", (), {"context": context})() @@ -26,7 +26,7 @@ def test_merge_artifacts_deduplicates_and_preserves_order(): def test_normalize_presented_artifact_path_accepts_host_path(): thread_id = "artifacts-host-path" - ensure_thread_dirs(thread_id) + ensure_thread_dirs(thread_id, "user-1") output_file = sandbox_outputs_dir(thread_id) / "report.md" output_file.write_text("# demo", encoding="utf-8") @@ -37,7 +37,7 @@ def test_normalize_presented_artifact_path_accepts_host_path(): def test_normalize_presented_artifact_path_accepts_virtual_path(): thread_id = "artifacts-virtual-path" - ensure_thread_dirs(thread_id) + ensure_thread_dirs(thread_id, "user-1") output_file = sandbox_outputs_dir(thread_id) / "summary.txt" output_file.write_text("demo", encoding="utf-8") @@ -51,7 +51,7 @@ def test_normalize_presented_artifact_path_accepts_virtual_path(): def test_normalize_presented_artifact_path_rejects_non_outputs_path(): thread_id = "artifacts-reject-path" - ensure_thread_dirs(thread_id) + ensure_thread_dirs(thread_id, "user-1") upload_file = sandbox_uploads_dir(thread_id) / "note.txt" upload_file.write_text("demo", encoding="utf-8") diff --git a/docker/sandbox_provisioner/app.py b/docker/sandbox_provisioner/app.py index 0e919b0f..8789198b 100644 --- a/docker/sandbox_provisioner/app.py +++ b/docker/sandbox_provisioner/app.py @@ -26,6 +26,7 @@ def canonical_backend_name(backend: str) -> str: class CreateSandboxRequest(BaseModel): sandbox_id: str thread_id: str + user_id: str class SandboxResponse(BaseModel): @@ -69,8 +70,9 @@ class MemoryProvisionerBackend: return template.format(sandbox_id=sandbox_id) return template - def create(self, sandbox_id: str, thread_id: str) -> SandboxRecord: + def create(self, sandbox_id: str, thread_id: str, user_id: str) -> SandboxRecord: _ = thread_id # unused in memory backend + _ = user_id # unused in memory backend with self._lock: existing = self._records.get(sandbox_id) if existing is not None: @@ -167,6 +169,17 @@ class LocalContainerProvisionerBackend: raise ValueError("thread_id contains invalid path traversal sequence") return candidate + @staticmethod + def _validate_user_id(user_id: str) -> str: + candidate = str(user_id or "").strip() + if not candidate: + raise ValueError("user_id is required") + if any(ch in candidate for ch in ("/", "\\", "\x00")): + raise ValueError("user_id must be a single safe path segment") + if ".." in candidate: + raise ValueError("user_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()) @@ -184,13 +197,13 @@ class LocalContainerProvisionerBackend: raise ValueError("thread skills path resolved outside threads host root") from exc return thread_skills - def _shared_workspace_host_path(self) -> Path: + def _shared_workspace_host_path(self, user_id: str) -> Path: threads_root = Path(self._threads_host_path).resolve() - workspace = (threads_root / "shared" / "workspace").resolve() + workspace = (threads_root / "shared" / user_id / "workspace").resolve() try: workspace.relative_to(threads_root) except ValueError as exc: - raise ValueError("shared workspace path resolved outside threads host root") from exc + raise ValueError("user workspace path resolved outside threads host root") from exc return workspace def _thread_uploads_host_path(self, thread_id: str) -> Path: @@ -221,9 +234,9 @@ class LocalContainerProvisionerBackend: return source == expected_source return False - def _has_expected_user_data_mounts(self, container, thread_id: str) -> bool: + def _has_expected_user_data_mounts(self, container, thread_id: str, user_id: str) -> bool: expected_mounts = { - "/home/gem/user-data/workspace": str(self._shared_workspace_host_path()), + "/home/gem/user-data/workspace": str(self._shared_workspace_host_path(user_id)), "/home/gem/user-data/uploads": str(self._thread_uploads_host_path(thread_id)), "/home/gem/user-data/outputs": str(self._thread_outputs_host_path(thread_id)), } @@ -306,9 +319,10 @@ class LocalContainerProvisionerBackend: except NotFound: return None - def create(self, sandbox_id: str, thread_id: str) -> SandboxRecord: + def create(self, sandbox_id: str, thread_id: str, user_id: str) -> SandboxRecord: with self._lock: safe_thread_id = self._validate_thread_id(thread_id) + safe_user_id = self._validate_user_id(user_id) existing = self._get_container(sandbox_id) if existing is not None: existing.reload() @@ -316,7 +330,7 @@ class LocalContainerProvisionerBackend: logger.info("Recreating sandbox %s because skills mount is stale", sandbox_id) self.delete(sandbox_id) existing = None - elif not self._has_expected_user_data_mounts(existing, safe_thread_id): + elif not self._has_expected_user_data_mounts(existing, safe_thread_id, safe_user_id): logger.info("Recreating sandbox %s because user-data mounts are stale", sandbox_id) self.delete(sandbox_id) existing = None @@ -339,7 +353,7 @@ class LocalContainerProvisionerBackend: logger.warning("Failed to delete stale sandbox %s before recreate: %s", sandbox_id, exc) threads_root = Path(self._threads_host_path).resolve() - shared_workspace = self._shared_workspace_host_path() + shared_workspace = self._shared_workspace_host_path(safe_user_id) shared_workspace.mkdir(parents=True, exist_ok=True) thread_uploads = self._thread_uploads_host_path(safe_thread_id) thread_outputs = self._thread_outputs_host_path(safe_thread_id) @@ -356,6 +370,7 @@ class LocalContainerProvisionerBackend: "app": "yuxi-sandbox", "sandbox-id": sandbox_id, "thread-id": thread_id, + "user-id": user_id, "managed-by": "yuxi-sandbox-provisioner", }, "volumes": { @@ -391,7 +406,11 @@ class LocalContainerProvisionerBackend: thread_id = str((container.labels or {}).get("thread-id") or "").strip() if not thread_id: return None + user_id = str((container.labels or {}).get("user-id") or "").strip() + if not user_id: + return None safe_thread_id = self._validate_thread_id(thread_id) + safe_user_id = self._validate_user_id(user_id) if not self._is_expected_skills_mount(container, safe_thread_id): logger.info("Discarding stale sandbox %s with legacy skills mount", sandbox_id) try: @@ -399,7 +418,7 @@ class LocalContainerProvisionerBackend: except Exception as exc: logger.warning("Failed to delete stale sandbox %s during discover: %s", sandbox_id, exc) return None - if not self._has_expected_user_data_mounts(container, safe_thread_id): + if not self._has_expected_user_data_mounts(container, safe_thread_id, safe_user_id): logger.info("Discarding stale sandbox %s with legacy user-data mounts", sandbox_id) try: self.delete(sandbox_id) @@ -470,13 +489,13 @@ class KubernetesProvisionerBackend: def _service_name(sandbox_id: str) -> str: return f"sandbox-{sandbox_id}" - def _build_pod_spec(self, sandbox_id: str, thread_id: str): + def _build_pod_spec(self, sandbox_id: str, thread_id: str, user_id: str): pod_name = self._pod_name(sandbox_id) return self._client.V1Pod( metadata=self._client.V1ObjectMeta( name=pod_name, labels={"app": "yuxi-sandbox", "sandbox-id": sandbox_id}, - annotations={"thread-id": thread_id}, + annotations={"thread-id": thread_id, "user-id": user_id}, ), spec=self._client.V1PodSpec( restart_policy="Never", @@ -491,11 +510,11 @@ class KubernetesProvisionerBackend: command=["sh", "-c"], args=[ "chmod 777 /home/gem " - "&& mkdir -p /mnt/shared-data/threads/shared/workspace " + f"&& mkdir -p /mnt/shared-data/threads/shared/{user_id}/workspace " f"/mnt/shared-data/threads/{thread_id}/user-data/uploads " f"/mnt/shared-data/threads/{thread_id}/user-data/outputs " f"/mnt/shared-data/threads/{thread_id}/skills " - f"&& chmod -R 777 /mnt/shared-data/threads/shared/workspace " + f"&& chmod -R 777 /mnt/shared-data/threads/shared/{user_id}/workspace " f"/mnt/shared-data/threads/{thread_id}/user-data ", ], volume_mounts=[ @@ -514,7 +533,7 @@ class KubernetesProvisionerBackend: self._client.V1VolumeMount( name="shared-data", mount_path="/home/gem/user-data/workspace", - sub_path="threads/shared/workspace", + sub_path=f"threads/shared/{user_id}/workspace", ), self._client.V1VolumeMount( name="shared-data", @@ -572,7 +591,7 @@ class KubernetesProvisionerBackend: ), ) - def create(self, sandbox_id: str, thread_id: str) -> SandboxRecord: + def create(self, sandbox_id: str, thread_id: str, user_id: str) -> SandboxRecord: from kubernetes.client.rest import ApiException with self._lock: @@ -586,7 +605,7 @@ class KubernetesProvisionerBackend: try: self._core_api.create_namespaced_pod( namespace=self._namespace, - body=self._build_pod_spec(sandbox_id, thread_id), + body=self._build_pod_spec(sandbox_id, thread_id, user_id), ) except ApiException as exc: if exc.status != 409: @@ -798,7 +817,7 @@ def health(): def create_sandbox(payload: CreateSandboxRequest): try: # Backend.create() already handles container reuse (discovers existing container first) - record = backend_impl.create(payload.sandbox_id, payload.thread_id) + record = backend_impl.create(payload.sandbox_id, payload.thread_id, payload.user_id) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc except Exception as exc: # noqa: BLE001 diff --git a/docs/develop-guides/roadmap.md b/docs/develop-guides/roadmap.md index dcaf5c26..7ea0a5a5 100644 --- a/docs/develop-guides/roadmap.md +++ b/docs/develop-guides/roadmap.md @@ -35,6 +35,7 @@ - 调整 backend Python 工作区依赖边界:将 `backend/package/yuxi` 明确为承载核心运行依赖的业务包,根 `backend/pyproject.toml` 仅保留工作区入口与开发/测试配置,减少依赖职责混淆。 +- 修复沙盒 `workspace` 隔离粒度:宿主机目录从共享 `saves/threads/shared/workspace` 收敛为用户级 `saves/threads/shared//workspace`,并同步传递 `user_id` 到 sandbox 路径解析、provisioner 挂载与 viewer/chat 测试,保证同用户跨线程共享、不同用户隔离。 历史版本发布记录已迁移到 [版本变更记录](./changelog.md)。