feat: 优化沙箱线程作用域管理
This commit is contained in:
Wenjie Zhang
parent
b6240a1b4f
commit
3fe5d307bf
@ -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
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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}")
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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")
|
||||
|
||||
@ -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",
|
||||
)
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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-<id>` 的形式，并在 Docker 网络中启动真正的沙盒镜像。
|
||||
当 `SANDBOX_PROVISIONER_BACKEND=docker` 时，`sandbox-provisioner` 会进入 `LocalContainerProvisionerBackend`。它会检查 Docker 是否可用，解析自身容器里 `/app/saves` 这个挂载点在宿主机上的真实路径，并据此推导出线程数据目录。随后它为每组文件线程与 skills 线程准备一个稳定的 `sandbox_id`，把容器命名为类似 `yuxi-sandbox-<id>` 的形式，并在 Docker 网络中启动真正的沙盒镜像。
|
||||
|
||||
这个沙盒镜像默认来自 `SANDBOX_IMAGE`，容器内部监听的端口默认是 `8080`。provisioner 在启动容器时，会把这个端口随机映射到宿主机上的一个可用端口，再用 `DOCKER_SANDBOX_HOST` 拼出形如 `http://host.docker.internal:<random_port>` 的访问地址。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/<thread_id>/user-data` 挂到 `/home/gem/user-data`，把 `threads/<thread_id>/skills` 挂到 `/home/gem/skills`。这样做的好处是线程之间的数据目录结构仍然可以和 Docker 模式保持一致。
|
||||
Kubernetes 后端下，沙盒还是同一套镜像，还是暴露同样的 HTTP API，但存储方式和暴露方式变了。它不会依赖宿主机 Docker bind mount，而是要求有一个可写的 PVC。当前实现里真正使用的是 `THREAD_PVC`，Pod 会把这块共享存储挂到 `/mnt/shared-data`，然后用 `subPath` 的方式把 `threads/shared/<uid>/workspace` 挂到 `/home/gem/user-data/workspace`，把 `threads/<file_thread_id>/user-data/uploads` 与 `threads/<file_thread_id>/user-data/outputs` 分别挂到 uploads/outputs，把 `threads/<skills_thread_id>/skills` 挂到 `/home/gem/skills`。这样做的好处是目录结构仍然可以和 Docker 模式保持一致，同时允许子智能体共享父对话文件但隔离 skills。
|
||||
|
||||
需要特别说明的是，代码里虽然读取了 `SKILLS_PVC` 这个环境变量，但当前 Pod 规格实际没有使用单独的 skills PVC，而是统一从 `THREAD_PVC` 中切 `threads/<thread_id>/skills` 这个子路径。因此，如果看到环境变量里同时出现 `SKILLS_PVC` 和 `THREAD_PVC`，应当以 `THREAD_PVC` 的真实挂载语义为准，`SKILLS_PVC` 目前更像一个预留字段。
|
||||
|
||||
@ -133,13 +133,14 @@ saves/
|
||||
│ │ ├── <skill-slug>/
|
||||
│ │ └── ...
|
||||
│ ├── shared/
|
||||
│ │ └── workspace/
|
||||
│ │ └── <uid>/
|
||||
│ │ └── workspace/
|
||||
│ └── ...
|
||||
```
|
||||
|
||||
这里要重点理解 `workspace` 和 `uploads/outputs` 的区别。按照当前宿主机路径解析逻辑，`workspace` 被定义为共享目录，位置是 `saves/threads/shared/workspace`；而 `uploads` 和 `outputs` 属于线程私有目录，位置分别是 `saves/threads/<thread_id>/user-data/uploads` 和 `saves/threads/<thread_id>/user-data/outputs`。viewer 文件系统、artifact 下载接口以及路径解析函数都按这个语义工作，因此不同线程可以看到同一个 workspace，但看不到彼此的 uploads。
|
||||
这里要重点理解 `workspace` 和 `uploads/outputs` 的区别。按照当前宿主机路径解析逻辑，`workspace` 被定义为用户级共享目录，位置是 `saves/threads/shared/<uid>/workspace`；而 `uploads` 和 `outputs` 属于文件线程目录，位置分别是 `saves/threads/<file_thread_id>/user-data/uploads` 和 `saves/threads/<file_thread_id>/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/<uid>/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/<thread_id>/skills`，再把这个线程目录只读挂进沙盒。这样做的结果是，不同线程看到的 skill 集可能不同，而且模型永远不能在运行时修改 skills 内容。
|
||||
`/home/gem/skills` 是只读目录。它不是简单地把 `saves/skills` 整个暴露进去，而是先根据当前运行时的 `_readable_skills`，把这些技能从全局 skills 根目录同步复制到 `saves/threads/<skills_thread_id>/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/<slug>/SKILL.md`。第二层是文件系统层，运行时会调用 `sync_thread_readable_skills`，把 `_readable_skills` 对应的 skill 目录复制到线程自己的 `saves/threads/<thread_id>/skills` 下，再由沙盒只读挂载到 `/home/gem/skills`。也就是说，skill 既是 prompt 中的能力说明，也是文件系统中的只读知识目录。
|
||||
skills 的结合方式分成两层。第一层是提示词层，`prepare_agent_runtime_context` 会先根据当前 Agent 配置的 `context.skills` 展开依赖闭包，`SkillsMiddleware` 再把 `_prompt_skills` 注入到系统提示里，让模型知道哪些 skill 存在、它们的入口文件一般在 `/home/gem/skills/<slug>/SKILL.md`。第二层是文件系统层，运行时会调用 `sync_thread_readable_skills`，把 `_readable_skills` 对应的 skill 目录复制到当前 `skills_thread_id` 的 `saves/threads/<skills_thread_id>/skills` 下，再由沙盒只读挂载到 `/home/gem/skills`。也就是说，skill 既是 prompt 中的能力说明，也是文件系统中的只读知识目录。
|
||||
|
||||
附件的结合方式更偏向“先落盘，再把路径告诉模型”。用户上传文件后，系统会先把原始文件写入 `saves/threads/<thread_id>/user-data/uploads`。如果该文件可以被解析，系统还会额外生成一个 Markdown 副本，写到 `saves/threads/<thread_id>/user-data/uploads/attachments/<name>.md`。随后，LangGraph state 中会维护一份 `uploads` 列表，`AttachmentMiddleware` 会把这些可读路径注入系统提示，告诉模型优先用 `read_file` 去读取这些路径。因此，附件并不是“作为消息大段内联塞给模型”，而是被转换成沙盒文件系统中的路径对象。
|
||||
附件的结合方式更偏向“先落盘，再把路径告诉模型”。用户上传文件后，系统会先把原始文件写入 `saves/threads/<file_thread_id>/user-data/uploads`。如果该文件可以被解析，系统还会额外生成一个 Markdown 副本，写到 `saves/threads/<file_thread_id>/user-data/uploads/attachments/<name>.md`。普通 Agent 的文件线程就是当前对话线程；子智能体沿用父对话文件线程，所以能访问父对话附件。随后，LangGraph state 中会维护一份 `uploads` 列表，`AttachmentMiddleware` 会把这些可读路径注入系统提示，告诉模型优先用 `read_file` 去读取这些路径。因此，附件并不是“作为消息大段内联塞给模型”，而是被转换成沙盒文件系统中的路径对象。
|
||||
|
||||
知识库不再与沙盒文件系统结合。它不会被复制到每个线程目录，也不会生成虚拟目录；模型通过专门的知识库工具检索，并在需要更完整上下文时用 `open_kb_document` 按 `resource_id` 和 `file_id` 打开文档内容。
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user