diff --git a/src/xagent/core/path_locks.py b/src/xagent/core/path_locks.py new file mode 100644 index 000000000..f93b5ad32 --- /dev/null +++ b/src/xagent/core/path_locks.py @@ -0,0 +1,69 @@ +"""Shared normalized-path mutation locks.""" + +from __future__ import annotations + +import threading +from contextlib import ExitStack, contextmanager +from pathlib import Path +from typing import Iterator + + +class PathMutationLockRegistry: + """Provide re-entrant locks keyed by normalized absolute filesystem path.""" + + def __init__(self) -> None: + self._guard = threading.Lock() + self._locks: dict[str, tuple[threading.RLock, int]] = {} + + @staticmethod + def normalize_path(path: str | Path) -> Path: + return Path(path).expanduser().resolve() + + def _acquire_lock_for_key(self, key: str) -> threading.RLock: + with self._guard: + lock_entry = self._locks.get(key) + if lock_entry is None: + lock = threading.RLock() + self._locks[key] = (lock, 1) + return lock + + lock, ref_count = lock_entry + self._locks[key] = (lock, ref_count + 1) + return lock + + def _release_lock_for_key(self, key: str) -> None: + with self._guard: + lock_entry = self._locks.get(key) + if lock_entry is None: + return + + lock, ref_count = lock_entry + if ref_count <= 1: + self._locks.pop(key, None) + else: + self._locks[key] = (lock, ref_count - 1) + + @contextmanager + def guard_path(self, path: str | Path) -> Iterator[Path]: + normalized_path = self.normalize_path(path) + key = str(normalized_path) + lock = self._acquire_lock_for_key(key) + try: + with lock: + yield normalized_path + finally: + self._release_lock_for_key(key) + + @contextmanager + def guard_paths(self, paths: list[str | Path]) -> Iterator[tuple[Path, ...]]: + normalized_paths = tuple(self.normalize_path(path) for path in paths) + unique_paths = {str(path): path for path in normalized_paths} + ordered_paths = [unique_paths[key] for key in sorted(unique_paths)] + + with ExitStack() as stack: + for normalized_path in ordered_paths: + stack.enter_context(self.guard_path(normalized_path)) + yield normalized_paths + + +GLOBAL_PATH_MUTATION_LOCKS = PathMutationLockRegistry() diff --git a/src/xagent/core/tools/adapters/vibe/browser_use.py b/src/xagent/core/tools/adapters/vibe/browser_use.py index c1102f682..d0317c20a 100644 --- a/src/xagent/core/tools/adapters/vibe/browser_use.py +++ b/src/xagent/core/tools/adapters/vibe/browser_use.py @@ -7,6 +7,7 @@ import logging import os +import uuid from typing import Any, Mapping, Optional, Type from pydantic import BaseModel, Field @@ -34,6 +35,19 @@ _STEP_SESSION_ARG = "_xagent_step_id" +def _browser_output_filename( + output_filename: Any, *, prefix: str, extension: str +) -> str: + if output_filename: + return os.path.basename(str(output_filename)) + + from datetime import datetime + + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f") + unique_suffix = uuid.uuid4().hex[:8] + return f"{prefix}_{timestamp}_{unique_suffix}.{extension}" + + class BrowserTaskSessionMixin: """Keeps browser tool default sessions aligned during task setup.""" @@ -324,6 +338,7 @@ class BrowserPdfResult(BaseModel): class BrowserNavigateTool(BrowserTaskSessionMixin, AbstractBaseTool): category = ToolCategory.BROWSER + concurrency_safe = True """Navigate to a URL in a browser session.""" def __init__( @@ -491,6 +506,7 @@ async def teardown(self, task_id: Optional[str] = None) -> None: class BrowserClickTool(BrowserTaskSessionMixin, AbstractBaseTool): category = ToolCategory.BROWSER + concurrency_safe = True """Click an element on the current page.""" def __init__(self, task_id: Optional[str] = None): @@ -539,6 +555,7 @@ async def run_json_async(self, args: Mapping[str, Any]) -> Any: class BrowserFillTool(BrowserTaskSessionMixin, AbstractBaseTool): category = ToolCategory.BROWSER + concurrency_safe = True """Fill an input field with text.""" def __init__(self, task_id: Optional[str] = None): @@ -587,6 +604,7 @@ async def run_json_async(self, args: Mapping[str, Any]) -> Any: class BrowserScreenshotTool(BrowserTaskSessionMixin, AbstractBaseTool): category = ToolCategory.BROWSER + concurrency_safe = True """Take a screenshot of the current page.""" def __init__( @@ -650,7 +668,6 @@ async def run_json_async(self, args: Mapping[str, Any]) -> Any: if self._workspace and result.get("success"): try: import base64 - from datetime import datetime # Extract base64 data from data URI screenshot_data = result.get("screenshot", "") @@ -663,29 +680,30 @@ async def run_json_async(self, args: Mapping[str, Any]) -> Any: image_bytes = base64.b64decode(base64_data) # Determine filename - always save to output directory - output_filename = args.get("output_filename") - if output_filename: - # Sanitize filename to prevent path traversal attacks - filename = os.path.basename(output_filename) - else: - timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") - filename = f"screenshot_{timestamp}.png" + filename = _browser_output_filename( + args.get("output_filename"), prefix="screenshot", extension="png" + ) # Always save to output directory file_path = self._workspace.output_dir / filename - # Save to file within auto_register context - with self._workspace.auto_register_files(): - with open(file_path, "wb") as f: - f.write(image_bytes) - - relative_path = str( - file_path.relative_to(self._workspace.workspace_dir) - ) - file_ref = build_workspace_file_ref( - workspace=self._workspace, - file_path=file_path, - mime_type="image/png", - ) + # Save and register under the shared path guard so concurrent + # explicit filenames cannot interleave writes or metadata. + with self._workspace.guard_workspace_mutation_path( + file_path + ) as guarded_path: + file_path = guarded_path + with self._workspace.auto_register_files(): + with open(file_path, "wb") as f: + f.write(image_bytes) + + relative_path = str( + file_path.relative_to(self._workspace.workspace_dir) + ) + file_ref = build_workspace_file_ref( + workspace=self._workspace, + file_path=file_path, + mime_type="image/png", + ) result["screenshot"] = relative_path result["format"] = "file" result["file_id"] = file_ref["file_id"] @@ -707,6 +725,7 @@ async def run_json_async(self, args: Mapping[str, Any]) -> Any: class BrowserExtractTextTool(BrowserTaskSessionMixin, AbstractBaseTool): category = ToolCategory.BROWSER + concurrency_safe = True """Extract text content from the page.""" def __init__(self, task_id: Optional[str] = None): @@ -754,6 +773,7 @@ async def run_json_async(self, args: Mapping[str, Any]) -> Any: class BrowserEvaluateTool(BrowserTaskSessionMixin, AbstractBaseTool): category = ToolCategory.BROWSER + concurrency_safe = True """Execute JavaScript code in the browser.""" def __init__(self, task_id: Optional[str] = None): @@ -861,6 +881,7 @@ async def run_json_async(self, args: Mapping[str, Any]) -> Any: class BrowserSelectOptionTool(BrowserTaskSessionMixin, AbstractBaseTool): category = ToolCategory.BROWSER + concurrency_safe = True """Select an option from a dropdown.""" def __init__(self, task_id: Optional[str] = None): @@ -910,6 +931,7 @@ async def run_json_async(self, args: Mapping[str, Any]) -> Any: class BrowserWaitForSelectorTool(BrowserTaskSessionMixin, AbstractBaseTool): category = ToolCategory.BROWSER + concurrency_safe = True """Wait for an element to appear on the page.""" def __init__(self, task_id: Optional[str] = None): @@ -1001,6 +1023,7 @@ async def run_json_async(self, args: Mapping[str, Any]) -> Any: class BrowserPdfTool(BrowserTaskSessionMixin, AbstractBaseTool): category = ToolCategory.BROWSER + concurrency_safe = True """Save current page as PDF.""" def __init__( @@ -1060,7 +1083,6 @@ async def run_json_async(self, args: Mapping[str, Any]) -> Any: if self._workspace and result.get("success"): try: import base64 - from datetime import datetime # Extract base64 data pdf_data = result.get("pdf", "") @@ -1073,30 +1095,31 @@ async def run_json_async(self, args: Mapping[str, Any]) -> Any: pdf_bytes = base64.b64decode(base64_data) # Determine filename - always save to output directory - output_filename = args.get("output_filename") - if output_filename: - # Sanitize filename to prevent path traversal attacks - filename = os.path.basename(output_filename) - else: - timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") - filename = f"page_{timestamp}.pdf" + filename = _browser_output_filename( + args.get("output_filename"), prefix="page", extension="pdf" + ) # Always save to output directory file_path = self._workspace.output_dir / filename - # Save to file within auto_register context - with self._workspace.auto_register_files(): - with open(file_path, "wb") as f: - f.write(pdf_bytes) - - relative_path = str( - file_path.relative_to(self._workspace.workspace_dir) - ) - file_ref = build_workspace_file_ref( - workspace=self._workspace, - file_path=file_path, - mime_type="application/pdf", - ) + # Save and register under the shared path guard so concurrent + # explicit filenames cannot interleave writes or metadata. + with self._workspace.guard_workspace_mutation_path( + file_path + ) as guarded_path: + file_path = guarded_path + with self._workspace.auto_register_files(): + with open(file_path, "wb") as f: + f.write(pdf_bytes) + + relative_path = str( + file_path.relative_to(self._workspace.workspace_dir) + ) + file_ref = build_workspace_file_ref( + workspace=self._workspace, + file_path=file_path, + mime_type="application/pdf", + ) result["output_path"] = relative_path result["format"] = "file" result["file_id"] = file_ref["file_id"] diff --git a/src/xagent/core/tools/adapters/vibe/document_parser.py b/src/xagent/core/tools/adapters/vibe/document_parser.py index 9bc0f896a..804606697 100644 --- a/src/xagent/core/tools/adapters/vibe/document_parser.py +++ b/src/xagent/core/tools/adapters/vibe/document_parser.py @@ -19,6 +19,8 @@ class DocumentParseTool(AbstractBaseTool): + read_only = True + def __init__(self, workspace: TaskWorkspace | None = None) -> None: self._visibility = ToolVisibility.PUBLIC self.workspace = workspace @@ -60,6 +62,8 @@ async def run_json_async(self, args: Mapping[str, Any]) -> Any: # This alternative tool will write the result to a file class DocumentParseWithOutputTool(AbstractBaseTool): + concurrency_safe = True + def __init__(self, workspace: TaskWorkspace | None = None) -> None: self._visibility = ToolVisibility.PUBLIC self.workspace = workspace diff --git a/src/xagent/core/tools/adapters/vibe/file_tool.py b/src/xagent/core/tools/adapters/vibe/file_tool.py index 80b243348..33b065715 100644 --- a/src/xagent/core/tools/adapters/vibe/file_tool.py +++ b/src/xagent/core/tools/adapters/vibe/file_tool.py @@ -45,12 +45,23 @@ class FileTool(FunctionTool): read_only=True, ) write_file_tool = FileTool( - write_file, name="write_file", description="Write content to file" + write_file, + name="write_file", + description="Write content to file", + concurrency_safe=True, ) append_file_tool = FileTool( - append_file, name="append_file", description="Append content to file" + append_file, + name="append_file", + description="Append content to file", + concurrency_safe=True, +) +delete_file_tool = FileTool( + delete_file, + name="delete_file", + description="Delete file", + concurrency_safe=True, ) -delete_file_tool = FileTool(delete_file, name="delete_file", description="Delete file") list_files_tool = FileTool( list_files, name="list_files", @@ -58,7 +69,10 @@ class FileTool(FunctionTool): read_only=True, ) create_directory_tool = FileTool( - create_directory, name="create_directory", description="Create directory" + create_directory, + name="create_directory", + description="Create directory", + concurrency_safe=True, ) file_exists_tool = FileTool( file_exists, @@ -79,7 +93,10 @@ class FileTool(FunctionTool): read_only=True, ) write_json_file_tool = FileTool( - write_json_file, name="write_json_file", description="Write JSON file" + write_json_file, + name="write_json_file", + description="Write JSON file", + concurrency_safe=True, ) read_csv_file_tool = FileTool( read_csv_file, @@ -88,17 +105,22 @@ class FileTool(FunctionTool): read_only=True, ) write_csv_file_tool = FileTool( - write_csv_file, name="write_csv_file", description="Write CSV file" + write_csv_file, + name="write_csv_file", + description="Write CSV file", + concurrency_safe=True, ) edit_file_tool = FileTool( edit_file, name="edit_file", description="Precisely edit file content, supporting various editing operations based on line numbers and pattern matching", + concurrency_safe=True, ) find_and_replace_tool = FileTool( find_and_replace, name="find_and_replace", description="Convenient function to find and replace text content", + concurrency_safe=True, ) diff --git a/src/xagent/core/tools/adapters/vibe/image_tool.py b/src/xagent/core/tools/adapters/vibe/image_tool.py index 8dd1235e1..6ba97edd4 100644 --- a/src/xagent/core/tools/adapters/vibe/image_tool.py +++ b/src/xagent/core/tools/adapters/vibe/image_tool.py @@ -80,16 +80,19 @@ def get_tools(self) -> list: self.generate_image, name="generate_image", description=generate_description, + concurrency_safe=True, ), ImageGenerationFunctionTool( self.edit_image, name="edit_image", description=edit_description, + concurrency_safe=True, ), ImageGenerationFunctionTool( self.list_available_models, name="list_image_models", description="List all available image generation models, including model ID, availability status, and detailed description information (Note: model information is already provided in the generate_image tool description)", + read_only=True, ), ] diff --git a/src/xagent/core/tools/adapters/vibe/python_executor.py b/src/xagent/core/tools/adapters/vibe/python_executor.py index 064f5a20f..da82d3800 100644 --- a/src/xagent/core/tools/adapters/vibe/python_executor.py +++ b/src/xagent/core/tools/adapters/vibe/python_executor.py @@ -5,6 +5,7 @@ import asyncio import logging +from pathlib import Path from typing import Any, Dict, Mapping, Optional, Type from pydantic import BaseModel, Field @@ -12,10 +13,8 @@ from ....workspace import TaskWorkspace from ...artifacts import ( build_generated_file_metadata, - changed_generated_artifact_files, - snapshot_generated_artifact_files, ) -from ...core.python_executor import PythonExecutorCore +from ...core.python_executor import _INTERNAL_WRITTEN_FILES_KEY, PythonExecutorCore from .base import AbstractBaseTool, ToolCategory, ToolVisibility from .function import FunctionTool from .sandboxed_tool.sandbox_config import sandbox_config @@ -48,6 +47,8 @@ class PythonExecutorResult(BaseModel): class PythonExecutorTool(AbstractBaseTool): """Framework wrapper for the pure Python executor tool""" + concurrency_safe = True + def __init__(self, workspace: Optional[TaskWorkspace] = None) -> None: self._visibility = ToolVisibility.PUBLIC self._workspace = workspace @@ -82,7 +83,10 @@ def run_json_sync(self, args: Mapping[str, Any]) -> Any: workspace_env = self._get_workspace_env() # Create core executor instance - executor = PythonExecutorCore(working_directory=working_directory) + executor = PythonExecutorCore( + working_directory=working_directory, + environment=workspace_env, + ) # Add workspace variables to the executor's globals if available if workspace_env: @@ -92,23 +96,23 @@ def run_json_sync(self, args: Mapping[str, Any]) -> Any: else: full_code = exec_args.code - # Execute code within auto_register context + # Execute code in an isolated subprocess. The child reports only files + # written by this execution, avoiding concurrent snapshot bleed-through. if self._workspace and working_directory: - files_before = snapshot_generated_artifact_files(working_directory) - with self._workspace.auto_register_files(): - result = executor.execute_code(full_code, exec_args.capture_output) + result = executor.execute_code(full_code, exec_args.capture_output) + written_files = result.pop(_INTERNAL_WRITTEN_FILES_KEY, []) if result.get("success"): - files_after = snapshot_generated_artifact_files(working_directory) + workspace_files = self._existing_workspace_files(written_files) + self._register_workspace_files(workspace_files) result.update( build_generated_file_metadata( workspace=self._workspace, - file_paths=changed_generated_artifact_files( - files_before, files_after - ), + file_paths=workspace_files, ) ) else: result = executor.execute_code(full_code, exec_args.capture_output) + result.pop(_INTERNAL_WRITTEN_FILES_KEY, None) return PythonExecutorResult(**result).model_dump() @@ -134,6 +138,49 @@ def _get_workspace_env(self) -> Optional[Dict[str, str]]: "WORKSPACE_DIR": str(self._workspace.workspace_dir.resolve()), } + def _existing_workspace_files(self, file_paths: Any) -> list[Path]: + """Return existing files from this workspace for this execution only.""" + if not self._workspace or not isinstance(file_paths, list): + return [] + + workspace_root = self._workspace.workspace_dir.resolve() + result: list[Path] = [] + seen: set[str] = set() + + for file_path in file_paths: + try: + resolved = Path(str(file_path)).resolve() + resolved.relative_to(workspace_root) + except (OSError, ValueError): + continue + if not resolved.exists() or not resolved.is_file(): + continue + if any( + part.startswith(".") or part == "__pycache__" + for part in resolved.relative_to(workspace_root).parts + ): + continue + key = str(resolved) + if key in seen: + continue + seen.add(key) + result.append(resolved) + + return result + + def _register_workspace_files(self, file_paths: list[Path]) -> None: + if not self._workspace: + return + for file_path in file_paths: + try: + self._workspace.register_file(str(file_path)) + except Exception as exc: # noqa: BLE001 + logger.warning( + "Failed to register Python executor generated file %s: %s", + file_path, + exc, + ) + @sandbox_config( packages=[ @@ -181,7 +228,9 @@ def execute_python_code(code: str, capture_output: bool = True) -> Dict[str, Any return result return PythonExecutorFunctionTool( - execute_python_code, description=executor.description + execute_python_code, + description=executor.description, + concurrency_safe=True, ) diff --git a/src/xagent/core/tools/adapters/vibe/workspace_file_tool.py b/src/xagent/core/tools/adapters/vibe/workspace_file_tool.py index b900089de..393e0f750 100644 --- a/src/xagent/core/tools/adapters/vibe/workspace_file_tool.py +++ b/src/xagent/core/tools/adapters/vibe/workspace_file_tool.py @@ -174,86 +174,103 @@ def get_tools(self) -> List[FunctionTool]: self.read_file, name="read_file", description="Read file content in workspace. Accepts either file paths (e.g., 'filename.txt') or file_ids (e.g., 'abc-123-def'). Automatically detects input type. For large files, results may be truncated in model context; use start_line/end_line to inspect a specific 1-based inclusive line range instead of repeating the same full-file read.", + read_only=True, ), FileTool( self.write_file, name="write_file", description="Write file content in workspace. Use relative paths (e.g., 'filename.txt'), not absolute paths. Returns a FileRef with file_id, preview_url, download_url, and markdown_link.\n\nImportant: For HTML files, do not guess paths to uploaded files or files from other tasks. First call prepare_html_asset(file_id, html_path, alias) for every external image/CSS/JS asset, then use the returned html_src in the HTML.", + concurrency_safe=True, ), FileTool( self.prepare_html_asset, name="prepare_html_asset", description="Prepare an uploaded or registered file for use inside an HTML artifact. Pass the source file_id, the target HTML output path such as 'index.html' or 'reports/index.html', and an optional alias such as 'logo.png'. The tool copies the asset next to that HTML file under assets_subdir and returns html_src relative to the HTML file. Use html_src in , ,