diff --git a/backend/app/services/sandbox_file_syncer.py b/backend/app/services/sandbox_file_syncer.py index 445c7f58e4..887bc1c964 100644 --- a/backend/app/services/sandbox_file_syncer.py +++ b/backend/app/services/sandbox_file_syncer.py @@ -12,13 +12,14 @@ block the attachment upload flow. """ -import asyncio import logging import os from typing import Optional import httpx +from shared.utils.attachment_block import sanitize_attachment_filename + logger = logging.getLogger(__name__) # Configuration @@ -39,13 +40,7 @@ def _sanitize_filename(filename: str) -> str: Returns: Sanitized filename safe for use in file paths """ - # Get basename to remove any directory components - safe_name = os.path.basename(filename or "attachment") - # Replace path separators that might have been encoded - safe_name = safe_name.replace("/", "_").replace("\\", "_") - # Remove control characters - safe_name = safe_name.replace("\n", "").replace("\r", "") - return safe_name if safe_name else "attachment" + return sanitize_attachment_filename(filename, fallback="attachment") def build_sandbox_attachment_path(task_id: int, subtask_id: int, filename: str) -> str: @@ -282,7 +277,7 @@ async def sync_attachment_to_sandbox_background( ) -> None: """Sync attachment to sandbox in background. - This function is designed to be called from asyncio.create_task() + This function is designed to be scheduled as an asynchronous task. and handles all exceptions internally. Args: diff --git a/backend/init_data/skills/sandbox/download_attachment_tool.py b/backend/init_data/skills/sandbox/download_attachment_tool.py index 610d90ebce..85e6e6cefc 100644 --- a/backend/init_data/skills/sandbox/download_attachment_tool.py +++ b/backend/init_data/skills/sandbox/download_attachment_tool.py @@ -11,8 +11,10 @@ import json import logging import os +import re import time from typing import Optional +from urllib.parse import urlsplit, urlunsplit from langchain_core.callbacks import CallbackManagerForToolRun from pydantic import BaseModel, Field @@ -21,6 +23,28 @@ # Default API base URL for attachment downloads DEFAULT_API_BASE_URL = "http://backend:8000" +_ATTACHMENT_DOWNLOAD_PATH = re.compile( + r"^/api/attachments/(?P\d+)/download/?$" +) + + +def _build_download_url(attachment_url: str, api_base_url: str) -> str: + """Build a URL that accepts the task token available to sandbox tools.""" + relative_url = ( + attachment_url if attachment_url.startswith("/") else f"/{attachment_url}" + ) + parsed = urlsplit( + attachment_url + if attachment_url.startswith(("http://", "https://")) + else relative_url + ) + match = _ATTACHMENT_DOWNLOAD_PATH.fullmatch(parsed.path) + if not match: + raise ValueError("Only Wegent attachment download URLs are supported") + + backend = urlsplit(api_base_url.rstrip("/")) + executor_path = f"/api/attachments/{match.group('attachment_id')}/executor-download" + return urlunsplit((backend.scheme, backend.netloc, executor_path, "", "")) class SandboxDownloadAttachmentInput(BaseModel): @@ -61,7 +85,7 @@ class SandboxDownloadAttachmentTool(BaseSandboxTool): """Tool for downloading files from Wegent Backend to E2B sandbox. This tool downloads files from Wegent's attachment storage to the - sandbox environment via the /api/attachments/{id}/download endpoint. + sandbox environment via the task-token attachment endpoint. """ name: str = "download_attachment" @@ -132,8 +156,8 @@ async def _arun( effective_timeout = timeout_seconds or self.default_download_timeout logger.info( - f"[SandboxDownloadAttachmentTool] Downloading: {attachment_url} -> {save_path}, " - f"timeout={effective_timeout}s" + "[SandboxDownloadAttachmentTool] Downloading attachment: " + f"save_path={save_path}, timeout={effective_timeout}s" ) # Emit status update via WebSocket if available @@ -159,7 +183,7 @@ async def _arun( # Get or create sandbox logger.info( - f"[SandboxDownloadAttachmentTool] Getting or creating sandbox..." + "[SandboxDownloadAttachmentTool] Getting or creating sandbox..." ) sandbox, error = await sandbox_manager.get_or_create_sandbox( shell_type=self.default_shell_type, @@ -202,17 +226,10 @@ async def _arun( ) api_base_url = api_base_url.rstrip("/") - # Build full download URL - # attachment_url can be relative (e.g., /api/attachments/123/download) or full URL - if attachment_url.startswith("http://") or attachment_url.startswith( - "https://" - ): - download_url = attachment_url - else: - # Ensure attachment_url starts with / - if not attachment_url.startswith("/"): - attachment_url = f"/{attachment_url}" - download_url = f"{api_base_url}{attachment_url}" + # The attachment block exposes the browser download URL. Translate + # that exact Wegent route to the executor route because sandbox tools + # authenticate with a task token rather than a browser login token. + download_url = _build_download_url(attachment_url, api_base_url) # Get auth token auth_token = self.auth_token @@ -226,23 +243,27 @@ async def _arun( await self._emit_tool_status("failed", error_msg) return result - # Build curl command to download file + # Keep credentials and user-provided paths out of the command string. + # E2B passes these values directly as process environment variables. curl_cmd = ( - f"curl -s -f -L " - f'-H "Authorization: Bearer {auth_token}" ' - f'-o "{save_path}" ' - f'"{download_url}"' + "curl --silent --show-error --fail --location " + '--header "Authorization: Bearer $WEGENT_ATTACHMENT_TOKEN" ' + '--output "$WEGENT_ATTACHMENT_SAVE_PATH" ' + '"$WEGENT_ATTACHMENT_DOWNLOAD_URL"' ) - logger.info( - f"[SandboxDownloadAttachmentTool] Executing download via curl from {download_url}" - ) + logger.info("[SandboxDownloadAttachmentTool] Executing attachment download") # Execute curl command result_obj = await sandbox.commands.run( cmd=curl_cmd, cwd="/home/user", timeout=effective_timeout, + envs={ + "WEGENT_ATTACHMENT_TOKEN": auth_token, + "WEGENT_ATTACHMENT_SAVE_PATH": save_path, + "WEGENT_ATTACHMENT_DOWNLOAD_URL": download_url, + }, ) execution_time = time.time() - start_time diff --git a/backend/init_data/skills/sandbox/provider.py b/backend/init_data/skills/sandbox/provider.py index 5fa31ad7fa..3c99f7b249 100644 --- a/backend/init_data/skills/sandbox/provider.py +++ b/backend/init_data/skills/sandbox/provider.py @@ -67,6 +67,7 @@ def _prepare_base_params( "timeout": config.get("timeout", 7200), "auth_token": context.auth_token, # For skill downloads in sandbox "skill_identity_token": context.skill_identity_token, + "load_skill_tool": context.load_skill_tool, } @property diff --git a/backend/tests/init_data/skills/sandbox/test_download_attachment_tool.py b/backend/tests/init_data/skills/sandbox/test_download_attachment_tool.py new file mode 100644 index 0000000000..d949619798 --- /dev/null +++ b/backend/tests/init_data/skills/sandbox/test_download_attachment_tool.py @@ -0,0 +1,96 @@ +# SPDX-FileCopyrightText: 2026 Weibo, Inc. +# +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for task-token attachment downloads in the sandbox skill.""" + +import json +from types import SimpleNamespace + +import pytest + +from init_data.skills.sandbox.download_attachment_tool import ( + SandboxDownloadAttachmentTool, + _build_download_url, +) + + +def test_build_download_url_uses_executor_endpoint_for_attachment_url() -> None: + assert ( + _build_download_url("/api/attachments/123/download", "http://backend:8000") + == "http://backend:8000/api/attachments/123/executor-download" + ) + assert ( + _build_download_url( + "https://wegent.example/api/attachments/123/download?download=1", + "http://backend:8000", + ) + == "http://backend:8000/api/attachments/123/executor-download" + ) + + +def test_build_download_url_rejects_non_attachment_url() -> None: + with pytest.raises(ValueError, match="Only Wegent attachment download URLs"): + _build_download_url( + "https://files.example/report.csv", + "http://backend:8000", + ) + + +@pytest.mark.asyncio +async def test_download_uses_environment_for_credentials_and_paths(monkeypatch) -> None: + calls: list[dict] = [] + + class FakeFiles: + async def make_dir(self, path: str) -> None: + return None + + async def get_info(self, path: str) -> SimpleNamespace: + return SimpleNamespace(size=4) + + class FakeCommands: + async def run(self, **kwargs) -> SimpleNamespace: + calls.append(kwargs) + return SimpleNamespace(exit_code=0, stderr="") + + sandbox = SimpleNamespace( + sandbox_id="sandbox-1", + files=FakeFiles(), + commands=FakeCommands(), + ) + + class FakeManager: + async def get_or_create_sandbox(self, **kwargs): + return sandbox, None + + monkeypatch.setattr( + SandboxDownloadAttachmentTool, + "_get_sandbox_manager", + lambda self: FakeManager(), + ) + tool = SandboxDownloadAttachmentTool( + task_id=1, + subtask_id=2, + user_id=3, + user_name="alice", + auth_token="task-token", + api_base_url="http://backend:8000", + ) + + result = json.loads( + await tool._arun( + attachment_url="/api/attachments/123/download", + save_path="/home/user/report.csv", + ) + ) + + assert result["success"] is True + assert "task-token" not in calls[0]["cmd"] + assert "/home/user/report.csv" not in calls[0]["cmd"] + assert calls[0]["envs"] == { + "WEGENT_ATTACHMENT_TOKEN": "task-token", + "WEGENT_ATTACHMENT_SAVE_PATH": "/home/user/report.csv", + "WEGENT_ATTACHMENT_DOWNLOAD_URL": ( + "http://backend:8000/api/attachments/123/executor-download" + ), + } diff --git a/backend/tests/init_data/skills/sandbox/test_provider.py b/backend/tests/init_data/skills/sandbox/test_provider.py index 7f4f753e48..b54edcef7c 100644 --- a/backend/tests/init_data/skills/sandbox/test_provider.py +++ b/backend/tests/init_data/skills/sandbox/test_provider.py @@ -25,3 +25,21 @@ def test_prepare_base_params_includes_skill_identity_token(): params = provider._prepare_base_params(context, {}) assert params["skill_identity_token"] == "skill-jwt" + + +def test_prepare_base_params_includes_load_skill_tool(): + """Sandbox provider should expose active Skill state to sandbox creation.""" + provider = SandboxToolProvider() + load_skill_tool = object() + context = SkillToolContext( + task_id=1, + subtask_id=2, + user_id=3, + db_session=None, + ws_emitter=None, + load_skill_tool=load_skill_tool, + ) + + params = provider._prepare_base_params(context, {}) + + assert params["load_skill_tool"] is load_skill_tool diff --git a/backend/tests/services/test_sandbox_file_syncer.py b/backend/tests/services/test_sandbox_file_syncer.py index 346275b850..efaccab4fc 100644 --- a/backend/tests/services/test_sandbox_file_syncer.py +++ b/backend/tests/services/test_sandbox_file_syncer.py @@ -31,12 +31,8 @@ def test_filename_with_path(self): def test_filename_with_backslash(self): """Test filename with backslash is sanitized.""" - # On Linux, os.path.basename doesn't recognize Windows paths - # So backslashes are replaced with underscores result = _sanitize_filename("C:\\Windows\\System32\\file.txt") - # Result should not contain backslashes - assert "\\" not in result - assert "/" not in result + assert result == "file.txt" def test_filename_with_control_chars(self): """Test filename with control characters is sanitized.""" diff --git a/chat_shell/chat_shell/services/chat_service.py b/chat_shell/chat_shell/services/chat_service.py index 0c1d5bbb75..a077fa35de 100644 --- a/chat_shell/chat_shell/services/chat_service.py +++ b/chat_shell/chat_shell/services/chat_service.py @@ -260,6 +260,16 @@ async def _process_chat( ) context_metrics_tracker: ContextMetricsTracker | None = None + # The prompt advertises attachment paths inside the task sandbox. + # Materialize those files before the model can call sandbox tools so + # a newly-created sandbox cannot race the upload-time best-effort sync. + from chat_shell.services.sandbox_attachment_sync import ( + sync_chat_attachments_to_sandbox, + ) + + add_span_event("syncing_sandbox_attachments") + await sync_chat_attachments_to_sandbox(request) + # Prepare all context resources in parallel add_span_event("preparing_context") t0 = time.perf_counter() diff --git a/chat_shell/chat_shell/services/sandbox_attachment_sync.py b/chat_shell/chat_shell/services/sandbox_attachment_sync.py new file mode 100644 index 0000000000..236d7e701f --- /dev/null +++ b/chat_shell/chat_shell/services/sandbox_attachment_sync.py @@ -0,0 +1,349 @@ +# SPDX-FileCopyrightText: 2026 Weibo, Inc. +# +# SPDX-License-Identifier: Apache-2.0 + +"""Prepare Chat Shell attachments in the task sandbox before model execution.""" + +from __future__ import annotations + +import logging +from typing import Any + +import httpx + +from chat_shell.core.config import settings +from shared.models.execution import ExecutionRequest +from shared.telemetry.decorators import add_span_event, trace_async +from shared.utils.attachment_block import ( + build_attachment_download_url, + build_sandbox_path, +) + +logger = logging.getLogger(__name__) + +_SANDBOX_SKILL_NAME = "sandbox" +_ATTACHMENT_DOWNLOAD_TIMEOUT = 180.0 + + +def _skill_name(value: Any) -> str: + """Return a normalized skill name from a name or skill config.""" + if isinstance(value, str): + return value.strip().lower() + if isinstance(value, dict): + return str(value.get("name") or "").strip().lower() + return "" + + +def _sandbox_skill_available(request: ExecutionRequest) -> bool: + """Return whether this request can expose sandbox-backed attachment paths.""" + configured_skills = ( + list(request.skill_names or []) + + list(request.preload_skills or []) + + list(request.user_selected_skills or []) + + list(request.skill_configs or []) + ) + return any(_skill_name(item) == _SANDBOX_SKILL_NAME for item in configured_skills) + + +def _sandbox_skill_config(request: ExecutionRequest) -> dict[str, Any]: + """Return the provider config used by the sandbox tools themselves.""" + for skill_config in request.skill_configs or []: + if _skill_name(skill_config) != _SANDBOX_SKILL_NAME: + continue + config = skill_config.get("config") if isinstance(skill_config, dict) else None + return config if isinstance(config, dict) else {} + return {} + + +def _backend_url(request: ExecutionRequest) -> str: + """Resolve the Backend base URL without its internal API suffix.""" + url = request.backend_url or settings.REMOTE_STORAGE_URL + return url.rstrip("/").removesuffix("/api/internal") + + +def _integer(value: Any, default: int = 0) -> int: + """Coerce trusted request metadata without aborting the entire chat.""" + try: + return int(value) + except (TypeError, ValueError): + return default + + +def _attachment_value(attachment: dict[str, Any], *keys: str) -> Any: + for key in keys: + value = attachment.get(key) + if value is not None: + return value + return None + + +def _attachment_filename(attachment: dict[str, Any]) -> str: + return str( + _attachment_value( + attachment, + "original_filename", + "originalFilename", + "filename", + "name", + ) + or "attachment" + ) + + +def _attachment_subtask_id( + request: ExecutionRequest, attachment: dict[str, Any] +) -> int: + value = _attachment_value(attachment, "subtask_id", "subtaskId") + return _integer(value or request.user_subtask_id or request.subtask_id) + + +def _failed_attachment_prompt( + prompt: str | list[dict[str, Any]], + failed_attachments: list[dict[str, Any]], +) -> str | list[dict[str, Any]]: + """Stop claiming failed attachments are already present in the sandbox.""" + if not failed_attachments: + return prompt + + warning = _failed_attachment_warning(failed_attachments) + + def rewrite(text: str, *, append_warning: bool) -> str: + rewritten = _rewrite_failed_attachment_paths(text, failed_attachments) + return rewritten + warning if append_warning else rewritten + + if isinstance(prompt, str): + return rewrite(prompt, append_warning=True) + if not isinstance(prompt, list): + return prompt + + warning_appended = False + rewritten_blocks: list[dict[str, Any]] = [] + for block in prompt: + if not isinstance(block, dict): + rewritten_blocks.append(block) + continue + text = block.get("text") + if block.get("type") not in {"input_text", "text"} or not isinstance(text, str): + rewritten_blocks.append(block) + continue + updated = dict(block) + updated["text"] = rewrite(text, append_warning=not warning_appended) + warning_appended = True + rewritten_blocks.append(updated) + return rewritten_blocks + + +def _rewrite_failed_attachment_paths( + text: str, failed_attachments: list[dict[str, Any]] +) -> str: + rewritten = text + for attachment in failed_attachments: + path = str(attachment.get("local_path") or "") + if not path: + continue + rewritten = rewritten.replace( + f"File Path(already in sandbox): {path}", + f"File Path(not synchronized): {path}", + ) + rewritten = rewritten.replace( + f"File Path in Sandbox: {path}", + f"File Path(not synchronized): {path}", + ) + return rewritten + + +def _failed_attachment_warning(failed_attachments: list[dict[str, Any]]) -> str: + lines = [ + "", + "", + "The following attachments are not yet synchronized to the sandbox:", + ] + for attachment in failed_attachments: + attachment_id = _integer(attachment.get("id")) + filename = _attachment_filename(attachment) + path = str(attachment.get("local_path") or "") + download_url = build_attachment_download_url(attachment_id) + lines.append( + f"- {filename} (ID: {attachment_id}). Use download_attachment with " + f"attachment_url={download_url} and save_path={path}." + ) + return "\n".join(lines) + + +def _mark_all_failed( + request: ExecutionRequest, attachments: list[dict[str, Any]], error: str +) -> None: + failed: list[dict[str, Any]] = [] + for attachment in attachments: + updated = dict(attachment) + filename = _attachment_filename(updated) + subtask_id = _attachment_subtask_id(request, updated) + updated.update( + { + "status": "failed", + "error": error, + "local_path": build_sandbox_path(request.task_id, subtask_id, filename), + "subtask_id": subtask_id, + } + ) + failed.append(updated) + request.attachments = failed + request.prompt = _failed_attachment_prompt(request.prompt, failed) + + +async def _file_already_synced( + sandbox: Any, path: str, expected_size: int | None +) -> bool: + """Return whether a sandbox file already exists with the expected size.""" + if not expected_size or expected_size < 0: + return False + try: + file_info = await sandbox.files.get_info(path) + except Exception: + return False + return int(file_info.size) == expected_size + + +async def _sync_one_attachment( + *, + client: httpx.AsyncClient, + sandbox: Any, + request: ExecutionRequest, + attachment: dict[str, Any], + backend_url: str, +) -> dict[str, Any]: + """Download one attachment through task-token auth and write it to sandbox.""" + updated = dict(attachment) + attachment_id = _integer(updated.get("id")) + filename = _attachment_filename(updated) + subtask_id = _attachment_subtask_id(request, updated) + local_path = build_sandbox_path(request.task_id, subtask_id, filename) + updated.update({"local_path": local_path, "subtask_id": subtask_id}) + + if attachment_id <= 0 or not local_path: + updated.update({"status": "failed", "error": "Invalid attachment metadata"}) + return updated + + raw_size = _attachment_value(updated, "file_size", "fileSize") + expected_size = _integer(raw_size, default=-1) if raw_size is not None else None + if await _file_already_synced(sandbox, local_path, expected_size): + updated.update({"status": "success", "error": None}) + return updated + + url = f"{backend_url}/api/attachments/{attachment_id}/executor-download" + try: + response = await client.get( + url, + headers={"Authorization": f"Bearer {request.auth_token}"}, + ) + response.raise_for_status() + parent_dir = local_path.rsplit("/", 1)[0] + try: + await sandbox.files.make_dir(parent_dir) + except Exception: + logger.debug( + "[sandbox_attachment_sync] Parent directory already exists: %s", + parent_dir, + ) + await sandbox.files.write(local_path, response.content) + updated.update({"status": "success", "error": None}) + except httpx.HTTPStatusError as exc: + updated.update( + { + "status": "failed", + "error": f"Attachment download returned HTTP {exc.response.status_code}", + } + ) + except Exception as exc: + updated.update({"status": "failed", "error": str(exc)}) + return updated + + +async def _create_task_sandbox(request: ExecutionRequest) -> tuple[Any, str | None]: + """Create or reconnect to the sandbox using the active skill's config.""" + from chat_shell.tools.sandbox._base import SandboxManager + + sandbox_config = _sandbox_skill_config(request) + manager = SandboxManager.get_instance( + task_id=request.task_id, + user_id=request.user_id, + user_name=request.user_name, + bot_config=sandbox_config.get("bot_config", []), + auth_token=request.auth_token, + skill_identity_token=request.skill_identity_token, + ) + return await manager.get_or_create_sandbox( + shell_type=sandbox_config.get("default_shell_type", "ClaudeCode"), + workspace_ref=None, + task_type="sandbox", + ) + + +async def _sync_attachments( + request: ExecutionRequest, + sandbox: Any, + attachments: list[dict[str, Any]], +) -> list[dict[str, Any]]: + backend_url = _backend_url(request) + async with httpx.AsyncClient( + timeout=_ATTACHMENT_DOWNLOAD_TIMEOUT, + follow_redirects=True, + ) as client: + return [ + await _sync_one_attachment( + client=client, + sandbox=sandbox, + request=request, + attachment=attachment, + backend_url=backend_url, + ) + for attachment in attachments + ] + + +@trace_async( + span_name="chat_service.sync_sandbox_attachments", + tracer_name="chat_shell.services", + extract_attributes=lambda request, *args, **kwargs: { + "attachment.task_id": request.task_id, + "attachment.subtask_id": request.subtask_id, + "attachment.count": len(request.attachments or []), + }, +) +async def sync_chat_attachments_to_sandbox(request: ExecutionRequest) -> None: + """Synchronize current-turn attachments before the model can use sandbox tools.""" + attachments = [ + dict(item) for item in (request.attachments or []) if isinstance(item, dict) + ] + if not attachments or not _sandbox_skill_available(request): + return + + if not request.auth_token: + _mark_all_failed(request, attachments, "Task authentication token is missing") + return + + # Import and initialize E2B only for requests that can use the sandbox. + sandbox, error = await _create_task_sandbox(request) + if error or sandbox is None: + _mark_all_failed(request, attachments, error or "Sandbox is unavailable") + return + + synced = await _sync_attachments(request, sandbox, attachments) + request.attachments = synced + failed = [item for item in synced if item.get("status") == "failed"] + request.prompt = _failed_attachment_prompt(request.prompt, failed) + add_span_event( + "sandbox_attachments_synced", + { + "success_count": len(synced) - len(failed), + "failed_count": len(failed), + }, + ) + logger.info( + "[sandbox_attachment_sync] Completed: task_id=%s, subtask_id=%s, " + "success_count=%s, failed_count=%s", + request.task_id, + request.subtask_id, + len(synced) - len(failed), + len(failed), + ) diff --git a/chat_shell/chat_shell/skills/context.py b/chat_shell/chat_shell/skills/context.py index 4d74872fb0..13ef8b2ad6 100644 --- a/chat_shell/chat_shell/skills/context.py +++ b/chat_shell/chat_shell/skills/context.py @@ -30,6 +30,7 @@ class SkillToolContext: user_name: Username for identifying the user auth_token: JWT token for API authentication (e.g., attachment upload/download) skill_identity_token: JWT token for skill identity verification + load_skill_tool: LoadSkillTool tracking skills active in the current session """ task_id: int @@ -41,6 +42,7 @@ class SkillToolContext: user_name: str = "" auth_token: str = "" # JWT token for API authentication skill_identity_token: str = "" # JWT token for skill identity verification + load_skill_tool: Any = None def get_config(self, key: str, default: Any = None) -> Any: """Get a configuration value from skill config. diff --git a/chat_shell/chat_shell/tools/sandbox/_base.py b/chat_shell/chat_shell/tools/sandbox/_base.py index c6d22c0a78..9a8a8f9b8d 100644 --- a/chat_shell/chat_shell/tools/sandbox/_base.py +++ b/chat_shell/chat_shell/tools/sandbox/_base.py @@ -210,6 +210,7 @@ def __init__( bot_config: list = None, auth_token: str = "", skill_identity_token: str = "", + load_skill_tool: Any = None, ): """Initialize sandbox manager. @@ -223,6 +224,7 @@ def __init__( bot_config: Bot configuration list (optional) auth_token: API auth token for skill downloads (optional) skill_identity_token: JWT token for skill identity verification (optional) + load_skill_tool: Tool tracking skills active in the current chat session """ self.task_id = task_id self.user_id = user_id @@ -231,6 +233,7 @@ def __init__( self.bot_config = bot_config or [] self.auth_token = auth_token self.skill_identity_token = skill_identity_token + self.load_skill_tool = load_skill_tool # Ensure E2B SDK is patched patch_e2b_sdk() @@ -245,6 +248,7 @@ def get_instance( bot_config: list = None, auth_token: str = "", skill_identity_token: str = "", + load_skill_tool: Any = None, ) -> "SandboxManager": """Get or create a singleton SandboxManager instance for the given task_id. @@ -256,6 +260,7 @@ def get_instance( bot_config: Bot configuration list (optional) auth_token: API auth token for skill downloads (optional) skill_identity_token: JWT token for skill identity verification (optional) + load_skill_tool: Tool tracking skills active in the current chat session Returns: SandboxManager instance for the task_id @@ -270,6 +275,7 @@ def get_instance( bot_config, auth_token, skill_identity_token, + load_skill_tool, ) else: instance = cls._instances[task_id] @@ -279,6 +285,8 @@ def get_instance( instance.skill_identity_token = skill_identity_token if bot_config: instance.bot_config = bot_config + if load_skill_tool is not None: + instance.load_skill_tool = load_skill_tool logger.debug( f"[SandboxManager] Reusing existing instance for task_id={task_id}" ) @@ -351,6 +359,17 @@ async def get_or_create_sandbox( if self.skill_identity_token: metadata["skill_identity_token"] = self.skill_identity_token + if self.load_skill_tool is not None and hasattr( + self.load_skill_tool, "get_loaded_skills" + ): + required_skills = sorted( + set(self.load_skill_tool.get_loaded_skills()) - {"sandbox"} + ) + if required_skills: + metadata["required_skills"] = json.dumps( + required_skills, ensure_ascii=False + ) + # Serialize bot_config to JSON string if available # E2B SDK only accepts string values in metadata if self.bot_config: @@ -429,6 +448,7 @@ class BaseSandboxTool(BaseTool): bot_config: list = [] # Bot config list [{shell_type, agent_config}, ...] auth_token: str = "" # API auth token for skill downloads skill_identity_token: str = "" # JWT token for skill identity verification + load_skill_tool: Any = None # Tracks skills active in this chat session # Configuration default_shell_type: str = "ClaudeCode" @@ -454,6 +474,7 @@ def _get_sandbox_manager(self) -> SandboxManager: bot_config=self.bot_config, auth_token=self.auth_token, skill_identity_token=self.skill_identity_token, + load_skill_tool=self.load_skill_tool, ) def kill_sandbox(self) -> None: diff --git a/chat_shell/chat_shell/tools/skill_factory.py b/chat_shell/chat_shell/tools/skill_factory.py index f8c87a4761..cbcd1df484 100644 --- a/chat_shell/chat_shell/tools/skill_factory.py +++ b/chat_shell/chat_shell/tools/skill_factory.py @@ -144,6 +144,7 @@ async def _create_provider_tools_for_skill( user_name: Optional[str] = None, auth_token: Optional[str] = None, skill_identity_token: Optional[str] = None, + load_skill_tool: Optional[Any] = None, ) -> list[Any]: """Load a skill provider if needed and create concrete tool instances.""" from chat_shell.skills import SkillToolContext @@ -237,6 +238,7 @@ async def _create_provider_tools_for_skill( user_name=user_name, auth_token=auth_token, skill_identity_token=skill_identity_token, + load_skill_tool=load_skill_tool, ) create_tools_start = time.perf_counter() @@ -483,6 +485,7 @@ async def load_deferred_tools( user_name=user_name, auth_token=auth_token, skill_identity_token=skill_identity_token, + load_skill_tool=load_skill_tool, ) logger.info( "[skill_factory_perf] skill=%s deferred_provider_load=%.2fms " @@ -525,6 +528,7 @@ async def load_deferred_tools( user_name=user_name, auth_token=auth_token, skill_identity_token=skill_identity_token, + load_skill_tool=load_skill_tool, ) if skill_tools: diff --git a/chat_shell/tests/test_sandbox_attachment_sync.py b/chat_shell/tests/test_sandbox_attachment_sync.py new file mode 100644 index 0000000000..7c329b5109 --- /dev/null +++ b/chat_shell/tests/test_sandbox_attachment_sync.py @@ -0,0 +1,209 @@ +# SPDX-FileCopyrightText: 2026 Weibo, Inc. +# +# SPDX-License-Identifier: Apache-2.0 + +"""Regression tests for preparing Chat attachments in the task sandbox.""" + +import httpx +import pytest + +from chat_shell.services import sandbox_attachment_sync +from chat_shell.services.sandbox_attachment_sync import ( + sync_chat_attachments_to_sandbox, +) +from chat_shell.tools.sandbox._base import SandboxManager +from shared.models.execution import ExecutionRequest +from shared.utils.attachment_block import build_sandbox_path + + +class _FakeFiles: + def __init__(self) -> None: + self.directories: list[str] = [] + self.writes: list[tuple[str, bytes]] = [] + + async def get_info(self, path: str) -> None: + raise FileNotFoundError(path) + + async def make_dir(self, path: str) -> None: + self.directories.append(path) + + async def write(self, path: str, content: bytes) -> None: + self.writes.append((path, content)) + + +class _FakeSandbox: + def __init__(self) -> None: + self.sandbox_id = "task-sandbox" + self.files = _FakeFiles() + + +class _FakeManager: + def __init__(self, sandbox: _FakeSandbox | None, error: str | None = None) -> None: + self.sandbox = sandbox + self.error = error + self.calls: list[dict] = [] + + async def get_or_create_sandbox(self, **kwargs): + self.calls.append(kwargs) + return self.sandbox, self.error + + +class _FakeAsyncClient: + calls: list[tuple[str, dict[str, str]]] = [] + + def __init__(self, **kwargs) -> None: + self.options = kwargs + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, traceback) -> None: + return None + + async def get(self, url: str, headers: dict[str, str]) -> httpx.Response: + self.calls.append((url, headers)) + return httpx.Response( + 200, + content=b"name,value\nalpha,1\n", + request=httpx.Request("GET", url), + ) + + +@pytest.fixture(autouse=True) +def _reset_fake_client() -> None: + _FakeAsyncClient.calls = [] + + +@pytest.mark.asyncio +async def test_syncs_attachment_before_sandbox_tools_can_read_it(monkeypatch) -> None: + sandbox = _FakeSandbox() + manager = _FakeManager(sandbox) + manager_factory_calls: list[dict] = [] + + def get_manager(cls, **kwargs): + manager_factory_calls.append(kwargs) + return manager + + monkeypatch.setattr( + SandboxManager, + "get_instance", + classmethod(get_manager), + ) + monkeypatch.setattr(sandbox_attachment_sync.httpx, "AsyncClient", _FakeAsyncClient) + + path = build_sandbox_path(100, 201, "热点复盘.csv") + request = ExecutionRequest( + task_id=100, + subtask_id=202, + user_subtask_id=201, + user_id=3, + user_name="alice", + prompt=f"File Path(already in sandbox): {path}", + skill_names=["sandbox"], + skill_configs=[ + { + "name": "sandbox", + "config": { + "default_shell_type": "Agno", + "bot_config": [{"shell_type": "Agno"}], + }, + } + ], + auth_token="task-token", + backend_url="http://backend:8000", + attachments=[ + { + "id": 77, + "original_filename": "热点复盘.csv", + "mime_type": "text/csv", + "file_size": 19, + "subtask_id": 201, + } + ], + ) + + await sync_chat_attachments_to_sandbox(request) + + assert manager_factory_calls == [ + { + "task_id": 100, + "user_id": 3, + "user_name": "alice", + "bot_config": [{"shell_type": "Agno"}], + "auth_token": "task-token", + "skill_identity_token": "", + } + ] + assert manager.calls == [ + { + "shell_type": "Agno", + "workspace_ref": None, + "task_type": "sandbox", + } + ] + assert _FakeAsyncClient.calls == [ + ( + "http://backend:8000/api/attachments/77/executor-download", + {"Authorization": "Bearer task-token"}, + ) + ] + assert sandbox.files.writes == [(path, b"name,value\nalpha,1\n")] + assert request.attachments[0]["status"] == "success" + assert request.attachments[0]["local_path"] == path + assert "File Path(already in sandbox)" in request.prompt + + +@pytest.mark.asyncio +async def test_failed_sync_stops_claiming_attachment_is_in_sandbox(monkeypatch) -> None: + manager = _FakeManager(None, "sandbox unavailable") + monkeypatch.setattr( + SandboxManager, + "get_instance", + classmethod(lambda cls, **kwargs: manager), + ) + + path = build_sandbox_path(100, 201, "report.csv") + request = ExecutionRequest( + task_id=100, + subtask_id=202, + user_subtask_id=201, + user_id=3, + user_name="alice", + prompt=f"File Path(already in sandbox): {path}", + preload_skills=["sandbox"], + auth_token="task-token", + attachments=[{"id": 77, "original_filename": "report.csv"}], + ) + + await sync_chat_attachments_to_sandbox(request) + + assert request.attachments[0]["status"] == "failed" + assert "File Path(already in sandbox)" not in request.prompt + assert f"File Path(not synchronized): {path}" in request.prompt + assert "attachment_url=/api/attachments/77/download" in request.prompt + assert f"save_path={path}" in request.prompt + + +@pytest.mark.asyncio +async def test_request_without_sandbox_skill_does_not_create_sandbox( + monkeypatch, +) -> None: + def fail_if_called(cls, **kwargs): + raise AssertionError("sandbox should not be created") + + monkeypatch.setattr( + SandboxManager, + "get_instance", + classmethod(fail_if_called), + ) + request = ExecutionRequest( + task_id=100, + subtask_id=202, + prompt="plain attachment", + auth_token="task-token", + attachments=[{"id": 77, "original_filename": "report.csv"}], + ) + + await sync_chat_attachments_to_sandbox(request) + + assert request.attachments == [{"id": 77, "original_filename": "report.csv"}] diff --git a/chat_shell/tests/test_sandbox_skill_identity.py b/chat_shell/tests/test_sandbox_skill_identity.py index eec740c749..faf1eb1495 100644 --- a/chat_shell/tests/test_sandbox_skill_identity.py +++ b/chat_shell/tests/test_sandbox_skill_identity.py @@ -4,6 +4,7 @@ """Regression tests for sandbox skill identity token propagation.""" +import json from types import SimpleNamespace import pytest @@ -47,3 +48,31 @@ async def test_sandbox_manager_includes_skill_identity_token_in_metadata( assert ( _AsyncSandboxStub.last_call["metadata"]["skill_identity_token"] == "skill-jwt" ) + + +@pytest.mark.asyncio +async def test_sandbox_manager_includes_active_skills_in_metadata(monkeypatch): + """Sandbox activation should require every Skill loaded by Chat Shell.""" + monkeypatch.setattr( + "e2b_code_interpreter.AsyncSandbox", + _AsyncSandboxStub, + raising=False, + ) + load_skill_tool = SimpleNamespace( + get_loaded_skills=lambda: {"sandbox", "abtest-file-analyzer"} + ) + manager = SandboxManager( + task_id=1, + user_id=2, + user_name="alice", + auth_token="task-jwt", + load_skill_tool=load_skill_tool, + ) + + sandbox, error = await manager.get_or_create_sandbox("ClaudeCode") + + assert error is None + assert sandbox is not None + assert json.loads(_AsyncSandboxStub.last_call["metadata"]["required_skills"]) == [ + "abtest-file-analyzer" + ] diff --git a/executor/src/agents/mod.rs b/executor/src/agents/mod.rs index 159679537e..69ed29807a 100644 --- a/executor/src/agents/mod.rs +++ b/executor/src/agents/mod.rs @@ -315,18 +315,24 @@ impl AgentEngine for AgentProcessEngine { } log_executor_event("command planned", &command_fields); if request.resolved_agent_kind() == AgentKind::ClaudeCode { - spec = runtime_capabilities::prepare_claude_runtime(&request, spec) - .await - .unwrap_or_else(|error| { + spec = match runtime_capabilities::prepare_claude_runtime( + &request, spec, + ) + .await + { + Ok(spec) => spec, + Err(message) => { let mut failed_fields = task_fields(&request.task_id, &request.subtask_id); - failed_fields.push(("error_len", error.len().to_string())); + failed_fields + .push(("error_len", message.len().to_string())); log_executor_event( "claude runtime capability preparation failed", &failed_fields, ); - build_claude_command(&request, &planner.claude_binary) - }); + return ExecutionOutcome::Failed { message }; + } + }; restore_claude_plugin_cache(&request, &spec); deploy_claude_task_skills(&request, &spec).await; configure_claude_default_settings(&request, &spec); @@ -409,18 +415,24 @@ impl AgentEngine for AgentProcessEngine { } log_executor_event("command planned", &command_fields); if request.resolved_agent_kind() == AgentKind::ClaudeCode { - spec = runtime_capabilities::prepare_claude_runtime(&request, spec) - .await - .unwrap_or_else(|error| { + spec = match runtime_capabilities::prepare_claude_runtime( + &request, spec, + ) + .await + { + Ok(spec) => spec, + Err(message) => { let mut failed_fields = task_fields(&request.task_id, &request.subtask_id); - failed_fields.push(("error_len", error.len().to_string())); + failed_fields + .push(("error_len", message.len().to_string())); log_executor_event( "claude runtime capability preparation failed", &failed_fields, ); - build_claude_command(&request, &planner.claude_binary) - }); + return ExecutionOutcome::Failed { message }; + } + }; restore_claude_plugin_cache(&request, &spec); deploy_claude_task_skills(&request, &spec).await; configure_claude_default_settings(&request, &spec); diff --git a/executor/src/agents/runtime_capabilities.rs b/executor/src/agents/runtime_capabilities.rs index 8a285e8543..a4f98690e8 100644 --- a/executor/src/agents/runtime_capabilities.rs +++ b/executor/src/agents/runtime_capabilities.rs @@ -3,7 +3,7 @@ // SPDX-License-Identifier: Apache-2.0 use std::{ - collections::{BTreeMap, HashMap}, + collections::{BTreeMap, BTreeSet, HashMap}, env, fs, io::Cursor, path::{Component, Path, PathBuf}, @@ -296,7 +296,7 @@ pub async fn prepare_claude_runtime( .get("SKILLS_DIR") .map(PathBuf::from) .unwrap_or_else(|| config_dir.join("skills")); - deploy_request_skills(request, &skills_dir).await; + deploy_request_skills(request, &skills_dir).await?; let global_mcps = load_global_mcp_records(); log_runtime_event( @@ -352,7 +352,11 @@ pub async fn prepare_codex_runtime(request: &ExecutionRequest) { .map(PathBuf::from) .unwrap_or_else(|| workspace_root().join(&request.task_id)); let codex_skills_dir = codex_skills_dir(&task_dir); - deploy_request_skills(request, &codex_skills_dir).await; + if let Err(error) = deploy_request_skills(request, &codex_skills_dir).await { + let mut fields = task_fields(&request.task_id, &request.subtask_id); + push_error_fields(&mut fields, error); + log_executor_event("codex Skill deployment failed", &fields); + } } pub fn request_mcp_config_overrides(request: &ExecutionRequest) -> Vec { @@ -364,9 +368,16 @@ pub fn request_mcp_config_overrides(request: &ExecutionRequest) -> Vec { overrides } -async fn deploy_request_skills(request: &ExecutionRequest, skills_dir: &Path) { +async fn deploy_request_skills( + request: &ExecutionRequest, + skills_dir: &Path, +) -> Result<(), String> { + let required_skills = required_skill_names(request); let Some(primary_bot) = primary_bot(request) else { - return; + return required_skills + .is_empty() + .then_some(()) + .ok_or_else(|| "required Skills cannot be deployed without a bot".to_owned()); }; let Some(plan) = build_skill_deployment_plan( primary_bot, @@ -377,15 +388,109 @@ async fn deploy_request_skills(request: &ExecutionRequest, skills_dir: &Path) { skip_existing: false, }, ) else { - return; + return required_skills.is_empty().then_some(()).ok_or_else(|| { + format!( + "required Skills are missing from the deployment plan: {}", + required_skills.join(", ") + ) + }); }; let api_base_url = request_api_base_url(request); - if let Err(error) = deploy_skills(&plan, &api_base_url).await { - let mut fields = task_fields(&request.task_id, &request.subtask_id); - fields.push(("error_len", error.len().to_string())); - log_executor_event("skill deployment skipped after error", &fields); + let report = deploy_skills(&plan, &api_base_url).await?; + let missing_required = missing_required_skills(&required_skills, &plan, &report); + if !missing_required.is_empty() { + return Err(format!( + "required Skill deployment failed: {}", + missing_required.join(", ") + )); + } + Ok(()) +} + +pub async fn sync_skills_for_request(request: ExecutionRequest) -> Result { + let required_skills = required_skill_names(&request); + let Some(primary_bot) = primary_bot(&request) else { + return required_skills + .is_empty() + .then(|| json!({"success": true, "skill_count": 0, "failed_skills": []})) + .ok_or_else(|| "required Skills cannot be deployed without a bot".to_owned()); + }; + let skills_dir = claude_config_dir(&request, None) + .ok_or_else(|| "Claude config directory is unavailable".to_owned())? + .join("skills"); + let Some(plan) = build_skill_deployment_plan( + primary_bot, + &request, + SkillDeploymentOptions { + skills_dir, + clear_cache: false, + skip_existing: false, + }, + ) else { + return required_skills + .is_empty() + .then(|| json!({"success": true, "skill_count": 0, "failed_skills": []})) + .ok_or_else(|| { + format!( + "required Skills are missing from the deployment plan: {}", + required_skills.join(", ") + ) + }); + }; + + let api_base_url = request_api_base_url(&request); + let report = deploy_skills(&plan, &api_base_url).await?; + let missing_required = missing_required_skills(&required_skills, &plan, &report); + if !missing_required.is_empty() { + return Err(format!( + "required Skill deployment failed: {}", + missing_required.join(", ") + )); } + + Ok(json!({ + "success": true, + "skill_count": report.skill_count, + "success_count": report.success_skills.len(), + "success_skills": report.success_skills, + "failed_skills": report.failed_skills, + "required_skills": required_skills, + "skills_dir": plan.skills_dir, + })) +} + +fn required_skill_names(request: &ExecutionRequest) -> Vec { + let mut names = BTreeSet::new(); + for key in ["required_skills", "preload_skills"] { + if let Some(values) = request.extra.get(key).and_then(Value::as_array) { + names.extend( + values + .iter() + .filter_map(Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(ToOwned::to_owned), + ); + } + } + names.into_iter().collect() +} + +fn missing_required_skills( + required_skills: &[String], + plan: &SkillDeploymentPlan, + report: &SkillDeploymentReport, +) -> Vec { + required_skills + .iter() + .filter(|skill| { + !plan.skills.contains(skill) + || report.failed_skills.contains(skill) + || !plan.skills_dir.join(skill).join("SKILL.md").is_file() + }) + .cloned() + .collect() } struct AttachmentDownloadOutcome { @@ -681,7 +786,10 @@ fn attachment_ids(attachments: &[AttachmentRecord]) -> String { .join(",") } -async fn deploy_skills(plan: &SkillDeploymentPlan, api_base_url: &str) -> Result<(), String> { +async fn deploy_skills( + plan: &SkillDeploymentPlan, + api_base_url: &str, +) -> Result { fs::create_dir_all(&plan.skills_dir).map_err(|error| { format!( "failed to create skills dir {}: {error}", @@ -696,13 +804,26 @@ async fn deploy_skills(plan: &SkillDeploymentPlan, api_base_url: &str) -> Result async move { let target = plan.skills_dir.join(&skill); let skill_ref = plan.resolved_skill_map.get(&skill); - let Some(cache_miss_reason) = - skill_cache_miss_reason(&plan.skills_dir, &skill, skill_ref)? - else { - return Ok::(SkillDeploymentResult { + let cache_miss_reason = + match skill_cache_miss_reason(&plan.skills_dir, &skill, skill_ref) { + Ok(reason) => reason, + Err(error) => { + let mut fields = vec![("skill", skill.clone())]; + push_error_fields(&mut fields, error); + log_executor_event("skill cache validation failed", &fields); + return SkillDeploymentResult { + skill_name: skill, + success: false, + installed: None, + }; + } + }; + let Some(cache_miss_reason) = cache_miss_reason else { + return SkillDeploymentResult { + skill_name: skill.clone(), success: true, installed: None, - }); + }; }; let mut fields = vec![ ("skill", skill.clone()), @@ -726,15 +847,16 @@ async fn deploy_skills(plan: &SkillDeploymentPlan, api_base_url: &str) -> Result let _ = fs::remove_dir_all(&target); } match download_skill(client, plan, &skill, skill_ref, api_base_url).await { - Ok(result) => Ok(result), + Ok(result) => result, Err(error) => { let mut fields = vec![("skill", skill.clone())]; push_error_fields(&mut fields, error); log_executor_event("skill deployment item skipped after error", &fields); - Ok(SkillDeploymentResult { + SkillDeploymentResult { + skill_name: skill, success: false, installed: None, - }) + } } } } @@ -743,15 +865,18 @@ async fn deploy_skills(plan: &SkillDeploymentPlan, api_base_url: &str) -> Result .collect::>() .await; - let success_count = results + let success_count = results.iter().filter(|result| result.success).count(); + let success_skills = results .iter() - .filter(|result| matches!(result, Ok(SkillDeploymentResult { success: true, .. }))) - .count(); - for installed in results - .into_iter() - .filter_map(Result::ok) - .filter_map(|result| result.installed) - { + .filter(|result| result.success) + .map(|result| result.skill_name.clone()) + .collect::>(); + let failed_skills = results + .iter() + .filter(|result| !result.success) + .map(|result| result.skill_name.clone()) + .collect::>(); + for installed in results.into_iter().filter_map(|result| result.installed) { record_installed_skill( &plan.skills_dir, &installed.skill_name, @@ -769,10 +894,22 @@ async fn deploy_skills(plan: &SkillDeploymentPlan, api_base_url: &str) -> Result ("skills_dir", plan.skills_dir.display().to_string()), ], ); - Ok(()) + Ok(SkillDeploymentReport { + skill_count: plan.skills.len(), + success_skills, + failed_skills, + }) +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct SkillDeploymentReport { + skill_count: usize, + success_skills: Vec, + failed_skills: Vec, } struct SkillDeploymentResult { + skill_name: String, success: bool, installed: Option, } @@ -902,6 +1039,7 @@ async fn download_skill( resolve_skill(client, plan, skill_name, skill_ref, api_base_url).await? else { return Ok(SkillDeploymentResult { + skill_name: skill_name.to_owned(), success: false, installed: None, }); @@ -922,6 +1060,7 @@ async fn download_skill( .await?; match download { SkillArchiveResponse::NotModified => Ok(SkillDeploymentResult { + skill_name: skill_name.to_owned(), success: true, installed: None, }), @@ -939,6 +1078,7 @@ async fn download_skill( .or(content_hash), }); Ok(SkillDeploymentResult { + skill_name: skill_name.to_owned(), success: extracted, installed, }) diff --git a/executor/src/server/mod.rs b/executor/src/server/mod.rs index f262c90c4a..127142c6df 100644 --- a/executor/src/server/mod.rs +++ b/executor/src/server/mod.rs @@ -117,6 +117,7 @@ where local_model_proxy::token_route(), ) .route("/v1/attachments/sync", post(sync_attachments)) + .route("/v1/skills/sync", post(sync_skills)) .route("/filesystem/list-dir", get(list_workspace_directory)) .route("/filesystem/file", get(download_workspace_file)) .route( @@ -170,6 +171,25 @@ async fn sync_attachments(Json(request): Json) -> Result) -> Result, HttpError> { + let mut fields = task_fields(&request.task_id, &request.subtask_id); + let required_count = request + .extra + .get("required_skills") + .and_then(Value::as_array) + .map(Vec::len) + .unwrap_or_default(); + fields.push(("required_skill_count", required_count.to_string())); + log_executor_event("sandbox skill sync request received", &fields); + runtime_capabilities::sync_skills_for_request(request) + .await + .map(Json) + .map_err(|detail| HttpError { + status: StatusCode::UNPROCESSABLE_ENTITY, + detail, + }) +} + pub fn create_docker_router_from_env() -> Result { let engine = AgentProcessEngine::new(AgentCommandPlanner::from_env()); let sink = CallbackSink::new(env::var("CALLBACK_URL").unwrap_or_default())?; diff --git a/executor/tests/agent_process_engine_contract.rs b/executor/tests/agent_process_engine_contract.rs index 9b937dd632..a94ad94687 100644 --- a/executor/tests/agent_process_engine_contract.rs +++ b/executor/tests/agent_process_engine_contract.rs @@ -61,6 +61,7 @@ impl Drop for EnvGuard { #[tokio::test] async fn agent_process_engine_runs_planned_claude_command_and_parses_stream_output() { let _lock = env_lock().lock().await; + let workspace_dir = unique_dir("claude-planned-command-workspace"); let fake_claude = write_fake_executable( "fake-claude", r#"#!/bin/sh @@ -70,6 +71,7 @@ printf '%s\n' '{"type":"assistant","message":{"content":[{"type":"text","text":" let planner = AgentCommandPlanner::new(fake_claude.display().to_string(), "codex"); let engine = AgentProcessEngine::new(planner); let request = ExecutionRequest { + project_workspace_path: Some(workspace_dir.display().to_string()), prompt: json!("run"), bot: json!([{"shell_type": "ClaudeCode"}]), model_config: json!({"model": "anthropic", "model_id": "claude-sonnet-4"}), @@ -90,6 +92,7 @@ printf '%s\n' '{"type":"assistant","message":{"content":[{"type":"text","text":" #[tokio::test] async fn agent_process_engine_does_not_inject_project_space_mcp_into_claude_runs() { let _lock = env_lock().lock().await; + let workspace_dir = unique_dir("claude-no-space-mcp-workspace"); let args_dir = unique_dir("claude-no-space-mcp"); fs::create_dir_all(&args_dir).unwrap(); let args_file = args_dir.join("args.txt"); @@ -106,6 +109,7 @@ printf '%s\n' '{{"type":"assistant","message":{{"content":[{{"type":"text","text let planner = AgentCommandPlanner::new(fake_claude.display().to_string(), "codex"); let engine = AgentProcessEngine::new(planner); let request = ExecutionRequest { + project_workspace_path: Some(workspace_dir.display().to_string()), prompt: json!("run"), bot: json!([{"shell_type": "ClaudeCode"}]), model_config: json!({"model": "anthropic", "model_id": "claude-sonnet-4"}), @@ -138,6 +142,7 @@ printf '%s\n' '{{"type":"assistant","message":{{"content":[{{"type":"text","text #[tokio::test] async fn agent_process_engine_applies_claude_specific_process_timeout() { let _lock = env_lock().lock().await; + let workspace_dir = unique_dir("claude-timeout-workspace"); let _legacy_timeout = EnvGuard::remove("WEGENT_EXECUTOR_PROCESS_TIMEOUT_SECONDS"); let _timeout = EnvGuard::set("WEGENT_CLAUDE_CODE_PROCESS_TIMEOUT_SECONDS", "1"); let fake_claude = write_fake_executable( @@ -149,6 +154,7 @@ sleep 5 let planner = AgentCommandPlanner::new(fake_claude.display().to_string(), "codex"); let engine = AgentProcessEngine::new(planner); let request = ExecutionRequest { + project_workspace_path: Some(workspace_dir.display().to_string()), prompt: json!("run"), bot: json!([{"shell_type": "ClaudeCode"}]), model_config: json!({"model": "anthropic", "model_id": "claude-sonnet-4"}), diff --git a/executor/tests/agent_runtime_capabilities_contract.rs b/executor/tests/agent_runtime_capabilities_contract.rs index c6a2374ab0..1ac998572e 100644 --- a/executor/tests/agent_runtime_capabilities_contract.rs +++ b/executor/tests/agent_runtime_capabilities_contract.rs @@ -267,6 +267,7 @@ async fn claude_runtime_downloads_request_skills_before_process_start() { let _home = EnvGuard::set("HOME", &home.display().to_string()); let _workspace = EnvGuard::set("WORKSPACE_ROOT", &workspace_root.display().to_string()); let _mode = EnvGuard::set("EXECUTOR_MODE", "docker"); + let _backend = EnvGuard::set("WEGENT_BACKEND_URL", &backend_url); let _api = EnvGuard::set("TASK_API_DOMAIN", &backend_url); let engine = AgentProcessEngine::new(AgentCommandPlanner::new( fake_claude.display().to_string(), @@ -303,6 +304,70 @@ async fn claude_runtime_downloads_request_skills_before_process_start() { assert_eq!(fs::read_to_string(skill_path).unwrap(), "# Example Skill\n"); } +#[tokio::test] +async fn claude_runtime_does_not_start_when_required_skill_download_fails() { + let _lock = env_lock().await; + let home = unique_dir("claude-required-skill-failure-home"); + let workspace_root = unique_dir("claude-required-skill-failure-workspace"); + let log_path = unique_dir("claude-required-skill-failure-log").join("args.json"); + let fake_claude = write_fake_claude(&log_path); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let backend_url = format!("http://{}", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let _ = read_http_request_headers(&mut stream).await; + stream + .write_all(b"HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") + .await + .unwrap(); + }); + let _home = EnvGuard::set("HOME", &home.display().to_string()); + let _workspace = EnvGuard::set("WORKSPACE_ROOT", &workspace_root.display().to_string()); + let _mode = EnvGuard::set("EXECUTOR_MODE", "docker"); + let _backend = EnvGuard::set("WEGENT_BACKEND_URL", &backend_url); + let _api = EnvGuard::set("TASK_API_DOMAIN", &backend_url); + let engine = AgentProcessEngine::new(AgentCommandPlanner::new( + fake_claude.display().to_string(), + "codex", + )); + let request = ExecutionRequest { + task_id: "7790".to_owned(), + subtask_id: "101".to_owned(), + prompt: json!("use required skill"), + auth_token: Some("task-token".to_owned()), + bot: json!([{ + "id": 7, + "shell_type": "ClaudeCode", + "skills": ["abtest-file-analyzer"] + }]), + extra: serde_json::Map::from_iter([ + ( + "skill_refs".to_owned(), + json!({ + "abtest-file-analyzer": { + "skill_id": 237510, + "namespace": "default" + } + }), + ), + ("preload_skills".to_owned(), json!(["abtest-file-analyzer"])), + ]), + model_config: json!({"model": "anthropic", "model_id": "claude-sonnet-4"}), + ..ExecutionRequest::default() + }; + + let outcome = engine.run(request).await; + + assert_eq!( + outcome, + ExecutionOutcome::Failed { + message: "required Skill deployment failed: abtest-file-analyzer".to_owned() + } + ); + assert!(!log_path.exists()); + server.await.unwrap(); +} + #[tokio::test] async fn claude_runtime_downloads_attachments_and_rewrites_prompt_before_process_start() { let _lock = env_lock().await; diff --git a/executor/tests/http_contract.rs b/executor/tests/http_contract.rs index fd1d3d8bc2..9df7d5e199 100644 --- a/executor/tests/http_contract.rs +++ b/executor/tests/http_contract.rs @@ -4,6 +4,7 @@ use std::{ fs, + io::{Cursor, Write}, path::PathBuf, sync::{Arc, Mutex}, time::{SystemTime, UNIX_EPOCH}, @@ -66,6 +67,102 @@ async fn health_check_matches_executor_readiness_contract() { ); } +#[tokio::test] +async fn sandbox_skill_sync_installs_required_abtest_script_before_success() { + let _lock = env_lock().lock().await; + let home = unique_dir("sandbox-skill-sync-home"); + let archive = abtest_skill_zip(); + let backend_url = spawn_skill_server(archive, StatusCode::OK).await; + let _home = EnvGuard::set("HOME", &home.display().to_string()); + let _backend = EnvGuard::set("WEGENT_BACKEND_URL", &backend_url); + let _task_api = EnvGuard::set("TASK_API_DOMAIN", &backend_url); + let app = create_router(AppState::new(RecordingRunner::default())); + let payload = json!({ + "task_id": 37520834448496_i64, + "subtask_id": 1, + "type": "sandbox", + "auth_token": "task-jwt", + "team_namespace": "default", + "bot": [{ + "shell_type": "ClaudeCode", + "skills": ["abtest-file-analyzer"], + "skill_refs": { + "abtest-file-analyzer": { + "skill_id": 237510, + "namespace": "default" + } + } + }], + "skill_names": ["abtest-file-analyzer"], + "required_skills": ["abtest-file-analyzer"] + }); + + let response = app + .oneshot( + Request::builder() + .method(Method::POST) + .uri("/v1/skills/sync") + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from(payload.to_string())) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::OK); + assert!(home + .join(".claude/skills/abtest-file-analyzer/scripts/abtest_cli.py") + .is_file()); +} + +#[tokio::test] +async fn sandbox_skill_sync_rejects_missing_required_skill() { + let _lock = env_lock().lock().await; + let home = unique_dir("sandbox-skill-sync-failure-home"); + let backend_url = spawn_skill_server(Vec::new(), StatusCode::NOT_FOUND).await; + let _home = EnvGuard::set("HOME", &home.display().to_string()); + let _backend = EnvGuard::set("WEGENT_BACKEND_URL", &backend_url); + let _task_api = EnvGuard::set("TASK_API_DOMAIN", &backend_url); + let app = create_router(AppState::new(RecordingRunner::default())); + let payload = json!({ + "task_id": 37520834448496_i64, + "subtask_id": 1, + "type": "sandbox", + "auth_token": "task-jwt", + "bot": [{ + "shell_type": "ClaudeCode", + "skills": ["abtest-file-analyzer"], + "skill_refs": { + "abtest-file-analyzer": { + "skill_id": 237510, + "namespace": "default" + } + } + }], + "required_skills": ["abtest-file-analyzer"] + }); + + let response = app + .oneshot( + Request::builder() + .method(Method::POST) + .uri("/v1/skills/sync") + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from(payload.to_string())) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY); + let body: Value = + serde_json::from_slice(&response.into_body().collect().await.unwrap().to_bytes()).unwrap(); + assert!(body["detail"] + .as_str() + .unwrap() + .contains("required Skill deployment failed: abtest-file-analyzer")); +} + #[tokio::test] async fn responses_endpoint_accepts_openai_background_requests() { let runner = RecordingRunner::default(); @@ -768,6 +865,37 @@ async fn spawn_storage_server(archive: Arc>>) -> String { format!("http://{addr}") } +async fn spawn_skill_server(archive: Vec, status: StatusCode) -> String { + let app = Router::new().route( + "/api/v1/kinds/skills/237510/download", + get(move || { + let archive = archive.clone(); + async move { (status, [(header::CONTENT_TYPE, "application/zip")], archive) } + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + format!("http://{addr}") +} + +fn abtest_skill_zip() -> Vec { + let cursor = Cursor::new(Vec::new()); + let mut writer = zip::ZipWriter::new(cursor); + let options = zip::write::FileOptions::default(); + writer + .start_file("abtest-file-analyzer/SKILL.md", options) + .unwrap(); + writer.write_all(b"# ABTest file analyzer\n").unwrap(); + writer + .start_file("abtest-file-analyzer/scripts/abtest_cli.py", options) + .unwrap(); + writer.write_all(b"print('ready')\n").unwrap(); + writer.finish().unwrap().into_inner() +} + fn connect_envelope(flags: u8, data: &[u8]) -> Vec { let mut envelope = Vec::with_capacity(5 + data.len()); envelope.push(flags); diff --git a/executor_manager/routers/e2b.py b/executor_manager/routers/e2b.py index 364cc387c3..7ee207e1ab 100644 --- a/executor_manager/routers/e2b.py +++ b/executor_manager/routers/e2b.py @@ -332,8 +332,11 @@ async def create_sandbox( ) if error: - # Sandbox created but with error - logger.warning(f"[E2B API] Sandbox created with error: {error}") + logger.error(f"[E2B API] Sandbox creation failed: {error}") + raise HTTPException( + status_code=503, + detail={"code": "sandbox_not_ready", "message": error}, + ) # Get domain from request for SDK to construct URLs domain = get_domain_from_request(http_request) diff --git a/executor_manager/services/sandbox/manager.py b/executor_manager/services/sandbox/manager.py index 7a5fc0ef36..b4283ba272 100644 --- a/executor_manager/services/sandbox/manager.py +++ b/executor_manager/services/sandbox/manager.py @@ -36,6 +36,12 @@ get_container_health_checker, ) from executor_manager.services.sandbox.repository import get_sandbox_repository +from executor_manager.services.sandbox.skill_sync import ( + ResolvedTaskSkills, + SandboxSkillSyncError, + SandboxSkillSynchronizer, + required_skill_names, +) from executor_manager.utils.executor_name import generate_executor_name from shared.logger import setup_logger from shared.telemetry.decorators import trace_async @@ -69,6 +75,7 @@ def __init__(self): self._repository = get_sandbox_repository() self._health_checker = get_container_health_checker() self._execution_runner = get_execution_runner() + self._skill_synchronizer = SandboxSkillSynchronizer() self._scheduler: Optional["SandboxScheduler"] = None self._shutting_down = False self._create_locks: Dict[str, asyncio.Lock] = {} @@ -173,6 +180,16 @@ async def _create_sandbox_locked( f"[SandboxManager] Reusing existing sandbox {existing_sandbox.sandbox_id} " f"for task {task_id} (health check passed)" ) + self._merge_activation_metadata( + existing_sandbox, sandbox_metadata, bot_config + ) + skill_error = await self._prepare_sandbox_skills( + existing_sandbox, existing_sandbox.base_url + ) + if skill_error: + existing_sandbox.set_failed(skill_error) + self._repository.save_sandbox(existing_sandbox) + return existing_sandbox, skill_error # Extend timeout existing_sandbox.extend_timeout(timeout) self._repository.save_sandbox(existing_sandbox) @@ -240,8 +257,13 @@ async def _start_sandbox_container(self, sandbox: Sandbox) -> Optional[str]: Returns: Error message if failed, None if successful """ + try: + resolved_skills = await self._skill_synchronizer.resolve(sandbox) + except SandboxSkillSyncError as exc: + return str(exc) + # Build task data for executor - task_data = self._build_sandbox_task(sandbox) + task_data = self._build_sandbox_task(sandbox, resolved_skills) # Get executor and create container executor = ExecutorDispatcher.get_executor(EXECUTOR_DISPATCHER_MODE) @@ -271,11 +293,54 @@ async def _start_sandbox_container(self, sandbox: Sandbox) -> Optional[str]: if base_url is None: return f"Container {container_name} failed to become ready" + try: + await self._skill_synchronizer.sync(base_url, task_data, resolved_skills) + except SandboxSkillSyncError as exc: + return str(exc) + sandbox.set_running(base_url) + sandbox.metadata["synced_required_skills"] = resolved_skills.required_skills self._repository.save_sandbox(sandbox) return None + async def _prepare_sandbox_skills( + self, sandbox: Sandbox, base_url: str + ) -> Optional[str]: + """Synchronize newly active Skills before reusing a running sandbox.""" + try: + resolved = await self._skill_synchronizer.resolve(sandbox) + task = self._build_sandbox_task(sandbox, resolved) + await self._skill_synchronizer.sync(base_url, task, resolved) + except SandboxSkillSyncError as exc: + return str(exc) + sandbox.metadata["synced_required_skills"] = resolved.required_skills + return None + + @staticmethod + def _merge_activation_metadata( + sandbox: Sandbox, + incoming: Dict[str, Any], + bot_config: Optional[Dict[str, Any]], + ) -> None: + """Merge credentials and active Skill names into a reused sandbox.""" + for key in ( + "auth_token", + "skill_identity_token", + "workspace_ref", + "task_type", + "e2b_sandbox_id", + "bot_config", + ): + if incoming.get(key): + sandbox.metadata[key] = incoming[key] + + required = set(required_skill_names(sandbox.metadata)) + required.update(required_skill_names(incoming)) + sandbox.metadata["required_skills"] = sorted(required) + if bot_config: + sandbox.metadata["bot_config"] = bot_config + async def _wait_for_container_ready( self, executor, @@ -339,7 +404,11 @@ async def _check_container_health(self, base_url: str) -> bool: except Exception: return False - def _build_sandbox_task(self, sandbox: Sandbox) -> Dict[str, Any]: + def _build_sandbox_task( + self, + sandbox: Sandbox, + resolved_skills: Optional[ResolvedTaskSkills] = None, + ) -> Dict[str, Any]: """Build task data for creating a sandbox container. Args: @@ -419,6 +488,9 @@ def _build_sandbox_task(self, sandbox: Sandbox) -> Dict[str, Any]: if skill_identity_token: task["skill_identity_token"] = skill_identity_token + if resolved_skills is not None: + resolved_skills.apply_to_task(task) + return task async def get_sandbox( diff --git a/executor_manager/services/sandbox/skill_sync.py b/executor_manager/services/sandbox/skill_sync.py new file mode 100644 index 0000000000..9efa771670 --- /dev/null +++ b/executor_manager/services/sandbox/skill_sync.py @@ -0,0 +1,200 @@ +# SPDX-FileCopyrightText: 2026 Weibo, Inc. +# +# SPDX-License-Identifier: Apache-2.0 + +"""Resolve and synchronize task Skills before a sandbox becomes usable.""" + +import json +import os +from dataclasses import dataclass, field +from typing import Any, Dict + +import httpx + +from executor_manager.config.config import TASK_API_DOMAIN +from executor_manager.models.sandbox import Sandbox +from shared.logger import setup_logger +from shared.utils.http_client import traced_async_client + +logger = setup_logger(__name__) + +TASK_SKILLS_TIMEOUT = float(os.getenv("SANDBOX_TASK_SKILLS_TIMEOUT", "30")) +SKILL_SYNC_TIMEOUT = float(os.getenv("SANDBOX_SKILL_SYNC_TIMEOUT", "180")) + + +class SandboxSkillSyncError(RuntimeError): + """Raised when task Skills cannot be prepared for a sandbox.""" + + +@dataclass(frozen=True) +class ResolvedTaskSkills: + """Authoritative task Skill configuration returned by Backend.""" + + team_namespace: str = "default" + skills: list[str] = field(default_factory=list) + preload_skills: list[str] = field(default_factory=list) + skill_refs: Dict[str, Any] = field(default_factory=dict) + preload_skill_refs: Dict[str, Any] = field(default_factory=dict) + required_skills: list[str] = field(default_factory=list) + + @property + def needs_sync(self) -> bool: + """Return whether the executor must receive a Skill sync request.""" + return bool(self.skills or self.preload_skills or self.required_skills) + + def apply_to_task(self, task: Dict[str, Any]) -> None: + """Populate an executor task with resolved Skill fields.""" + bot = task["bot"][0] + bot["skills"] = list(self.skills) + bot["skill_refs"] = dict(self.skill_refs) + bot["preload_skill_refs"] = dict(self.preload_skill_refs) + task.update( + { + "backend_url": TASK_API_DOMAIN, + "team_namespace": self.team_namespace, + "skill_names": list(self.skills), + "preload_skills": list(self.preload_skills), + "skill_refs": dict(self.skill_refs), + "preload_skill_refs": dict(self.preload_skill_refs), + "required_skills": list(self.required_skills), + } + ) + + +def required_skill_names(metadata: Dict[str, Any]) -> list[str]: + """Parse active Skill names from E2B string metadata or native lists.""" + raw = metadata.get("required_skills", []) + if isinstance(raw, str): + try: + raw = json.loads(raw) + except json.JSONDecodeError as exc: + raise SandboxSkillSyncError("required_skills metadata is invalid") from exc + if not isinstance(raw, list): + raise SandboxSkillSyncError("required_skills metadata must be a list") + return sorted( + {item.strip() for item in raw if isinstance(item, str) and item.strip()} + ) + + +class SandboxSkillSynchronizer: + """Fetch task Skills and require executor confirmation before sandbox use.""" + + async def resolve(self, sandbox: Sandbox) -> ResolvedTaskSkills: + """Resolve the task's Skills with its task-scoped authorization token.""" + required = required_skill_names(sandbox.metadata) + auth_token = str(sandbox.metadata.get("auth_token") or "").strip() + if not auth_token: + if required: + raise SandboxSkillSyncError( + "Cannot prepare required sandbox Skills: auth_token is missing" + ) + return ResolvedTaskSkills() + + task_id = sandbox.metadata.get("task_id") + if task_id is None: + raise SandboxSkillSyncError( + "Cannot resolve sandbox Skills: task_id is missing" + ) + url = f"{TASK_API_DOMAIN.rstrip('/')}/api/v1/tasks/{task_id}/skills" + try: + async with traced_async_client(timeout=TASK_SKILLS_TIMEOUT) as client: + response = await client.get( + url, headers={"Authorization": f"Bearer {auth_token}"} + ) + except httpx.HTTPError as exc: + raise SandboxSkillSyncError( + f"Failed to resolve task Skills: {exc}" + ) from exc + + if response.status_code != 200: + body = response.text[:300] + raise SandboxSkillSyncError( + "Failed to resolve task Skills: " + f"HTTP {response.status_code}; body={body}" + ) + try: + payload = response.json() + except ValueError as exc: + raise SandboxSkillSyncError( + "Failed to resolve task Skills: response is not valid JSON" + ) from exc + if not isinstance(payload, dict): + raise SandboxSkillSyncError( + "Failed to resolve task Skills: response must be an object" + ) + return self._parse_response(payload, required) + + async def sync( + self, + base_url: str, + task: Dict[str, Any], + resolved: ResolvedTaskSkills, + ) -> Dict[str, Any]: + """Ask the executor to deploy and validate Skills synchronously.""" + if not resolved.needs_sync: + return {"success": True, "skipped": True} + + url = f"{base_url.rstrip('/')}/v1/skills/sync" + try: + async with traced_async_client(timeout=SKILL_SYNC_TIMEOUT) as client: + response = await client.post(url, json=task) + except httpx.HTTPError as exc: + raise SandboxSkillSyncError( + f"Sandbox Skill deployment request failed: {exc}" + ) from exc + + if response.status_code != 200: + body = response.text[:500] + raise SandboxSkillSyncError( + "Sandbox Skill deployment failed: " + f"HTTP {response.status_code}; body={body}" + ) + try: + result = response.json() + except ValueError as exc: + raise SandboxSkillSyncError( + "Sandbox Skill deployment returned invalid JSON" + ) from exc + if not isinstance(result, dict): + raise SandboxSkillSyncError( + "Sandbox Skill deployment response must be an object" + ) + if not result.get("success"): + raise SandboxSkillSyncError(f"Sandbox Skill deployment failed: {result}") + logger.info( + "[SandboxSkillSync] Skills ready task_id=%s required=%s failed_optional=%s", + task.get("task_id"), + resolved.required_skills, + result.get("failed_skills", []), + ) + return result + + @staticmethod + def _parse_response( + payload: Dict[str, Any], required: list[str] + ) -> ResolvedTaskSkills: + """Validate and normalize the Backend response.""" + skills = _string_list(payload.get("skills")) + missing = sorted(set(required) - set(skills)) + if missing: + raise SandboxSkillSyncError( + f"Required task Skills are unavailable: {', '.join(missing)}" + ) + return ResolvedTaskSkills( + team_namespace=str(payload.get("team_namespace") or "default"), + skills=skills, + preload_skills=_string_list(payload.get("preload_skills")), + skill_refs=_object(payload.get("skill_refs")), + preload_skill_refs=_object(payload.get("preload_skill_refs")), + required_skills=required, + ) + + +def _string_list(value: Any) -> list[str]: + if not isinstance(value, list): + return [] + return sorted({item for item in value if isinstance(item, str) and item}) + + +def _object(value: Any) -> Dict[str, Any]: + return dict(value) if isinstance(value, dict) else {} diff --git a/executor_manager/tests/routers/test_e2b_sandbox_creation.py b/executor_manager/tests/routers/test_e2b_sandbox_creation.py new file mode 100644 index 0000000000..79ff9d3b86 --- /dev/null +++ b/executor_manager/tests/routers/test_e2b_sandbox_creation.py @@ -0,0 +1,45 @@ +# SPDX-FileCopyrightText: 2026 Weibo, Inc. +# +# SPDX-License-Identifier: Apache-2.0 + +"""E2B sandbox creation error propagation tests.""" + +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from fastapi import HTTPException + +from executor_manager.routers import e2b + + +@pytest.mark.asyncio +async def test_create_sandbox_surfaces_skill_readiness_failure(sample_sandbox, mocker): + """The SDK must receive an explicit error when required Skills are missing.""" + manager = SimpleNamespace( + create_sandbox=AsyncMock( + return_value=( + sample_sandbox, + "required Skill deployment failed: abtest-file-analyzer", + ) + ) + ) + mocker.patch.object(e2b, "get_sandbox_manager", return_value=manager) + http_request = SimpleNamespace( + client=SimpleNamespace(host="127.0.0.1"), + url=SimpleNamespace(scheme="http"), + headers={"host": "localhost"}, + ) + request = e2b.CreateSandboxRequest( + templateId="ClaudeCode", + metadata={"task_id": "12345"}, + ) + + with pytest.raises(HTTPException) as exc_info: + await e2b.create_sandbox(request, http_request) + + assert exc_info.value.status_code == 503 + assert exc_info.value.detail == { + "code": "sandbox_not_ready", + "message": "required Skill deployment failed: abtest-file-analyzer", + } diff --git a/executor_manager/tests/services/test_sandbox_manager.py b/executor_manager/tests/services/test_sandbox_manager.py index a415269aed..da089260a7 100644 --- a/executor_manager/tests/services/test_sandbox_manager.py +++ b/executor_manager/tests/services/test_sandbox_manager.py @@ -615,6 +615,150 @@ def test_build_sandbox_task_propagates_skip_git_clone( assert task["skip_git_clone"] is True + def test_build_sandbox_task_populates_resolved_skills( + self, sandbox_manager_with_mock_redis, sample_sandbox + ): + """Sandbox task should carry every field required by Skill deployment.""" + from executor_manager.services.sandbox.skill_sync import ResolvedTaskSkills + + manager = sandbox_manager_with_mock_redis + resolved = ResolvedTaskSkills( + team_namespace="team-a", + skills=["abtest-file-analyzer", "sandbox"], + preload_skills=["sandbox"], + skill_refs={ + "abtest-file-analyzer": { + "skill_id": 237510, + "namespace": "default", + } + }, + preload_skill_refs={"sandbox": {"skill_id": 1, "namespace": "default"}}, + required_skills=["abtest-file-analyzer"], + ) + + task = manager._build_sandbox_task(sample_sandbox, resolved) + + assert task["team_namespace"] == "team-a" + assert task["skill_names"] == ["abtest-file-analyzer", "sandbox"] + assert task["preload_skills"] == ["sandbox"] + assert task["required_skills"] == ["abtest-file-analyzer"] + assert task["bot"][0]["skills"] == ["abtest-file-analyzer", "sandbox"] + assert task["bot"][0]["skill_refs"] == resolved.skill_refs + + @pytest.mark.asyncio + async def test_cold_sandbox_waits_for_required_skills_before_running( + self, sandbox_manager_with_mock_redis, sample_sandbox, mocker + ): + """A cold sandbox must stay pending until the executor confirms Skills.""" + from executor_manager.models.sandbox import SandboxStatus + from executor_manager.services.sandbox.skill_sync import ResolvedTaskSkills + + manager = sandbox_manager_with_mock_redis + sample_sandbox.status = SandboxStatus.PENDING + sample_sandbox.base_url = None + sample_sandbox.metadata["auth_token"] = "task-jwt" + resolved = ResolvedTaskSkills( + skills=["abtest-file-analyzer"], + required_skills=["abtest-file-analyzer"], + ) + mocker.patch.object( + manager._skill_synchronizer, + "resolve", + new_callable=AsyncMock, + return_value=resolved, + ) + + async def sync_before_running(base_url, task, task_skills): + assert sample_sandbox.status == SandboxStatus.PENDING + assert task["required_skills"] == ["abtest-file-analyzer"] + assert task_skills is resolved + return {"success": True} + + sync = mocker.patch.object( + manager._skill_synchronizer, + "sync", + new_callable=AsyncMock, + side_effect=sync_before_running, + ) + executor = mocker.MagicMock() + executor.submit_executor.return_value = { + "status": "success", + "executor_name": "cold-sandbox", + } + mocker.patch( + "executor_manager.services.sandbox.manager.ExecutorDispatcher.get_executor", + return_value=executor, + ) + mocker.patch.object( + manager, + "_wait_for_container_ready", + new_callable=AsyncMock, + return_value="http://sandbox:8080", + ) + + error = await manager._start_sandbox_container(sample_sandbox) + + assert error is None + assert sample_sandbox.status == SandboxStatus.RUNNING + sync.assert_awaited_once() + + @pytest.mark.asyncio + async def test_reused_warm_sandbox_syncs_newly_loaded_skill( + self, + sandbox_manager_with_mock_redis, + mock_redis_client, + mocker, + sample_sandbox_redis_data, + ): + """A reused warm sandbox must sync Skills loaded after its first use.""" + from executor_manager.services.sandbox.skill_sync import ResolvedTaskSkills + + manager = sandbox_manager_with_mock_redis + mock_redis_client.hget.return_value = sample_sandbox_redis_data + mocker.patch.object( + manager._health_checker, "check_health_sync", return_value=True + ) + ensure_workspace = mocker.patch.object( + manager, + "_ensure_sandbox_workspace", + new_callable=AsyncMock, + return_value=None, + ) + resolved = ResolvedTaskSkills( + skills=["abtest-file-analyzer"], + required_skills=["abtest-file-analyzer"], + ) + mocker.patch.object( + manager._skill_synchronizer, + "resolve", + new_callable=AsyncMock, + return_value=resolved, + ) + sync = mocker.patch.object( + manager._skill_synchronizer, + "sync", + new_callable=AsyncMock, + return_value={"success": True}, + ) + start = mocker.patch.object(manager, "_start_sandbox_container") + + sandbox, error = await manager.create_sandbox( + shell_type="ClaudeCode", + user_id=100, + user_name="testuser", + metadata={ + "task_id": 12345, + "auth_token": "task-jwt", + "required_skills": '["abtest-file-analyzer"]', + }, + ) + + assert error is None + assert sandbox.metadata["required_skills"] == ["abtest-file-analyzer"] + ensure_workspace.assert_awaited_once_with(sandbox) + sync.assert_awaited_once() + start.assert_not_called() + # ----- get_sandbox Tests ----- @pytest.mark.asyncio diff --git a/executor_manager/tests/services/test_sandbox_skill_sync.py b/executor_manager/tests/services/test_sandbox_skill_sync.py new file mode 100644 index 0000000000..3a86440dfa --- /dev/null +++ b/executor_manager/tests/services/test_sandbox_skill_sync.py @@ -0,0 +1,135 @@ +# SPDX-FileCopyrightText: 2026 Weibo, Inc. +# +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for sandbox task Skill resolution and deployment gating.""" + +import json + +import httpx +import pytest + +from executor_manager.models.sandbox import Sandbox +from executor_manager.services.sandbox.skill_sync import ( + ResolvedTaskSkills, + SandboxSkillSyncError, + SandboxSkillSynchronizer, +) + + +def _sandbox(metadata): + return Sandbox.create( + shell_type="ClaudeCode", + user_id=7, + user_name="alice", + timeout=600, + metadata={"task_id": 123, **metadata}, + ) + + +def _mock_client(mocker, handler): + transport = httpx.MockTransport(handler) + mocker.patch( + "executor_manager.services.sandbox.skill_sync.traced_async_client", + side_effect=lambda **kwargs: httpx.AsyncClient(transport=transport, **kwargs), + ) + + +@pytest.mark.asyncio +async def test_resolve_fetches_authoritative_task_skills(mocker): + """Resolution should forward auth and preserve all Skill reference fields.""" + + def handler(request): + assert request.url.path == "/api/v1/tasks/123/skills" + assert request.headers["Authorization"] == "Bearer task-jwt" + return httpx.Response( + 200, + json={ + "team_namespace": "team-a", + "skills": ["sandbox", "abtest-file-analyzer"], + "preload_skills": ["sandbox"], + "skill_refs": { + "abtest-file-analyzer": { + "skill_id": 237510, + "namespace": "default", + } + }, + "preload_skill_refs": { + "sandbox": {"skill_id": 1, "namespace": "default"} + }, + }, + ) + + _mock_client(mocker, handler) + sandbox = _sandbox( + { + "auth_token": "task-jwt", + "required_skills": json.dumps(["abtest-file-analyzer"]), + } + ) + + resolved = await SandboxSkillSynchronizer().resolve(sandbox) + + assert resolved.team_namespace == "team-a" + assert resolved.required_skills == ["abtest-file-analyzer"] + assert resolved.skill_refs["abtest-file-analyzer"]["skill_id"] == 237510 + + +@pytest.mark.asyncio +async def test_resolve_rejects_required_skill_missing_from_task(mocker): + """A loaded Skill missing from Backend resolution must block activation.""" + _mock_client( + mocker, + lambda request: httpx.Response(200, json={"skills": ["sandbox"]}), + ) + sandbox = _sandbox( + { + "auth_token": "task-jwt", + "required_skills": ["abtest-file-analyzer"], + } + ) + + with pytest.raises( + SandboxSkillSyncError, + match="Required task Skills are unavailable: abtest-file-analyzer", + ): + await SandboxSkillSynchronizer().resolve(sandbox) + + +@pytest.mark.asyncio +async def test_resolve_requires_task_id_when_auth_token_is_present(): + """Task-scoped Skill resolution must never call a synthetic None task URL.""" + sandbox = _sandbox({"auth_token": "task-jwt"}) + sandbox.metadata.pop("task_id") + + with pytest.raises( + SandboxSkillSyncError, + match="Cannot resolve sandbox Skills: task_id is missing", + ): + await SandboxSkillSynchronizer().resolve(sandbox) + + +@pytest.mark.asyncio +async def test_sync_surfaces_executor_required_skill_failure(mocker): + """Executor validation failures should become explicit manager errors.""" + _mock_client( + mocker, + lambda request: httpx.Response( + 422, + json={"detail": "required Skill deployment failed: abtest-file-analyzer"}, + ), + ) + resolved = ResolvedTaskSkills( + skills=["abtest-file-analyzer"], + required_skills=["abtest-file-analyzer"], + ) + + with pytest.raises( + SandboxSkillSyncError, + match="Sandbox Skill deployment failed: HTTP 422", + ): + await SandboxSkillSynchronizer().sync( + "http://sandbox:8080", + {"task_id": 123}, + resolved, + ) diff --git a/shared/tests/utils/test_attachment_block.py b/shared/tests/utils/test_attachment_block.py index a1eccdc713..c71d0e7bdd 100644 --- a/shared/tests/utils/test_attachment_block.py +++ b/shared/tests/utils/test_attachment_block.py @@ -9,10 +9,17 @@ build_sandbox_path, build_truncation_note, format_file_size, + sanitize_attachment_filename, truncate_for_injection, ) +def test_sanitize_attachment_filename_removes_path_traversal() -> None: + assert sanitize_attachment_filename("../../etc/passwd") == "passwd" + assert sanitize_attachment_filename(r"C:\\temp\\report.csv") == "report.csv" + assert sanitize_attachment_filename("..") == "document" + + def test_truncate_for_injection_short_text_unchanged(): text, truncated = truncate_for_injection("short", 64000) assert text == "short" diff --git a/shared/utils/attachment_block.py b/shared/utils/attachment_block.py index c4443ab157..fe7cb9305c 100644 --- a/shared/utils/attachment_block.py +++ b/shared/utils/attachment_block.py @@ -20,6 +20,19 @@ from __future__ import annotations +def sanitize_attachment_filename( + filename: str | None, *, fallback: str = "document" +) -> str: + """Return a single safe path component for an attachment filename.""" + candidate = (filename or fallback).replace("\\", "/").rsplit("/", 1)[-1] + candidate = "".join( + character + for character in candidate + if ord(character) >= 32 and ord(character) != 127 + ) + return candidate if candidate not in {"", ".", ".."} else fallback + + def format_file_size(size_bytes: int) -> str: """Format a byte count into a human-readable string (B / KB / MB).""" if size_bytes >= 1024 * 1024: @@ -41,12 +54,12 @@ def build_sandbox_path( ) -> str | None: """Build the sandbox file path where the Executor downloads an attachment. - Returns ``None`` when *task_id* or *subtask_id* is missing. Control - characters in *filename* are stripped to keep the path single-line. + Returns ``None`` when *task_id* or *subtask_id* is missing. Path components + and control characters in *filename* are stripped. """ if task_id is None or subtask_id is None: return None - safe_filename = (filename or "document").replace("\n", "").replace("\r", "") + safe_filename = sanitize_attachment_filename(filename) return f"/home/user/{task_id}:executor:attachments/{subtask_id}/{safe_filename}" @@ -122,9 +135,9 @@ def build_attachment_header( label = "Image Attachment" if is_image else "Attachment" formatted_size = format_file_size(file_size or 0) url = build_attachment_download_url(attachment_id) - # Strip control chars so a crafted filename can't break the single-line - # header or inject extra prompt content (mirrors build_sandbox_path). - safe_filename = (filename or "document").replace("\n", "").replace("\r", "") + # Sanitize so a crafted filename cannot inject prompt content or advertise + # a path outside the attachment directory (mirrors build_sandbox_path). + safe_filename = sanitize_attachment_filename(filename) parts = [ f"[{label}: {safe_filename} | ID: {attachment_id} | " f"Type: {mime_type or 'unknown'} | Size: {formatted_size} | URL: {url}"