From 3fe5d307bf4cb5ef1b37f5f4957dacc09fc74ceb Mon Sep 17 00:00:00 2001 From: Wenjie Zhang Date: Mon, 1 Jun 2026 22:28:01 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E4=BC=98=E5=8C=96=E6=B2=99=E7=AE=B1?= =?UTF-8?q?=E7=BA=BF=E7=A8=8B=E4=BD=9C=E7=94=A8=E5=9F=9F=E7=AE=A1=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../package/yuxi/agents/backends/composite.py | 122 ++++++---- .../yuxi/agents/backends/sandbox/backend.py | 43 +++- .../yuxi/agents/backends/sandbox/provider.py | 156 +++++++++---- .../backends/sandbox/provisioner_client.py | 20 +- .../package/yuxi/storage/postgres/manager.py | 4 +- .../unit/backends/test_sandbox_backends.py | 215 ++++++++++++++++++ .../test_sandbox_provisioner_config.py | 87 +++++++ docker/sandbox_provisioner/app.py | 184 ++++++++++++--- docs/agents/sandbox-architecture.md | 21 +- 9 files changed, 699 insertions(+), 153 deletions(-) diff --git a/backend/package/yuxi/agents/backends/composite.py b/backend/package/yuxi/agents/backends/composite.py index f8b8e089..0c71f38e 100644 --- a/backend/package/yuxi/agents/backends/composite.py +++ b/backend/package/yuxi/agents/backends/composite.py @@ -1,5 +1,7 @@ from __future__ import annotations +from dataclasses import dataclass + from deepagents.backends.composite import ( CompositeBackend, _remap_file_info_path, @@ -23,11 +25,7 @@ def _coerce_glob_result(result) -> GlobResult: class CustomCompositeBackend(CompositeBackend): - """修复 glob 路由逻辑的 CompositeBackend。 - - 修复内容：当 path 不匹配任何路由时应该只搜索 default 后端， - 而不是错误地遍历所有路由后端搜索。 - """ + """修复 glob 路由逻辑的 CompositeBackend。""" def glob(self, pattern: str, path: str = "/") -> GlobResult: backend, backend_path, route_prefix = _route_for_path( @@ -88,62 +86,88 @@ class CustomCompositeBackend(CompositeBackend): return _coerce_glob_result(await self.default.aglob(pattern, path)) -def _get_readable_skills_from_runtime(runtime) -> list[str]: - context = getattr(runtime, "context", None) - selected = getattr(context, "_readable_skills", []) - return normalize_string_list(selected if isinstance(selected, list) else []) +@dataclass(frozen=True) +class _BackendScope: + thread_id: str + uid: str + readable_skills: list[str] + file_thread_id: str + skills_thread_id: str + @classmethod + def from_runtime(cls, runtime) -> _BackendScope: + config = getattr(runtime, "config", None) + configurable = config.get("configurable", {}) if isinstance(config, dict) else {} + context = getattr(runtime, "context", None) + state = getattr(runtime, "state", None) + return cls.from_sources( + configurable if isinstance(configurable, dict) else {}, + context, + state if isinstance(state, dict) else {}, + readable_skills_source=context, + error_context="runtime configurable context", + ) -def _extract_thread_id(runtime) -> str: - config = getattr(runtime, "config", None) - if isinstance(config, dict): - configurable = config.get("configurable", {}) - if isinstance(configurable, dict): - thread_id = configurable.get("thread_id") - if isinstance(thread_id, str) and thread_id.strip(): - return thread_id.strip() + @classmethod + def from_sources(cls, *sources, readable_skills_source, error_context: str) -> _BackendScope: + def string_value(key: str) -> str | None: + for source in sources: + value = source.get(key) if isinstance(source, dict) else getattr(source, key, None) + if isinstance(value, str) and value.strip(): + return value.strip() + return None - context = getattr(runtime, "context", None) - thread_id = getattr(context, "thread_id", None) - if isinstance(thread_id, str) and thread_id.strip(): - return thread_id.strip() + thread_id = string_value("thread_id") + if not thread_id: + raise ValueError(f"thread_id is required in {error_context}") - raise ValueError("thread_id is required in runtime configurable context") + uid = string_value("uid") + if not uid: + raise ValueError(f"uid is required in {error_context}") + selected = getattr(readable_skills_source, "_readable_skills", []) + return cls( + thread_id=thread_id, + uid=uid, + readable_skills=normalize_string_list(selected if isinstance(selected, list) else []), + file_thread_id=string_value("file_thread_id") or thread_id, + skills_thread_id=string_value("skills_thread_id") or thread_id, + ) -def _extract_uid(runtime) -> str: - config = getattr(runtime, "config", None) - if isinstance(config, dict): - configurable = config.get("configurable", {}) - if isinstance(configurable, dict): - uid = configurable.get("uid") - if isinstance(uid, str) and uid.strip(): - return uid.strip() - - context = getattr(runtime, "context", None) - uid = getattr(context, "uid", None) - if isinstance(uid, str) and uid.strip(): - return uid.strip() - - raise ValueError("uid is required in runtime configurable context") + def create_backend(self) -> CompositeBackend: + return CustomCompositeBackend( + default=ProvisionerSandboxBackend( + thread_id=self.thread_id, + uid=self.uid, + readable_skills=self.readable_skills, + file_thread_id=self.file_thread_id, + skills_thread_id=self.skills_thread_id, + ), + routes={ + "/skills/": SelectedSkillsReadonlyBackend(selected_slugs=self.readable_skills), + }, + artifacts_root=VIRTUAL_PATH_OUTPUTS, + ) def create_agent_composite_backend(runtime) -> CompositeBackend: - readable_skills = _get_readable_skills_from_runtime(runtime) - thread_id = _extract_thread_id(runtime) - uid = _extract_uid(runtime) - return CustomCompositeBackend( - default=ProvisionerSandboxBackend(thread_id=thread_id, uid=uid, readable_skills=readable_skills), - routes={ - "/skills/": SelectedSkillsReadonlyBackend(selected_slugs=readable_skills), - }, - artifacts_root=VIRTUAL_PATH_OUTPUTS, - ) + return _BackendScope.from_runtime(runtime).create_backend() -def create_agent_filesystem_middleware(tool_token_limit_before_evict: int | None = None) -> FilesystemMiddleware: +def create_agent_filesystem_middleware( + tool_token_limit_before_evict: int | None = None, + *, + context=None, +) -> FilesystemMiddleware: + backend = create_agent_composite_backend + if context is not None: + backend = _BackendScope.from_sources( + context, + readable_skills_source=context, + error_context="runtime context", + ).create_backend() middleware = FilesystemMiddleware( - backend=create_agent_composite_backend, + backend=backend, tool_token_limit_before_evict=tool_token_limit_before_evict, ) middleware._large_tool_results_prefix = VIRTUAL_PATH_LARGE_TOOL_RESULTS diff --git a/backend/package/yuxi/agents/backends/sandbox/backend.py b/backend/package/yuxi/agents/backends/sandbox/backend.py index 14ec5ddb..1eaf3010 100644 --- a/backend/package/yuxi/agents/backends/sandbox/backend.py +++ b/backend/package/yuxi/agents/backends/sandbox/backend.py @@ -150,17 +150,31 @@ def _looks_like_binary(content: bytes) -> bool: class ProvisionerSandboxBackend(BaseSandbox): - def __init__(self, thread_id: str, *, uid: str, readable_skills: list[str] | None = None): + def __init__( + self, + thread_id: str, + *, + uid: str, + readable_skills: list[str] | None = None, + file_thread_id: str | None = None, + skills_thread_id: 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._file_thread_id = str(file_thread_id or self._thread_id).strip() + if not self._file_thread_id: + raise ValueError("file_thread_id is required for ProvisionerSandboxBackend") + self._skills_thread_id = str(skills_thread_id or self._thread_id).strip() + if not self._skills_thread_id: + raise ValueError("skills_thread_id is required for ProvisionerSandboxBackend") self._uid = str(uid or "").strip() if not self._uid: raise ValueError("uid is required for ProvisionerSandboxBackend") self._readable_skills = list(readable_skills or []) self._provider = get_sandbox_provider() - self._id = sandbox_id_for_thread(self._thread_id) + self._id = sandbox_id_for_thread(self._file_thread_id, self._skills_thread_id) self._client: Any | None = None self._client_url: str | None = None self._command_timeout_seconds = int(getattr(conf, "sandbox_exec_timeout_seconds", 180)) @@ -181,8 +195,14 @@ class ProvisionerSandboxBackend(BaseSandbox): return AgentSandboxClient(base_url=sandbox_url, timeout=self._command_timeout_seconds) def _get_client(self) -> Any: - sync_thread_readable_skills(self._thread_id, self._readable_skills) - connection = self._provider.get(self._thread_id, uid=self._uid, create_if_missing=True) + sync_thread_readable_skills(self._skills_thread_id, self._readable_skills) + connection = self._provider.get( + self._thread_id, + uid=self._uid, + create_if_missing=True, + file_thread_id=self._file_thread_id, + skills_thread_id=self._skills_thread_id, + ) if connection is None: raise RuntimeError(f"sandbox is unavailable for thread {self._thread_id}") @@ -195,9 +215,9 @@ class ProvisionerSandboxBackend(BaseSandbox): def _read_binary(self, path: str, offset: int = 0, limit: int | None = None) -> bytes: """Read file content from the sandbox file API and normalize it to bytes. - The underlying API may return base64 text, raw bytes, or plain strings. - This helper is the single normalization point used by read(), edit(), and - download_files() so all read paths share the same transport semantics. + The underlying API returns plain text by default and may include an + explicit `encoding="base64"` marker for binary payloads. This helper is + the single normalization point used by read(), edit(), and download_files(). """ start_line = max(0, int(offset)) end_line = start_line + int(limit) if limit is not None else None @@ -216,10 +236,10 @@ class ProvisionerSandboxBackend(BaseSandbox): if not isinstance(content, str): return str(content).encode("utf-8") - try: + encoding = getattr(result.data, "encoding", None) + if isinstance(encoding, str) and encoding.lower() == "base64": return base64.b64decode(content, validate=True) - except Exception: # noqa: BLE001 - return content.encode("utf-8") + return content.encode("utf-8") def read( self, @@ -487,8 +507,7 @@ class ProvisionerSandboxBackend(BaseSandbox): def download_files(self, paths: list[str]) -> list[FileDownloadResponse]: """Download file payloads as raw bytes from the sandbox file API. - The underlying API is read with base64 encoding and decoded back into - bytes by _read_binary(). + _read_binary() normalizes the sandbox file API response to bytes. """ responses: list[FileDownloadResponse] = [] for path in paths: diff --git a/backend/package/yuxi/agents/backends/sandbox/provider.py b/backend/package/yuxi/agents/backends/sandbox/provider.py index 0b292747..979858a7 100644 --- a/backend/package/yuxi/agents/backends/sandbox/provider.py +++ b/backend/package/yuxi/agents/backends/sandbox/provider.py @@ -13,11 +13,18 @@ from yuxi.utils.logging_config import logger from .provisioner_client import ProvisionerClient, SandboxRecord -def sandbox_id_for_thread(thread_id: str) -> str: - digest = hashlib.sha256(thread_id.encode("utf-8")).hexdigest() +def sandbox_id_for_thread(thread_id: str, skills_thread_id: str | None = None) -> str: + file_thread_id = str(thread_id or "").strip() + skills_id = str(skills_thread_id or file_thread_id).strip() + identity = file_thread_id if skills_id == file_thread_id else f"{file_thread_id}:{skills_id}" + digest = hashlib.sha256(identity.encode("utf-8")).hexdigest() return digest[:12] +def _sandbox_key(file_thread_id: str, skills_thread_id: str) -> str: + return f"{file_thread_id}::{skills_thread_id}" + + def normalize_env(env: dict | None) -> dict[str, str]: if not isinstance(env, dict): return {} @@ -58,7 +65,10 @@ def load_user_agent_env(uid: str) -> dict[str, str]: @dataclass(slots=True) class SandboxConnection: + cache_key: str thread_id: str + file_thread_id: str + skills_thread_id: str uid: str sandbox_id: str sandbox_url: str @@ -81,91 +91,149 @@ class ProvisionerSandboxProvider: 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: + def _thread_lock(self, cache_key: str) -> threading.Lock: with self._lock: - lock = self._thread_locks.get(thread_id) + lock = self._thread_locks.get(cache_key) if lock is None: lock = threading.Lock() - self._thread_locks[thread_id] = lock + self._thread_locks[cache_key] = lock return lock - def _record_to_connection(self, thread_id: str, uid: str, record: SandboxRecord) -> SandboxConnection: + def _record_to_connection( + self, + *, + cache_key: str, + thread_id: str, + file_thread_id: str, + skills_thread_id: str, + uid: str, + record: SandboxRecord, + ) -> SandboxConnection: connection = SandboxConnection( + cache_key=cache_key, thread_id=thread_id, + file_thread_id=file_thread_id, + skills_thread_id=skills_thread_id, uid=uid, sandbox_id=record.sandbox_id, sandbox_url=record.sandbox_url, ) - self._connections[thread_id] = connection - self._last_touch_at[thread_id] = time.time() + self._connections[cache_key] = connection + self._last_touch_at[cache_key] = time.time() return connection - def _should_touch(self, thread_id: str) -> bool: + def _should_touch(self, cache_key: str) -> bool: if self._touch_interval_seconds <= 0: return False - last_touch = self._last_touch_at.get(thread_id) + last_touch = self._last_touch_at.get(cache_key) 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): + if not self._should_touch(connection.cache_key): return True is_alive = self._client.touch(connection.sandbox_id) - self._last_touch_at[connection.thread_id] = time.time() + self._last_touch_at[connection.cache_key] = time.time() return is_alive - def acquire(self, thread_id: str, *, uid: str) -> str: - lock = self._thread_lock(thread_id) + def acquire( + self, + thread_id: str, + *, + uid: str, + file_thread_id: str | None = None, + skills_thread_id: str | None = None, + ) -> str: + file_id = str(file_thread_id or thread_id).strip() + skills_id = str(skills_thread_id or thread_id).strip() + cache_key = _sandbox_key(file_id, skills_id) + lock = self._thread_lock(cache_key) with lock: - current = self._connections.get(thread_id) + current = self._connections.get(cache_key) if current: + if current.uid != uid: + raise RuntimeError(f"sandbox scope {cache_key} belongs to uid {current.uid}, not {uid}") 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) + self._connections.pop(cache_key, None) + self._last_touch_at.pop(cache_key, None) except Exception as exc: # noqa: BLE001 - logger.warning(f"Failed to touch sandbox {current.sandbox_id} for thread {thread_id}: {exc}") + logger.warning(f"Failed to touch sandbox {current.sandbox_id} for {cache_key}: {exc}") return current.sandbox_id - sandbox_id = sandbox_id_for_thread(thread_id) - 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, uid, load_user_agent_env(uid)) - else: - logger.info(f"Reusing sandbox {sandbox_id} for thread {thread_id}") + sandbox_id = sandbox_id_for_thread(file_id, skills_id) + logger.info(f"Ensuring sandbox {sandbox_id} for file thread {file_id} and skills thread {skills_id}") + record = self._client.create( + sandbox_id, + thread_id, + uid, + load_user_agent_env(uid), + file_thread_id=file_id, + skills_thread_id=skills_id, + ) - connection = self._record_to_connection(thread_id, uid, record) + connection = self._record_to_connection( + cache_key=cache_key, + thread_id=thread_id, + file_thread_id=file_id, + skills_thread_id=skills_id, + uid=uid, + record=record, + ) return connection.sandbox_id - def get(self, thread_id: str, *, uid: str, create_if_missing: bool = False) -> SandboxConnection | None: - lock = self._thread_lock(thread_id) + def get( + self, + thread_id: str, + *, + uid: str, + create_if_missing: bool = False, + file_thread_id: str | None = None, + skills_thread_id: str | None = None, + ) -> SandboxConnection | None: + file_id = str(file_thread_id or thread_id).strip() + skills_id = str(skills_thread_id or thread_id).strip() + cache_key = _sandbox_key(file_id, skills_id) + lock = self._thread_lock(cache_key) with lock: - current = self._connections.get(thread_id) + current = self._connections.get(cache_key) if current: + if current.uid != uid: + raise RuntimeError(f"sandbox scope {cache_key} belongs to uid {current.uid}, not {uid}") try: if self._touch_if_needed(current): return current - self._connections.pop(thread_id, None) - self._last_touch_at.pop(thread_id, None) + self._connections.pop(cache_key, None) + self._last_touch_at.pop(cache_key, None) except Exception as exc: # noqa: BLE001 - logger.warning(f"Failed to touch sandbox {current.sandbox_id} for thread {thread_id}: {exc}") + logger.warning(f"Failed to touch sandbox {current.sandbox_id} for {cache_key}: {exc}") return current - if current.uid == uid: - 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: + sandbox_id = sandbox_id_for_thread(file_id, skills_id) + if create_if_missing: + record = self._client.create( + sandbox_id, + thread_id, + uid, + load_user_agent_env(uid), + file_thread_id=file_id, + skills_thread_id=skills_id, + ) + else: + record = self._client.discover(sandbox_id) + if record is None: return None - record = self._client.create(sandbox_id, thread_id, uid, load_user_agent_env(uid)) - return self._record_to_connection(thread_id, uid, record) + return self._record_to_connection( + cache_key=cache_key, + thread_id=thread_id, + file_thread_id=file_id, + skills_thread_id=skills_id, + uid=uid, + record=record, + ) def shutdown(self) -> None: with self._lock: @@ -177,9 +245,7 @@ class ProvisionerSandboxProvider: try: self._client.delete(connection.sandbox_id) except Exception as exc: # noqa: BLE001 - logger.warning( - f"Failed to release sandbox {connection.sandbox_id} for thread {connection.thread_id}: {exc}" - ) + logger.warning(f"Failed to release sandbox {connection.sandbox_id} for {connection.cache_key}: {exc}") _sandbox_provider: ProvisionerSandboxProvider | None = None diff --git a/backend/package/yuxi/agents/backends/sandbox/provisioner_client.py b/backend/package/yuxi/agents/backends/sandbox/provisioner_client.py index c6bcf1a3..18bc881c 100644 --- a/backend/package/yuxi/agents/backends/sandbox/provisioner_client.py +++ b/backend/package/yuxi/agents/backends/sandbox/provisioner_client.py @@ -29,11 +29,27 @@ class ProvisionerClient: response = self._request("GET", "/health") return response.status_code == 200 - def create(self, sandbox_id: str, thread_id: str, uid: str, env: dict[str, str] | None = None) -> SandboxRecord: + def create( + self, + sandbox_id: str, + thread_id: str, + uid: str, + env: dict[str, str] | None = None, + *, + file_thread_id: str | None = None, + skills_thread_id: str | None = None, + ) -> SandboxRecord: response = self._request( "POST", "/api/sandboxes", - json={"sandbox_id": sandbox_id, "thread_id": thread_id, "uid": uid, "env": env or {}}, + json={ + "sandbox_id": sandbox_id, + "thread_id": thread_id, + "file_thread_id": file_thread_id or thread_id, + "skills_thread_id": skills_thread_id or thread_id, + "uid": uid, + "env": env or {}, + }, ) 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/storage/postgres/manager.py b/backend/package/yuxi/storage/postgres/manager.py index f1727e1d..eea82c98 100644 --- a/backend/package/yuxi/storage/postgres/manager.py +++ b/backend/package/yuxi/storage/postgres/manager.py @@ -378,7 +378,6 @@ class PostgresManager(metaclass=SingletonMeta): ), "ALTER TABLE IF EXISTS skills ADD COLUMN IF NOT EXISTS enabled BOOLEAN NOT NULL DEFAULT TRUE", "ALTER TABLE IF EXISTS skills ADD COLUMN IF NOT EXISTS content_hash VARCHAR(128)", - "ALTER TABLE IF EXISTS subagents ADD COLUMN IF NOT EXISTS enabled BOOLEAN NOT NULL DEFAULT TRUE", "ALTER TABLE IF EXISTS conversations ADD COLUMN IF NOT EXISTS is_pinned BOOLEAN NOT NULL DEFAULT FALSE", "ALTER TABLE IF EXISTS mcp_servers ADD COLUMN IF NOT EXISTS env JSONB", """ @@ -403,6 +402,7 @@ class PostgresManager(metaclass=SingletonMeta): config_json JSONB NOT NULL DEFAULT '{}'::jsonb, share_config JSONB NOT NULL DEFAULT '{}'::jsonb, is_default BOOLEAN NOT NULL DEFAULT FALSE, + is_subagent BOOLEAN NOT NULL DEFAULT FALSE, created_by VARCHAR(64), updated_by VARCHAR(64), created_at TIMESTAMPTZ DEFAULT NOW(), @@ -411,8 +411,10 @@ class PostgresManager(metaclass=SingletonMeta): """, "ALTER TABLE IF EXISTS agents ADD COLUMN IF NOT EXISTS backend_id VARCHAR(64)", "ALTER TABLE IF EXISTS agents ADD COLUMN IF NOT EXISTS share_config JSONB NOT NULL DEFAULT '{}'::jsonb", + "ALTER TABLE IF EXISTS agents ADD COLUMN IF NOT EXISTS is_subagent BOOLEAN NOT NULL DEFAULT FALSE", "CREATE UNIQUE INDEX IF NOT EXISTS ix_agents_slug ON agents(slug)", "CREATE INDEX IF NOT EXISTS ix_agents_backend_id ON agents(backend_id)", + "CREATE INDEX IF NOT EXISTS ix_agents_is_subagent ON agents(is_subagent)", "CREATE INDEX IF NOT EXISTS ix_agents_created_by ON agents(created_by)", """ CREATE UNIQUE INDEX IF NOT EXISTS uq_agents_default diff --git a/backend/test/unit/backends/test_sandbox_backends.py b/backend/test/unit/backends/test_sandbox_backends.py index 0352fa77..eb56dbb5 100644 --- a/backend/test/unit/backends/test_sandbox_backends.py +++ b/backend/test/unit/backends/test_sandbox_backends.py @@ -2,6 +2,7 @@ from __future__ import annotations +import threading from types import MethodType, SimpleNamespace import pytest @@ -64,6 +65,52 @@ def test_create_agent_composite_backend_ignores_unprepared_context_skills(monkey assert backend.default._readable_skills == [] +def test_create_agent_composite_backend_uses_split_thread_scopes(monkeypatch): + monkeypatch.setattr("yuxi.agents.backends.sandbox.backend.get_sandbox_provider", lambda: object()) + runtime = _runtime(thread_id="child-thread", uid="user-1", readable_skills=["worker-skill"]) + runtime.config["configurable"].update( + {"file_thread_id": "parent-thread", "skills_thread_id": "child-skills-thread"} + ) + + backend = create_agent_composite_backend(runtime) + + assert backend.default._thread_id == "child-thread" + assert backend.default._file_thread_id == "parent-thread" + assert backend.default._skills_thread_id == "child-skills-thread" + assert backend.default._readable_skills == ["worker-skill"] + + +def test_create_agent_composite_backend_uses_split_thread_scopes_from_state(monkeypatch): + monkeypatch.setattr("yuxi.agents.backends.sandbox.backend.get_sandbox_provider", lambda: object()) + runtime = _runtime(thread_id="child-thread", uid="user-1", readable_skills=["worker-skill"]) + runtime.state = {"file_thread_id": "parent-thread", "skills_thread_id": "child-skills-thread"} + + backend = create_agent_composite_backend(runtime) + + assert backend.default._thread_id == "child-thread" + assert backend.default._file_thread_id == "parent-thread" + assert backend.default._skills_thread_id == "child-skills-thread" + + +def test_create_agent_filesystem_middleware_uses_context_scope(monkeypatch): + monkeypatch.setattr("yuxi.agents.backends.sandbox.backend.get_sandbox_provider", lambda: object()) + context = SimpleNamespace( + thread_id="child-thread", + uid="user-1", + file_thread_id="parent-thread", + skills_thread_id="child-skills-thread", + _readable_skills=["worker-skill"], + ) + + middleware = create_agent_filesystem_middleware(context=context) + backend = middleware.backend + + assert backend.default._thread_id == "child-thread" + assert backend.default._file_thread_id == "parent-thread" + assert backend.default._skills_thread_id == "child-skills-thread" + assert backend.default._readable_skills == ["worker-skill"] + + def test_create_agent_filesystem_middleware_uses_outputs_for_internal_artifacts() -> None: middleware = create_agent_filesystem_middleware(tool_token_limit_before_evict=500) @@ -118,6 +165,141 @@ def test_sandbox_id_for_thread_is_stable(): assert len(sid1) == 12 +def test_sandbox_id_for_thread_includes_skills_scope(): + parent_only = sandbox_id_for_thread("parent-thread") + split_scope = sandbox_id_for_thread("parent-thread", "child-skills-thread") + + assert split_scope == sandbox_id_for_thread("parent-thread", "child-skills-thread") + assert split_scope != parent_only + assert sandbox_id_for_thread("parent-thread", "parent-thread") == parent_only + + +def test_provider_rejects_same_sandbox_scope_for_different_uid(monkeypatch) -> None: + from yuxi.agents.backends.sandbox.provider import ProvisionerSandboxProvider + + class FakeClient: + def touch(self, _sandbox_id): + return True + + provider = ProvisionerSandboxProvider.__new__(ProvisionerSandboxProvider) + provider._client = FakeClient() + provider._lock = threading.Lock() + provider._thread_locks = {} + provider._connections = {} + provider._last_touch_at = {} + provider._touch_interval_seconds = 30 + provider._record_to_connection( + cache_key="parent-thread::child-skills-thread", + thread_id="child-thread", + file_thread_id="parent-thread", + skills_thread_id="child-skills-thread", + uid="user-1", + record=SimpleNamespace(sandbox_id="sandbox-1", sandbox_url="http://sandbox"), + ) + + with pytest.raises(RuntimeError, match="belongs to uid user-1"): + provider.get( + "child-thread", + uid="user-2", + file_thread_id="parent-thread", + skills_thread_id="child-skills-thread", + ) + with pytest.raises(RuntimeError, match="belongs to uid user-1"): + provider.acquire( + "child-thread", + uid="user-2", + file_thread_id="parent-thread", + skills_thread_id="child-skills-thread", + ) + + +def test_provider_get_create_if_missing_ensures_expected_split_scope(monkeypatch) -> None: + from yuxi.agents.backends.sandbox.provider import ProvisionerSandboxProvider + + calls = [] + + class FakeClient: + def create(self, sandbox_id, thread_id, uid, env, *, file_thread_id=None, skills_thread_id=None): + calls.append((sandbox_id, thread_id, uid, env, file_thread_id, skills_thread_id)) + return SimpleNamespace(sandbox_id=sandbox_id, sandbox_url="http://sandbox") + + def discover(self, _sandbox_id): + raise AssertionError("create_if_missing should ensure sandbox through provisioner create") + + provider = ProvisionerSandboxProvider.__new__(ProvisionerSandboxProvider) + provider._client = FakeClient() + provider._lock = threading.Lock() + provider._thread_locks = {} + provider._connections = {} + provider._last_touch_at = {} + provider._touch_interval_seconds = 30 + monkeypatch.setattr("yuxi.agents.backends.sandbox.provider.load_user_agent_env", lambda uid: {"A": uid}) + + connection = provider.get( + "child-thread", + uid="user-1", + create_if_missing=True, + file_thread_id="parent-thread", + skills_thread_id="child-skills-thread", + ) + + sandbox_id = sandbox_id_for_thread("parent-thread", "child-skills-thread") + assert connection.sandbox_id == sandbox_id + assert connection.file_thread_id == "parent-thread" + assert connection.skills_thread_id == "child-skills-thread" + assert calls == [ + ( + sandbox_id, + "child-thread", + "user-1", + {"A": "user-1"}, + "parent-thread", + "child-skills-thread", + ) + ] + + +def test_provisioner_uses_file_and_skills_thread_ids(monkeypatch) -> None: + provider_calls = [] + synced = [] + + class FakeProvider: + def get(self, thread_id, **kwargs): + provider_calls.append((thread_id, kwargs)) + return SimpleNamespace(sandbox_url="http://sandbox") + + monkeypatch.setattr("yuxi.agents.backends.sandbox.backend.get_sandbox_provider", lambda: FakeProvider()) + monkeypatch.setattr( + "yuxi.agents.backends.sandbox.backend.sync_thread_readable_skills", + lambda thread_id, skills: synced.append((thread_id, skills)), + ) + + backend = ProvisionerSandboxBackend( + thread_id="child-thread", + uid="user-1", + readable_skills=["worker-skill"], + file_thread_id="parent-thread", + skills_thread_id="child-skills-thread", + ) + backend._build_client = MethodType(lambda self, sandbox_url: SimpleNamespace(url=sandbox_url), backend) + + client = backend._get_client() + + assert client.url == "http://sandbox" + assert synced == [("child-skills-thread", ["worker-skill"])] + assert provider_calls == [ + ( + "child-thread", + { + "uid": "user-1", + "create_if_missing": True, + "file_thread_id": "parent-thread", + "skills_thread_id": "child-skills-thread", + }, + ) + ] + + def test_provisioner_denies_reads_outside_allowed_roots(monkeypatch) -> None: monkeypatch.setattr("yuxi.agents.backends.sandbox.backend.get_sandbox_provider", lambda: object()) backend = ProvisionerSandboxBackend(thread_id="thread-1", uid="user-1") @@ -185,6 +367,39 @@ def test_provisioner_glob_root_searches_readable_roots(monkeypatch) -> None: ] +def test_provisioner_read_preserves_base64_like_plain_text(monkeypatch) -> None: + monkeypatch.setattr("yuxi.agents.backends.sandbox.backend.get_sandbox_provider", lambda: object()) + backend = ProvisionerSandboxBackend(thread_id="thread-1", uid="user-1") + + fake_client = SimpleNamespace( + file=SimpleNamespace( + read_file=lambda **_kwargs: SimpleNamespace(data=SimpleNamespace(content="SGVsbG8=")) + ) + ) + backend._get_client = MethodType(lambda self: fake_client, backend) + + result = backend.read("/home/gem/user-data/outputs/base64-looking.txt") + + assert result.error is None + assert result.file_data == {"content": "SGVsbG8=", "encoding": "utf-8"} + + +def test_provisioner_read_decodes_explicit_base64(monkeypatch) -> None: + monkeypatch.setattr("yuxi.agents.backends.sandbox.backend.get_sandbox_provider", lambda: object()) + backend = ProvisionerSandboxBackend(thread_id="thread-1", uid="user-1") + + fake_client = SimpleNamespace( + file=SimpleNamespace( + read_file=lambda **_kwargs: SimpleNamespace( + data=SimpleNamespace(content="SGVsbG8=", encoding="base64") + ) + ) + ) + backend._get_client = MethodType(lambda self: fake_client, backend) + + assert backend._read_binary("/home/gem/user-data/outputs/file.bin") == b"Hello" + + 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", uid="user-1") diff --git a/backend/test/unit/backends/test_sandbox_provisioner_config.py b/backend/test/unit/backends/test_sandbox_provisioner_config.py index 6b03cdf0..3d22a329 100644 --- a/backend/test/unit/backends/test_sandbox_provisioner_config.py +++ b/backend/test/unit/backends/test_sandbox_provisioner_config.py @@ -3,6 +3,7 @@ from __future__ import annotations import importlib.util import sys from pathlib import Path +from types import SimpleNamespace MODULE_NAME = "sandbox_provisioner_app_for_test" @@ -61,3 +62,89 @@ def test_normalize_env_converts_values_to_strings(monkeypatch): module = _load_module() assert module.normalize_env({"A": 1, "B": None, "": "ignored"}) == {"A": "1", "B": ""} + + +def test_memory_backend_accepts_split_thread_ids(monkeypatch): + monkeypatch.setenv("PROVISIONER_BACKEND", "memory") + module = _load_module() + backend = module.MemoryProvisionerBackend() + + record = backend.create( + "sandbox-1", + "child-thread", + "user-1", + file_thread_id="parent-thread", + skills_thread_id="child-skills-thread", + ) + + assert record.sandbox_id == "sandbox-1" + assert backend.discover("sandbox-1") is record + + +def test_docker_mount_checks_use_file_and_skills_thread_ids(monkeypatch, tmp_path): + monkeypatch.setenv("PROVISIONER_BACKEND", "memory") + module = _load_module() + backend = object.__new__(module.LocalContainerProvisionerBackend) + backend._threads_host_path = str(tmp_path) + + workspace = tmp_path / "shared" / "user-1" / "workspace" + uploads = tmp_path / "parent-thread" / "user-data" / "uploads" + outputs = tmp_path / "parent-thread" / "user-data" / "outputs" + skills = tmp_path / "child-skills-thread" / "skills" + container = SimpleNamespace( + attrs={ + "Mounts": [ + {"Destination": "/home/gem/user-data/workspace", "Source": str(workspace)}, + {"Destination": "/home/gem/user-data/uploads", "Source": str(uploads)}, + {"Destination": "/home/gem/user-data/outputs", "Source": str(outputs)}, + {"Destination": "/home/gem/skills", "Source": str(skills)}, + ] + } + ) + + assert backend._has_expected_user_data_mounts(container, "parent-thread", "user-1") is True + assert backend._is_expected_skills_mount(container, "child-skills-thread") is True + assert backend._has_expected_user_data_mounts(container, "child-thread", "user-1") is False + assert backend._is_expected_skills_mount(container, "parent-thread") is False + + +def test_kubernetes_mount_check_uses_file_and_skills_thread_ids(monkeypatch): + monkeypatch.setenv("PROVISIONER_BACKEND", "memory") + module = _load_module() + pod = SimpleNamespace( + spec=SimpleNamespace( + containers=[ + SimpleNamespace( + name="sandbox", + volume_mounts=[ + SimpleNamespace( + mount_path="/home/gem/user-data/workspace", + sub_path="threads/shared/user-1/workspace", + ), + SimpleNamespace( + mount_path="/home/gem/user-data/uploads", + sub_path="threads/parent-thread/user-data/uploads", + ), + SimpleNamespace( + mount_path="/home/gem/user-data/outputs", + sub_path="threads/parent-thread/user-data/outputs", + ), + SimpleNamespace(mount_path="/home/gem/skills", sub_path="threads/child-skills-thread/skills"), + ], + ) + ] + ) + ) + + assert module.KubernetesProvisionerBackend._pod_has_expected_mounts( + pod, + file_thread_id="parent-thread", + skills_thread_id="child-skills-thread", + uid="user-1", + ) + assert not module.KubernetesProvisionerBackend._pod_has_expected_mounts( + pod, + file_thread_id="child-thread", + skills_thread_id="child-skills-thread", + uid="user-1", + ) diff --git a/docker/sandbox_provisioner/app.py b/docker/sandbox_provisioner/app.py index 282d16c9..e0656642 100644 --- a/docker/sandbox_provisioner/app.py +++ b/docker/sandbox_provisioner/app.py @@ -41,6 +41,8 @@ def merged_sandbox_env(global_env: dict[str, str], user_env: dict[str, str]) -> class CreateSandboxRequest(BaseModel): sandbox_id: str thread_id: str + file_thread_id: str | None = None + skills_thread_id: str | None = None uid: str env: dict[str, str] = Field(default_factory=dict) @@ -86,10 +88,21 @@ class MemoryProvisionerBackend: return template.format(sandbox_id=sandbox_id) return template - def create(self, sandbox_id: str, thread_id: str, uid: str, env: dict[str, str] | None = None) -> SandboxRecord: - _ = thread_id # unused in memory backend - _ = uid # unused in memory backend - _ = env # unused in memory backend + def create( + self, + sandbox_id: str, + thread_id: str, + uid: str, + env: dict[str, str] | None = None, + *, + file_thread_id: str | None = None, + skills_thread_id: str | None = None, + ) -> SandboxRecord: + _ = thread_id + _ = file_thread_id + _ = skills_thread_id + _ = uid + _ = env with self._lock: existing = self._records.get(sandbox_id) if existing is not None: @@ -242,8 +255,8 @@ class LocalContainerProvisionerBackend: raise ValueError("thread outputs path resolved outside threads host root") from exc return outputs - def _is_expected_skills_mount(self, container, thread_id: str) -> bool: - expected_source = str(self._thread_skills_host_path(thread_id)) + def _is_expected_skills_mount(self, container, skills_thread_id: str) -> bool: + expected_source = str(self._thread_skills_host_path(skills_thread_id)) for mount in container.attrs.get("Mounts") or []: destination = (mount.get("Destination") or "").rstrip("/") if destination != "/home/gem/skills": @@ -252,11 +265,11 @@ class LocalContainerProvisionerBackend: return source == expected_source return False - def _has_expected_user_data_mounts(self, container, thread_id: str, uid: str) -> bool: + def _has_expected_user_data_mounts(self, container, file_thread_id: str, uid: str) -> bool: expected_mounts = { "/home/gem/user-data/workspace": str(self._shared_workspace_host_path(uid)), - "/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)), + "/home/gem/user-data/uploads": str(self._thread_uploads_host_path(file_thread_id)), + "/home/gem/user-data/outputs": str(self._thread_outputs_host_path(file_thread_id)), } actual_mounts = { str((mount.get("Destination") or "").rstrip("/")): str((mount.get("Source") or "").rstrip("/")) @@ -337,18 +350,29 @@ class LocalContainerProvisionerBackend: except NotFound: return None - def create(self, sandbox_id: str, thread_id: str, uid: str, env: dict[str, str] | None = None) -> SandboxRecord: + def create( + self, + sandbox_id: str, + thread_id: str, + uid: str, + env: dict[str, str] | None = None, + *, + file_thread_id: str | None = None, + skills_thread_id: str | None = None, + ) -> SandboxRecord: with self._lock: safe_thread_id = self._validate_thread_id(thread_id) + safe_file_thread_id = self._validate_thread_id(file_thread_id or safe_thread_id) + safe_skills_thread_id = self._validate_thread_id(skills_thread_id or safe_thread_id) safe_uid = self._validate_uid(uid) existing = self._get_container(sandbox_id) if existing is not None: existing.reload() - if not self._is_expected_skills_mount(existing, safe_thread_id): + if not self._is_expected_skills_mount(existing, safe_skills_thread_id): 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, safe_uid): + elif not self._has_expected_user_data_mounts(existing, safe_file_thread_id, safe_uid): logger.info("Recreating sandbox %s because user-data mounts are stale", sandbox_id) self.delete(sandbox_id) existing = None @@ -373,11 +397,11 @@ class LocalContainerProvisionerBackend: threads_root = Path(self._threads_host_path).resolve() shared_workspace = self._shared_workspace_host_path(safe_uid) 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) + thread_uploads = self._thread_uploads_host_path(safe_file_thread_id) + thread_outputs = self._thread_outputs_host_path(safe_file_thread_id) thread_uploads.mkdir(parents=True, exist_ok=True) thread_outputs.mkdir(parents=True, exist_ok=True) - thread_skills = self._thread_skills_host_path(safe_thread_id) + thread_skills = self._thread_skills_host_path(safe_skills_thread_id) thread_skills.mkdir(parents=True, exist_ok=True) container_name = self._container_name(sandbox_id) @@ -387,8 +411,10 @@ class LocalContainerProvisionerBackend: "labels": { "app": "yuxi-sandbox", "sandbox-id": sandbox_id, - "thread-id": thread_id, - "uid": uid, + "thread-id": safe_thread_id, + "file-thread-id": safe_file_thread_id, + "skills-thread-id": safe_skills_thread_id, + "uid": safe_uid, "managed-by": "yuxi-sandbox-provisioner", }, "volumes": { @@ -424,22 +450,26 @@ class LocalContainerProvisionerBackend: if container is None: return None container.reload() - thread_id = str((container.labels or {}).get("thread-id") or "").strip() + labels = container.labels or {} + thread_id = str(labels.get("thread-id") or "").strip() if not thread_id: return None - uid = str((container.labels or {}).get("uid") or "").strip() + file_thread_id = str(labels.get("file-thread-id") or thread_id).strip() + skills_thread_id = str(labels.get("skills-thread-id") or thread_id).strip() + uid = str(labels.get("uid") or "").strip() if not uid: return None - safe_thread_id = self._validate_thread_id(thread_id) + safe_file_thread_id = self._validate_thread_id(file_thread_id) + safe_skills_thread_id = self._validate_thread_id(skills_thread_id) safe_uid = self._validate_uid(uid) - if not self._is_expected_skills_mount(container, safe_thread_id): + if not self._is_expected_skills_mount(container, safe_skills_thread_id): logger.info("Discarding stale sandbox %s with unexpected skills mount", sandbox_id) try: self.delete(sandbox_id) 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, safe_uid): + if not self._has_expected_user_data_mounts(container, safe_file_thread_id, safe_uid): logger.info("Discarding stale sandbox %s with unexpected user-data mounts", sandbox_id) try: self.delete(sandbox_id) @@ -511,7 +541,16 @@ 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, uid: str, env: dict[str, str]): + def _build_pod_spec( + self, + sandbox_id: str, + thread_id: str, + uid: str, + env: dict[str, str], + *, + file_thread_id: str, + skills_thread_id: str, + ): pod_name = self._pod_name(sandbox_id) env_vars = [ self._client.V1EnvVar(name=key, value=value) @@ -521,7 +560,12 @@ class KubernetesProvisionerBackend: metadata=self._client.V1ObjectMeta( name=pod_name, labels={"app": "yuxi-sandbox", "sandbox-id": sandbox_id}, - annotations={"thread-id": thread_id, "uid": uid}, + annotations={ + "thread-id": thread_id, + "file-thread-id": file_thread_id, + "skills-thread-id": skills_thread_id, + "uid": uid, + }, ), spec=self._client.V1PodSpec( restart_policy="Never", @@ -537,11 +581,11 @@ class KubernetesProvisionerBackend: args=[ "chmod 777 /home/gem " f"&& mkdir -p /mnt/shared-data/threads/shared/{uid}/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"/mnt/shared-data/threads/{file_thread_id}/user-data/uploads " + f"/mnt/shared-data/threads/{file_thread_id}/user-data/outputs " + f"/mnt/shared-data/threads/{skills_thread_id}/skills " f"&& chmod -R 777 /mnt/shared-data/threads/shared/{uid}/workspace " - f"/mnt/shared-data/threads/{thread_id}/user-data ", + f"/mnt/shared-data/threads/{file_thread_id}/user-data ", ], volume_mounts=[ self._client.V1VolumeMount(name="home-dir", mount_path="/home/gem"), @@ -565,17 +609,17 @@ class KubernetesProvisionerBackend: self._client.V1VolumeMount( name="shared-data", mount_path="/home/gem/user-data/uploads", - sub_path=f"threads/{thread_id}/user-data/uploads", + sub_path=f"threads/{file_thread_id}/user-data/uploads", ), self._client.V1VolumeMount( name="shared-data", mount_path="/home/gem/user-data/outputs", - sub_path=f"threads/{thread_id}/user-data/outputs", + sub_path=f"threads/{file_thread_id}/user-data/outputs", ), self._client.V1VolumeMount( name="shared-data", mount_path="/home/gem/skills", - sub_path=f"threads/{thread_id}/skills", + sub_path=f"threads/{skills_thread_id}/skills", read_only=True, ), ], @@ -618,10 +662,43 @@ class KubernetesProvisionerBackend: ), ) - def create(self, sandbox_id: str, thread_id: str, uid: str, env: dict[str, str] | None = None) -> SandboxRecord: + @staticmethod + def _pod_has_expected_mounts(pod, *, file_thread_id: str, skills_thread_id: str, uid: str) -> bool: + expected_mounts = { + "/home/gem/user-data/workspace": f"threads/shared/{uid}/workspace", + "/home/gem/user-data/uploads": f"threads/{file_thread_id}/user-data/uploads", + "/home/gem/user-data/outputs": f"threads/{file_thread_id}/user-data/outputs", + "/home/gem/skills": f"threads/{skills_thread_id}/skills", + } + for container in getattr(pod.spec, "containers", []) or []: + if getattr(container, "name", None) != "sandbox": + continue + actual_mounts = { + str(getattr(mount, "mount_path", "") or "").rstrip("/"): str( + getattr(mount, "sub_path", "") or "" + ) + for mount in getattr(container, "volume_mounts", []) or [] + } + return all(actual_mounts.get(path) == sub_path for path, sub_path in expected_mounts.items()) + return False + + def create( + self, + sandbox_id: str, + thread_id: str, + uid: str, + env: dict[str, str] | None = None, + *, + file_thread_id: str | None = None, + skills_thread_id: str | None = None, + ) -> SandboxRecord: from kubernetes.client.rest import ApiException with self._lock: + safe_thread_id = LocalContainerProvisionerBackend._validate_thread_id(thread_id) + safe_file_thread_id = LocalContainerProvisionerBackend._validate_thread_id(file_thread_id or safe_thread_id) + safe_skills_thread_id = LocalContainerProvisionerBackend._validate_thread_id(skills_thread_id or safe_thread_id) + safe_uid = LocalContainerProvisionerBackend._validate_uid(uid) discovered = self.discover(sandbox_id) if discovered is not None: return discovered @@ -632,7 +709,14 @@ class KubernetesProvisionerBackend: try: self._core_api.create_namespaced_pod( namespace=self._namespace, - body=self._build_pod_spec(sandbox_id, thread_id, uid, env or {}), + body=self._build_pod_spec( + sandbox_id, + safe_thread_id, + safe_uid, + env or {}, + file_thread_id=safe_file_thread_id, + skills_thread_id=safe_skills_thread_id, + ), ) except ApiException as exc: if exc.status != 409: @@ -672,6 +756,31 @@ class KubernetesProvisionerBackend: return None raise + annotations = pod.metadata.annotations or {} + thread_id = str(annotations.get("thread-id") or "").strip() + if not thread_id: + return None + file_thread_id = str(annotations.get("file-thread-id") or thread_id).strip() + skills_thread_id = str(annotations.get("skills-thread-id") or thread_id).strip() + uid = str(annotations.get("uid") or "").strip() + if not uid: + return None + safe_file_thread_id = LocalContainerProvisionerBackend._validate_thread_id(file_thread_id) + safe_skills_thread_id = LocalContainerProvisionerBackend._validate_thread_id(skills_thread_id) + safe_uid = LocalContainerProvisionerBackend._validate_uid(uid) + if not self._pod_has_expected_mounts( + pod, + file_thread_id=safe_file_thread_id, + skills_thread_id=safe_skills_thread_id, + uid=safe_uid, + ): + logger.info("Discarding stale sandbox %s with unexpected pod mounts", sandbox_id) + try: + self.delete(sandbox_id) + except Exception as exc: + logger.warning("Failed to delete stale sandbox %s during discover: %s", sandbox_id, exc) + return None + node_port = None if service.spec and service.spec.ports: node_port = service.spec.ports[0].node_port @@ -843,7 +952,14 @@ 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, payload.uid, payload.env) + record = backend_impl.create( + payload.sandbox_id, + payload.thread_id, + payload.uid, + payload.env, + file_thread_id=payload.file_thread_id, + skills_thread_id=payload.skills_thread_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/agents/sandbox-architecture.md b/docs/agents/sandbox-architecture.md index 0642c7fa..966fb1d1 100644 --- a/docs/agents/sandbox-architecture.md +++ b/docs/agents/sandbox-architecture.md @@ -20,7 +20,7 @@ Docker 和 Kubernetes 不是互斥关系。Docker 解决的是“把一个进程 ## 二、当前项目的真实沙盒调用链 -当前仓库里，后端只支持 `SANDBOX_PROVIDER=provisioner`。当某个对话线程第一次需要执行文件操作或命令执行时，后端会基于 `thread_id` 生成一个稳定的 `sandbox_id`，然后请求 `sandbox-provisioner` 创建或复用对应沙盒。应用层拿到返回的 `sandbox_url` 之后，才会真正通过 `agent-sandbox` 客户端去调用远程沙盒的文件 API 和 shell API。 +当前仓库里，后端只支持 `SANDBOX_PROVIDER=provisioner`。当某个对话线程第一次需要执行文件操作或命令执行时，后端会基于文件线程与 skills 线程生成稳定的 `sandbox_id`，然后请求 `sandbox-provisioner` 创建或复用对应沙盒；普通 Agent 的文件线程和 skills 线程都回退为当前 `thread_id`。应用层拿到返回的 `sandbox_url` 之后，才会真正通过 `agent-sandbox` 客户端去调用远程沙盒的文件 API 和 shell API。 调用链可以概括为：Web/API 请求进入 Yuxi 后端，后端构造 `ProvisionerSandboxBackend`，再经由 `ProvisionerClient` 调用 `sandbox-provisioner` 的 `/api/sandboxes` 接口。`sandbox-provisioner` 根据 `SANDBOX_PROVISIONER_BACKEND` 选择内存占位实现、Docker 容器实现或 Kubernetes 实现。沙盒真正启动后，对外暴露一个 HTTP 地址，Yuxi 再使用这个地址完成执行命令、上传文件、下载文件、目录遍历等操作。 @@ -50,11 +50,11 @@ Docker 和 Kubernetes 不是互斥关系。Docker 解决的是“把一个进程 ## 五、Docker 本机后端是如何工作的 -当 `SANDBOX_PROVISIONER_BACKEND=docker` 时，`sandbox-provisioner` 会进入 `LocalContainerProvisionerBackend`。它会检查 Docker 是否可用，解析自身容器里 `/app/saves` 这个挂载点在宿主机上的真实路径，并据此推导出线程数据目录。随后它为每个 `thread_id` 准备一个稳定的 `sandbox_id`，把容器命名为类似 `yuxi-sandbox-` 的形式，并在 Docker 网络中启动真正的沙盒镜像。 +当 `SANDBOX_PROVISIONER_BACKEND=docker` 时，`sandbox-provisioner` 会进入 `LocalContainerProvisionerBackend`。它会检查 Docker 是否可用，解析自身容器里 `/app/saves` 这个挂载点在宿主机上的真实路径，并据此推导出线程数据目录。随后它为每组文件线程与 skills 线程准备一个稳定的 `sandbox_id`，把容器命名为类似 `yuxi-sandbox-` 的形式，并在 Docker 网络中启动真正的沙盒镜像。 这个沙盒镜像默认来自 `SANDBOX_IMAGE`，容器内部监听的端口默认是 `8080`。provisioner 在启动容器时，会把这个端口随机映射到宿主机上的一个可用端口，再用 `DOCKER_SANDBOX_HOST` 拼出形如 `http://host.docker.internal:` 的访问地址。Yuxi 后端拿到的就是这个地址。 -Docker 后端在启动沙盒时，会挂载两类关键目录。第一类是线程用户数据目录，挂载到容器内的 `/home/gem/user-data`，用于承载上传文件、输出文件以及工作目录。第二类是线程可见的 skills 目录，挂载到 `/home/gem/skills`，而且是只读挂载。除此之外，容器的 `/home/gem` 本身还会额外挂一个 `tmpfs`，原因是当前沙盒镜像启动时要求 `/home/gem` 可写，但 Yuxi 希望真正持久化的只有 `user-data` 下面的内容。 +Docker 后端在启动沙盒时，会挂载三类关键目录。第一类是用户级 workspace，挂载到容器内的 `/home/gem/user-data/workspace`。第二类是文件线程级 uploads/outputs，分别挂载到 `/home/gem/user-data/uploads` 和 `/home/gem/user-data/outputs`。第三类是 skills 线程可见的 skills 目录，挂载到 `/home/gem/skills`，而且是只读挂载。除此之外，容器的 `/home/gem` 本身还会额外挂一个 `tmpfs`，原因是当前沙盒镜像启动时要求 `/home/gem` 可写，但 Yuxi 希望真正持久化的只有 `user-data` 下面的内容。 为了避免长期空闲的沙盒一直占资源，provisioner 还带了一个 idle reaper。它会记录每个沙盒最近一次被 touch 的时间，超过 `SANDBOX_IDLE_TIMEOUT_SECONDS` 之后自动删除。当前默认空闲超时是 120 秒，但如果这个值小于命令执行超时，系统会自动把它提高到“命令超时 + 30 秒”，以免执行中的任务被误回收。 @@ -70,7 +70,7 @@ Docker 后端在启动沙盒时，会挂载两类关键目录。第一类是线 当 `SANDBOX_PROVISIONER_BACKEND=kubernetes` 时，`sandbox-provisioner` 会改用 Kubernetes Python 客户端。它会先加载 kubeconfig 或集群内配置，然后在指定的 namespace 中创建一个沙盒 Pod，再创建一个同名的 NodePort Service，把这个 Service 的 `nodePort` 暴露给 Yuxi 后端使用。 -Kubernetes 后端下，沙盒还是同一套镜像，还是暴露同样的 HTTP API，但存储方式和暴露方式变了。它不会依赖宿主机 Docker bind mount，而是要求有一个可写的 PVC。当前实现里真正使用的是 `THREAD_PVC`，Pod 会把这块共享存储挂到 `/mnt/shared-data`，然后用 `subPath` 的方式把 `threads//user-data` 挂到 `/home/gem/user-data`，把 `threads//skills` 挂到 `/home/gem/skills`。这样做的好处是线程之间的数据目录结构仍然可以和 Docker 模式保持一致。 +Kubernetes 后端下，沙盒还是同一套镜像，还是暴露同样的 HTTP API，但存储方式和暴露方式变了。它不会依赖宿主机 Docker bind mount，而是要求有一个可写的 PVC。当前实现里真正使用的是 `THREAD_PVC`，Pod 会把这块共享存储挂到 `/mnt/shared-data`，然后用 `subPath` 的方式把 `threads/shared//workspace` 挂到 `/home/gem/user-data/workspace`，把 `threads//user-data/uploads` 与 `threads//user-data/outputs` 分别挂到 uploads/outputs，把 `threads//skills` 挂到 `/home/gem/skills`。这样做的好处是目录结构仍然可以和 Docker 模式保持一致，同时允许子智能体共享父对话文件但隔离 skills。 需要特别说明的是，代码里虽然读取了 `SKILLS_PVC` 这个环境变量，但当前 Pod 规格实际没有使用单独的 skills PVC，而是统一从 `THREAD_PVC` 中切 `threads//skills` 这个子路径。因此，如果看到环境变量里同时出现 `SKILLS_PVC` 和 `THREAD_PVC`，应当以 `THREAD_PVC` 的真实挂载语义为准，`SKILLS_PVC` 目前更像一个预留字段。 @@ -133,13 +133,14 @@ saves/ │ │ ├── / │ │ └── ... │ ├── shared/ -│ │ └── workspace/ +│ │ └── / +│ │ └── workspace/ │ └── ... ``` -这里要重点理解 `workspace` 和 `uploads/outputs` 的区别。按照当前宿主机路径解析逻辑，`workspace` 被定义为共享目录，位置是 `saves/threads/shared/workspace`；而 `uploads` 和 `outputs` 属于线程私有目录，位置分别是 `saves/threads//user-data/uploads` 和 `saves/threads//user-data/outputs`。viewer 文件系统、artifact 下载接口以及路径解析函数都按这个语义工作，因此不同线程可以看到同一个 workspace，但看不到彼此的 uploads。 +这里要重点理解 `workspace` 和 `uploads/outputs` 的区别。按照当前宿主机路径解析逻辑，`workspace` 被定义为用户级共享目录，位置是 `saves/threads/shared//workspace`；而 `uploads` 和 `outputs` 属于文件线程目录，位置分别是 `saves/threads//user-data/uploads` 和 `saves/threads//user-data/outputs`。普通 Agent 的 `file_thread_id` 就是当前对话 `thread_id`，子智能体运行时则使用父对话作为 `file_thread_id`，因此可以读取父对话附件并把产物写回父对话 outputs。 -与此同时，运行时 provisioner 在创建 Docker 容器或 Kubernetes Pod 时，会把共享的 `saves/threads/shared/workspace` 单独挂到 `/home/gem/user-data/workspace`，再把当前线程自己的 `uploads/outputs` 分别挂到 `/home/gem/user-data/uploads` 和 `/home/gem/user-data/outputs`。因此在排查文件问题时，需要先明确一个前提：当前项目里同时存在“宿主机侧目录组织”和“容器内统一虚拟路径”两层概念。对外接口和 viewer 语义与底层挂载实现现在是一致的，workspace 是共享空间，而 uploads/outputs 仍然保持线程隔离。 +与此同时，运行时 provisioner 在创建 Docker 容器或 Kubernetes Pod 时，会把用户级 `saves/threads/shared//workspace` 单独挂到 `/home/gem/user-data/workspace`，再把文件线程的 `uploads/outputs` 分别挂到 `/home/gem/user-data/uploads` 和 `/home/gem/user-data/outputs`。因此在排查文件问题时，需要先明确一个前提：当前项目里同时存在“宿主机侧目录组织”和“容器内统一虚拟路径”两层概念。对外接口和 viewer 语义与底层挂载实现现在是一致的，workspace 是用户共享空间，而 uploads/outputs 跟随文件线程隔离或共享。 ## 九、路径暴露规则是什么 @@ -147,15 +148,15 @@ Yuxi 不会把整个容器文件系统都开放给 Agent 或 viewer。当前 vie `/home/gem/user-data` 是主要工作区。它允许模型和工具写入，但推荐语义并不相同。内置 prompt 中已经明确说明，`workspace` 应当放中间文件，`outputs` 应当放最终产物，`uploads` 是用户上传文件的位置。对于普通对话 Agent，文案甚至提示“非必要不要写 workspace，而优先写 outputs”。 -`/home/gem/skills` 是只读目录。它不是简单地把 `saves/skills` 整个暴露进去，而是先根据当前线程的 `_readable_skills`，把这些技能从全局 skills 根目录同步复制到 `saves/threads//skills`，再把这个线程目录只读挂进沙盒。这样做的结果是，不同线程看到的 skill 集可能不同，而且模型永远不能在运行时修改 skills 内容。 +`/home/gem/skills` 是只读目录。它不是简单地把 `saves/skills` 整个暴露进去，而是先根据当前运行时的 `_readable_skills`，把这些技能从全局 skills 根目录同步复制到 `saves/threads//skills`，再把这个 skills 线程目录只读挂进沙盒。这样做的结果是，不同主/子 Agent 看到的 skill 集可能不同，而且模型永远不能在运行时修改 skills 内容。 知识库访问不属于沙盒文件系统暴露规则。当前 Agent 可见知识库仍由用户权限和 Agent 配置共同决定，但只通过 `query_kb`、`open_kb_document` 等工具访问，不提供沙盒目录投影。 ## 十、skills、知识库、附件是怎么和沙盒结合的 -skills 的结合方式分成两层。第一层是提示词层，`prepare_agent_runtime_context` 会先根据当前线程配置的 `context.skills` 展开依赖闭包，`SkillsMiddleware` 再把 `_prompt_skills` 注入到系统提示里，让模型知道哪些 skill 存在、它们的入口文件一般在 `/home/gem/skills//SKILL.md`。第二层是文件系统层，运行时会调用 `sync_thread_readable_skills`，把 `_readable_skills` 对应的 skill 目录复制到线程自己的 `saves/threads//skills` 下，再由沙盒只读挂载到 `/home/gem/skills`。也就是说，skill 既是 prompt 中的能力说明，也是文件系统中的只读知识目录。 +skills 的结合方式分成两层。第一层是提示词层，`prepare_agent_runtime_context` 会先根据当前 Agent 配置的 `context.skills` 展开依赖闭包，`SkillsMiddleware` 再把 `_prompt_skills` 注入到系统提示里，让模型知道哪些 skill 存在、它们的入口文件一般在 `/home/gem/skills//SKILL.md`。第二层是文件系统层，运行时会调用 `sync_thread_readable_skills`，把 `_readable_skills` 对应的 skill 目录复制到当前 `skills_thread_id` 的 `saves/threads//skills` 下，再由沙盒只读挂载到 `/home/gem/skills`。也就是说，skill 既是 prompt 中的能力说明，也是文件系统中的只读知识目录。 -附件的结合方式更偏向“先落盘，再把路径告诉模型”。用户上传文件后，系统会先把原始文件写入 `saves/threads//user-data/uploads`。如果该文件可以被解析，系统还会额外生成一个 Markdown 副本，写到 `saves/threads//user-data/uploads/attachments/.md`。随后，LangGraph state 中会维护一份 `uploads` 列表，`AttachmentMiddleware` 会把这些可读路径注入系统提示，告诉模型优先用 `read_file` 去读取这些路径。因此，附件并不是“作为消息大段内联塞给模型”，而是被转换成沙盒文件系统中的路径对象。 +附件的结合方式更偏向“先落盘，再把路径告诉模型”。用户上传文件后，系统会先把原始文件写入 `saves/threads//user-data/uploads`。如果该文件可以被解析，系统还会额外生成一个 Markdown 副本，写到 `saves/threads//user-data/uploads/attachments/.md`。普通 Agent 的文件线程就是当前对话线程；子智能体沿用父对话文件线程，所以能访问父对话附件。随后，LangGraph state 中会维护一份 `uploads` 列表，`AttachmentMiddleware` 会把这些可读路径注入系统提示，告诉模型优先用 `read_file` 去读取这些路径。因此，附件并不是“作为消息大段内联塞给模型”，而是被转换成沙盒文件系统中的路径对象。 知识库不再与沙盒文件系统结合。它不会被复制到每个线程目录，也不会生成虚拟目录；模型通过专门的知识库工具检索，并在需要更完整上下文时用 `open_kb_document` 按 `resource_id` 和 `file_id` 打开文档内容。